feat(webui): add assistant reply fork-from-here

nanobot/channels/websocket.py

@@ -45,6 +45,11 @@ from nanobot.webui.http_utils import (
     query_first as _query_first,
 )
 from nanobot.webui.mcp_presets_api import normalize_mcp_preset_mentions
+from nanobot.webui.transcript import (
+    delete_webui_transcript,
+    fork_transcript_before_user_index,
+    write_session_messages_as_transcript,
+)
 from nanobot.webui.transcription_ws import webui_transcription_event
 from nanobot.webui.websocket_logging import websockets_server_logger
 
@@ -668,6 +673,61 @@ class WebSocketChannel(BaseChannel):
             )
             await self._hydrate_after_subscribe(new_id)
             return
+        if t == "fork_chat":
+            source_chat_id = envelope.get("source_chat_id")
+            raw_index = envelope.get("before_user_index")
+            if not _is_valid_chat_id(source_chat_id):
+                await self._send_event(connection, "error", detail="invalid source_chat_id")
+                return
+            if (
+                isinstance(raw_index, bool)
+                or not isinstance(raw_index, int)
+                or raw_index < 0
+            ):
+                await self._send_event(connection, "error", detail="invalid before_user_index")
+                return
+            if self.gateway.session_manager is None:
+                await self._send_event(connection, "error", detail="session_manager_unavailable")
+                return
+
+            new_id = str(uuid.uuid4())
+            source_key = f"websocket:{source_chat_id}"
+            target_key = f"websocket:{new_id}"
+            try:
+                forked = self.gateway.session_manager.fork_session_before_user_index(
+                    source_key,
+                    target_key,
+                    raw_index,
+                )
+                if forked is None:
+                    await self._send_event(connection, "error", detail="invalid fork source or index")
+                    return
+                transcript_ok = fork_transcript_before_user_index(
+                    source_key,
+                    target_key,
+                    raw_index,
+                )
+                if not transcript_ok:
+                    write_session_messages_as_transcript(target_key, forked.messages)
+            except Exception as exc:
+                delete_webui_transcript(target_key)
+                self.gateway.session_manager.delete_session(target_key)
+                self.logger.warning("fork_chat failed: {}", exc)
+                await self._send_event(connection, "error", detail="fork_chat_failed")
+                return
+
+            scope = self._workspaces.scope_for_session_key(target_key)
+            self._attach(connection, new_id)
+            await self._send_event(connection, "attached", chat_id=new_id)
+            await self._send_event(
+                connection,
+                "session_updated",
+                chat_id=new_id,
+                scope="metadata",
+                workspace_scope=scope.payload(),
+            )
+            await self._hydrate_after_subscribe(new_id)
+            return
         if t == "attach":
             cid = envelope.get("chat_id")
             if not _is_valid_chat_id(cid):

nanobot/session/manager.py

@@ -5,6 +5,7 @@ import os
 import re
 import shutil
 from contextlib import suppress
+from copy import deepcopy
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
@@ -30,6 +31,14 @@ _TOOL_CALL_ECHO_RE = re.compile(r'^\s*(?:generate_image|message)\([^)]*\)\s*$')
 _SESSION_PREVIEW_MAX_CHARS = 120
 _SESSION_LIST_PREVIEW_MAX_RECORDS = 200
 _SESSION_LIST_PREVIEW_MAX_CHARS = 1_000_000
+_FORK_VOLATILE_METADATA_KEYS = {
+    "goal_state",
+    "pending_user_turn",
+    "runtime_checkpoint",
+    "thread_goal",
+    "title",
+    "title_user_edited",
+}
 
 
 def _sanitize_assistant_replay_text(content: str) -> str:
@@ -628,6 +637,62 @@ class SessionManager:
             logger.warning("Failed to delete session file {}: {}", path, e)
             return False
 
+    def fork_session_before_user_index(
+        self,
+        source_key: str,
+        target_key: str,
+        before_user_index: int,
+    ) -> Session | None:
+        """Create *target_key* from *source_key* before a global user-message index.
+
+        ``before_user_index`` is zero-based over user messages in the full session:
+        ``0`` means "before the first user message", ``1`` means "before the
+        second user message", and so on. A value equal to the total user-message
+        count copies the full session prefix. The target user message itself is
+        not copied; the WebUI pre-fills it in the composer for editing and resend.
+        """
+        if before_user_index < 0:
+            return None
+        source = self._cache.get(source_key) or self._load(source_key)
+        if source is None:
+            return None
+
+        copied: list[dict[str, Any]] = []
+        user_index = 0
+        found_target = False
+        for message in source.messages:
+            if message.get("role") == "user":
+                if user_index == before_user_index:
+                    found_target = True
+                    break
+                user_index += 1
+            copied.append(deepcopy(message))
+        if user_index == before_user_index:
+            found_target = True
+        if not found_target:
+            return None
+
+        metadata = deepcopy(source.metadata)
+        for key in _FORK_VOLATILE_METADATA_KEYS:
+            metadata.pop(key, None)
+
+        last_consolidated = min(source.last_consolidated, len(copied))
+        if source.last_consolidated > len(copied):
+            metadata.pop("_last_summary", None)
+            last_consolidated = 0
+
+        now = datetime.now()
+        target = Session(
+            key=target_key,
+            messages=copied,
+            created_at=now,
+            updated_at=now,
+            metadata=metadata,
+            last_consolidated=last_consolidated,
+        )
+        self.save(target, fsync=True)
+        return target
+
     def read_session_file(self, key: str) -> dict[str, Any] | None:
         """Load a session from disk without caching; intended for read-only HTTP endpoints.
 

nanobot/webui/transcript.py

@@ -274,6 +274,125 @@ class WebUITranscriptRecorder:
             self._turn_sequences.pop((chat_id, turn_id), None)
 
 
+def _chat_id_from_session_key(session_key: str) -> str | None:
+    if not session_key.startswith("websocket:"):
+        return None
+    chat_id = session_key.split(":", 1)[1].strip()
+    return chat_id or None
+
+
+def _is_user_transcript_row(row: dict[str, Any]) -> bool:
+    return row.get("event") == "user" or row.get("role") == "user"
+
+
+def fork_transcript_before_user_index(
+    source_key: str,
+    target_key: str,
+    before_user_index: int,
+) -> bool:
+    """Copy transcript rows before a zero-based global user-message index.
+
+    ``before_user_index == user_count`` copies the full transcript prefix. WebUI
+    uses that when forking from an assistant reply at the end of a chat.
+    """
+    if before_user_index < 0:
+        return False
+    lines = read_transcript_lines(source_key)
+    if not lines:
+        return False
+
+    target_chat_id = _chat_id_from_session_key(target_key)
+    copied: list[dict[str, Any]] = []
+    user_index = 0
+    found_target = False
+    for row in lines:
+        if _is_user_transcript_row(row):
+            if user_index == before_user_index:
+                found_target = True
+                break
+            user_index += 1
+        dup = json.loads(json.dumps(row, ensure_ascii=False))
+        if target_chat_id is not None:
+            dup["chat_id"] = target_chat_id
+        copied.append(dup)
+    if user_index == before_user_index:
+        found_target = True
+
+    if not found_target:
+        return False
+
+    path = webui_transcript_path(target_key)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp_path = path.with_suffix(".jsonl.tmp")
+    try:
+        with open(tmp_path, "w", encoding="utf-8") as f:
+            for row in copied:
+                raw = json.dumps(row, ensure_ascii=False, separators=(",", ":"))
+                if len(raw.encode("utf-8")) > _MAX_TRANSCRIPT_FILE_BYTES:
+                    raise ValueError("webui transcript line too large")
+                f.write(raw + "\n")
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(tmp_path, path)
+    except BaseException:
+        tmp_path.unlink(missing_ok=True)
+        raise
+    return True
+
+
+def write_session_messages_as_transcript(
+    target_key: str,
+    messages: list[dict[str, Any]],
+) -> None:
+    """Write a minimal WebUI transcript from already-truncated session messages."""
+    target_chat_id = _chat_id_from_session_key(target_key)
+    path = webui_transcript_path(target_key)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp_path = path.with_suffix(".jsonl.tmp")
+    try:
+        with open(tmp_path, "w", encoding="utf-8") as f:
+            for msg in messages:
+                role = msg.get("role")
+                content = msg.get("content")
+                text = content if isinstance(content, str) else ""
+                if role == "user":
+                    row: dict[str, Any] = {
+                        "event": "user",
+                        "chat_id": target_chat_id,
+                        "text": text,
+                    }
+                    media = msg.get("media")
+                    if isinstance(media, list) and media:
+                        row["media_paths"] = [str(p) for p in media if isinstance(p, str) and p]
+                    for key in ("cli_apps", "mcp_presets"):
+                        value = msg.get(key)
+                        if isinstance(value, list) and value:
+                            row[key] = json.loads(json.dumps(value, ensure_ascii=False))
+                elif role == "assistant":
+                    if not text.strip():
+                        continue
+                    row = {
+                        "event": "message",
+                        "chat_id": target_chat_id,
+                        "text": text,
+                    }
+                    media = msg.get("media")
+                    if isinstance(media, list) and media:
+                        row["media"] = [str(p) for p in media if isinstance(p, str) and p]
+                else:
+                    continue
+                raw = json.dumps(row, ensure_ascii=False, separators=(",", ":"))
+                if len(raw.encode("utf-8")) > _MAX_TRANSCRIPT_FILE_BYTES:
+                    raise ValueError("webui transcript line too large")
+                f.write(raw + "\n")
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(tmp_path, path)
+    except BaseException:
+        tmp_path.unlink(missing_ok=True)
+        raise
+
+
 def delete_webui_transcript(session_key: str) -> bool:
     path = webui_transcript_path(session_key)
     if not path.is_file():