From 715513b940386bbe7aba61ddf075dadff5d67cfd Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 10 Nov 2026 09:59:47 +0000 Subject: [PATCH 1/2] Sessions: a fork branches from its stored source until its first completed turn (#673) fork_session stores the parent's native id and the fork turn on the fork row (fork_source_native_id, fork_source_turn_id, migration v072). Every client build of the fork reads them: a session with no usable own native id and a stored source forks from it, whatever its status. The first completed turn in the fork's own conversation removes them. - The fork row never takes the parent's native id as its own: a fork build gives mark_active no id, and the Claude client reports no id for a fork until the CLI sends one. - Both crash-retry builds read the same row, so a retry forks again. - A Stop or an error after the first message keeps the fork's own id for the resume. If that conversation is missing, the fork branches from its source again. - A missing source fails the turn once with a message that says so (ForkSourceMissingError, also from the Codex backend), and removes the source, so the next message starts a new conversation. - A workflow leg or a session nested under another one has no stored source, so it never forks. The sidebar nesting route therefore no longer refuses to nest or un-nest a session that has not run yet. Co-Authored-By: Claude Opus 5.5 --- docs/api.md | 8 + docs/architecture.md | 2 +- nerve/agent/backends/__init__.py | 2 + nerve/agent/backends/base.py | 4 + nerve/agent/backends/claude.py | 6 +- nerve/agent/backends/codex/backend.py | 3 +- nerve/agent/engine.py | 140 +++-- nerve/agent/sessions.py | 16 +- .../db/migrations/v072_session_fork_source.py | 93 ++++ nerve/db/sessions.py | 12 +- nerve/gateway/routes/sessions.py | 24 +- tests/test_codex_appserver.py | 15 +- tests/test_engine.py | 42 -- tests/test_fork_intent.py | 490 ++++++++++++++++++ tests/test_session_parent_patch.py | 22 +- 15 files changed, 749 insertions(+), 130 deletions(-) create mode 100644 nerve/db/migrations/v072_session_fork_source.py create mode 100644 tests/test_fork_intent.py diff --git a/docs/api.md b/docs/api.md index 70862d3a8..2831b4ea8 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1166,6 +1166,14 @@ a numeric message row id as a string). The fork branches the source's pre-populated with the source's messages up to the fork point, so the new chat displays exactly the history the agent remembers. +The fork stores the source's native id and the fork turn when it is created +(`fork_source_native_id`, `fork_source_turn_id`). Each start of the fork's +runtime branches from them until a turn completes in the fork's own +conversation. A stop, a crash, a restart, an archive, or a later change to +the source does not change what the fork branches from. If the source +conversation no longer exists, the fork's next turn fails with a message +that says so, and the message after it starts a new conversation. + Errors: `404` unknown source; `409` not forkable — the source has no native conversation yet (no completed turn), or the anchor message has no native turn mapping (sessions predating per-turn recording can only be forked diff --git a/docs/architecture.md b/docs/architecture.md index 161934071..1b25ea2c1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -314,7 +314,7 @@ Every database open runs the runner: the gateway's start and every CLI command t ### Schema SQLite with WAL mode (schema version 16): -- `sessions` — Session metadata with lifecycle columns (`status`, `sdk_session_id`, `connected_at`, `parent_session_id`, `forked_from_message`, `last_activity_at`, `archived_at`, `message_count`, `total_cost_usd`) and `created_by_actor_id` → `actor_refs`: who caused the session to exist +- `sessions` — Session metadata with lifecycle columns (`status`, `sdk_session_id`, `connected_at`, `parent_session_id`, `forked_from_message`, `fork_source_native_id`, `fork_source_turn_id`, `last_activity_at`, `archived_at`, `message_count`, `total_cost_usd`) and `created_by_actor_id` → `actor_refs`: who caused the session to exist - `messages` — Conversation messages with tool call data and ordered `blocks` JSON column (preserves interleaving of text/thinking/tool_call blocks across page reloads), plus `actor_id` → `actor_refs`: whose input the row records. NULL means attribution was not recorded (legacy history, unidentified external people, and assistant/tool output). Nothing is backfilled — see [Accounts and identity](accounts.md) - `session_events`: append-only lifecycle audit log (created, started, idle, stopped, archived, error, and the changes listed in [Session changes](accounts.md#session-changes)), with `actor_id` and `caused_by_actor_id` → `actor_refs` - `session_deletions`: one row for each deleted session: last title, time, `actor_id` and `caused_by_actor_id`. Written in the transaction of the delete, which removes the session's events diff --git a/nerve/agent/backends/__init__.py b/nerve/agent/backends/__init__.py index f03967393..20f641a4a 100644 --- a/nerve/agent/backends/__init__.py +++ b/nerve/agent/backends/__init__.py @@ -16,6 +16,7 @@ AgentClient, BackendCapabilities, BackendError, + ForkSourceMissingError, SessionSpec, TransportDiedError, TurnInput, @@ -95,6 +96,7 @@ def build_backends(deps: BackendDeps) -> dict[str, AgentBackend]: "BackendCapabilities", "BackendDeps", "BackendError", + "ForkSourceMissingError", "ModelObserved", "NormalizedUsage", "SessionSpec", diff --git a/nerve/agent/backends/base.py b/nerve/agent/backends/base.py index d2ddc69ed..6ca4d36ef 100644 --- a/nerve/agent/backends/base.py +++ b/nerve/agent/backends/base.py @@ -36,6 +36,10 @@ class TransportDiedError(BackendError): """The backend subprocess/stream died mid-operation.""" +class ForkSourceMissingError(BackendError): + """The native conversation that a fork branches from no longer exists.""" + + @dataclass class TurnInput: """One user turn, engine-normalized. diff --git a/nerve/agent/backends/claude.py b/nerve/agent/backends/claude.py index 417ace0a6..8863e0363 100644 --- a/nerve/agent/backends/claude.py +++ b/nerve/agent/backends/claude.py @@ -1284,7 +1284,11 @@ def __init__( # at connect, and a pre-configured copy would set both can_use_tool and # permission_prompt_tool_name (a ValueError there). self._sdk = ClaudeSDKClient(options=options, transport=self._transport) - self._native_session_id: str | None = spec.resume_native_id + # A fork gets its own id from the CLI's first message. Until then it + # has none: the resume target is the parent's conversation. + self._native_session_id: str | None = ( + None if spec.fork else spec.resume_native_id + ) # The resolved model this client was built with (engine reads it # to detect mid-session model switches). self.model: str = options.model or "" diff --git a/nerve/agent/backends/codex/backend.py b/nerve/agent/backends/codex/backend.py index 9be616469..e5d83a722 100644 --- a/nerve/agent/backends/codex/backend.py +++ b/nerve/agent/backends/codex/backend.py @@ -39,6 +39,7 @@ AgentClient, BackendCapabilities, BackendError, + ForkSourceMissingError, SessionSpec, TransportDiedError, TurnInput, @@ -902,7 +903,7 @@ async def connect(self) -> None: if not _is_missing_thread_error(e): raise if spec.fork: - raise BackendError( + raise ForkSourceMissingError( f"Cannot fork missing Codex thread {spec.resume_native_id!r}" ) from e logger.warning( diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index dd25e1195..7cc172f8e 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -40,6 +40,7 @@ AgentBackend, AgentClient, BackendDeps, + ForkSourceMissingError, ModelObserved, SessionSpec, SubagentStarted, @@ -365,6 +366,9 @@ class _TurnState: ordered_blocks: list[dict] = field(default_factory=list) last_usage: dict | None = None sdk_session_id: str | None = None + # True when ``sdk_session_id`` comes from the turn's terminal event, not + # from the live client after a Stop or an error. + native_id_from_result: bool = False native_turn_id: str | None = None result_meta: dict | None = None last_model: str | None = None @@ -548,6 +552,9 @@ def __init__(self, config: NerveConfig, db: Database): self._mcp_catalog_task: asyncio.Task | None = None # The catalog revision that each session's client was built with. self._client_mcp_revisions: dict[str, str | None] = {} + # The fork source of each session whose last client build found one + # on the row (a fork before its first completed turn). + self._fork_sources: dict[str, str] = {} self._mcp_servers_cache = ( # hot-reloadable [] if self._managed_mcp else list(config.mcp_servers) ) @@ -1532,8 +1539,6 @@ def _base_effort_for_source(source: str, effort: str, cron_effort: str) -> str: async def _get_or_create_client( self, session_id: str, source: str, model: str | None, - fork_from: str | None = None, - fork_last_turn_id: str | None = None, effort_override: str | None = None, ) -> AgentClient: """Get an existing persistent client or create a new one. @@ -1547,6 +1552,12 @@ async def _get_or_create_client( A live client is built again when its process is dead, its model changed, or the MCP gateway catalog changed after its build (:meth:`_mcp_catalog_rebuild_due`). + + A new client resumes the session's own native conversation. A fork + that has none yet branches from the fork source on its row + (``fork_source_native_id`` at ``fork_source_turn_id``, written by + ``SessionManager.fork_session``). Every build of the session reads + the same row, so a crash-retry forks again, whatever the status. """ lock = self.sessions.get_lock(session_id) async with lock: @@ -1635,17 +1646,23 @@ async def _get_or_create_client( if session_meta.get("observed_model"): self._observed_models[session_id] = session_meta["observed_model"] - # For forks, use the source session's native ID - if fork_from and not sdk_resume_id: - sdk_resume_id = fork_from + # A fork keeps its source until its first completed turn + # (_persist_turn clears it). The source is never the fork's own + # conversation: resuming it without a fork would write into the + # parent's transcript. + fork_from = (session or {}).get("fork_source_native_id") or None + if fork_from: + self._fork_sources[session_id] = fork_from + if sdk_resume_id == fork_from: + sdk_resume_id = None + else: + self._fork_sources.pop(session_id, None) # Defensive: let the backend verify the resume target is still # materialized (claude: the conversation .jsonl under # ~/.claude/projects; codex: no cheap check — create_client - # falls back and reports resume_dropped instead). Forks are - # exempt: the source session's context lives in the source's - # row, and a fresh fork has nothing to recover to. - if sdk_resume_id and not fork_from: + # falls back and reports resume_dropped instead). + if sdk_resume_id: if not backend.validate_resume_target( sdk_resume_id, session_cwd, ): @@ -1659,7 +1676,20 @@ async def _get_or_create_client( ) sdk_resume_id = None - if sdk_resume_id: + is_fork = bool(fork_from) and not sdk_resume_id + fork_last_turn_id: str | None = None + if is_fork: + await self._check_fork_backend(session or {}, backend.name) + if not backend.validate_resume_target(fork_from, session_cwd): + raise await self._fork_source_missing(session_id, fork_from) + sdk_resume_id = fork_from + fork_last_turn_id = (session or {}).get("fork_source_turn_id") or None + logger.info( + "Forking session %s from native session %s%s", + session_id, fork_from[:12], + f" at turn {fork_last_turn_id[:12]}" if fork_last_turn_id else "", + ) + elif sdk_resume_id: logger.info( "Resuming session %s with native session %s", session_id, sdk_resume_id[:12], @@ -1728,9 +1758,6 @@ async def _get_or_create_client( if session_meta.get("cache_ttl") != cache_ttl: meta_updates["cache_ttl"] = cache_ttl - # Determine if this is a fork - is_fork = fork_from is not None - # Static-prompt mode (agent.static_system_prompt): the session # id and the frozen recall stay OUT of the system prompt so its # bytes are shared across sessions (exact-prefix cache), and @@ -1877,7 +1904,12 @@ async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: ), ) - client = await backend.create_client(spec) + try: + client = await backend.create_client(spec) + except ForkSourceMissingError: + if not is_fork: + raise + raise await self._fork_source_missing(session_id, fork_from) from None if getattr(client, "resume_dropped", False): # The backend had to discard the stale native id (codex # resume-miss recovery) — clear the persisted column AND @@ -1922,7 +1954,9 @@ async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: connected_at = session.get("connected_at") if session and sdk_resume_id else now await self.sessions.mark_active( session_id, - sdk_session_id=sdk_resume_id, + # The resume target of a fork is its source, not its own id: + # the fork's own id comes from the runtime. + sdk_session_id=None if is_fork else sdk_resume_id, connected_at=connected_at, ) await self.db.update_session_fields(session_id, {"model": resolved_model}) @@ -1941,6 +1975,38 @@ async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: ) return client + async def _check_fork_backend(self, session: dict, backend_name: str) -> None: + """Refuse a fork whose parent runs on another backend: a native + conversation cannot move between runtimes. A deleted parent is not + checked (``fork_session`` copied the parent's backend).""" + parent_id = session.get("parent_session_id") + parent = await self.db.get_session(parent_id) if parent_id else None + if parent and parent.get("backend") and parent["backend"] != backend_name: + raise ValueError("Cannot fork a native session across agent backends") + + async def _fork_source_missing( + self, session_id: str, fork_from: str, + ) -> ForkSourceMissingError: + """The error for a fork whose source conversation is gone. Removes + the fork source first, so that the next message starts a new + conversation instead of failing again.""" + logger.warning( + "Session %s: fork source %s is missing; the fork source is removed", + session_id, fork_from[:12], + ) + await self._clear_fork_source(session_id) + return ForkSourceMissingError( + "The conversation that this fork branches from no longer exists, " + "so the agent does not have the copied messages. Your next " + "message starts a new conversation without them." + ) + + async def _clear_fork_source(self, session_id: str) -> None: + await self.db.update_session_fields(session_id, { + "fork_source_native_id": None, "fork_source_turn_id": None, + }) + self._fork_sources.pop(session_id, None) + def _wire_interaction_hooks( self, session_id: str, handler: InteractiveToolHandler, ) -> None: @@ -3338,6 +3404,7 @@ def _record_turn_result(event: TurnCompleted, st: _TurnState) -> None: st.last_usage = event.usage.to_anthropic_shape() if event.native_session_id: st.sdk_session_id = event.native_session_id + st.native_id_from_result = True if event.native_turn_id: st.native_turn_id = event.native_turn_id if event.model and not st.last_model: @@ -3721,6 +3788,10 @@ async def _persist_turn( await self.db.bind_native_thread( backend.name, st.sdk_session_id, session_id, ) + if st.native_id_from_result and session_id in self._fork_sources: + # The runtime finished a turn in the fork's own conversation: + # from now on the fork resumes it. + await self._clear_fork_source(session_id) # Persist usage for context bar on session switch. Backends that # report the serving model's context window (codex, via @@ -4296,41 +4367,10 @@ async def _run_inner( st.ordered_blocks.append({"type": "wakeup"}) try: - # Get or create persistent client for this session - # Check if we need to fork from a parent - fork_from = None - fork_last_turn_id = None - if session: - parent_id = session.get("parent_session_id") - fork_msg = session.get("forked_from_message") - # Workflow legs carry parent_session_id for sidebar nesting only — never fork their fresh prompt from the parent (nor cross-backend). - if (parent_id and session.get("status") == SessionStatus.CREATED.value - and session.get("source") != "workflow"): - parent = await self.db.get_session(parent_id) - if parent and parent.get("sdk_session_id"): - if parent.get("backend") != session.get("backend"): - raise ValueError( - "Cannot fork a native session across agent backends" - ) - fork_from = parent["sdk_session_id"] - if fork_msg: - fork_last_turn_id = await self.db.get_native_turn_at_message( - parent_id, fork_msg, - ) - # Both backends truncate by native turn id now - # (codex: thread/fork lastTurnId; claude: - # --resume-session-at). A message fork without a - # mapping would silently carry the FULL context — - # the opposite of what was asked — so refuse. - if not fork_last_turn_id: - raise ValueError( - "Cannot fork this session at the selected " - "message: no native turn mapping is available." - ) - + # Get or create persistent client for this session (a fork + # branches from its stored source, see _get_or_create_client) client = await self._get_or_create_client( - session_id, source, model, fork_from=fork_from, - fork_last_turn_id=fork_last_turn_id, + session_id, source, model, effort_override=effort_override, ) @@ -4446,7 +4486,7 @@ def _with_session_preamble(base: TurnInput) -> TurnInput: _lf_tags.append(f"channel:{channel}") _lf_metadata = { "parent_session_id": session.get("parent_session_id") if session else None, - "fork_from": fork_from, + "fork_from": session.get("fork_source_native_id") if session else None, } # Live marker so the UI shows the "scheduled wakeup" chip as the # turn streams (the persisted block above covers reload). diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py index 3e41f8de9..5c323cc22 100644 --- a/nerve/agent/sessions.py +++ b/nerve/agent/sessions.py @@ -217,6 +217,8 @@ async def _create_session( actor: Actor | None, cause: CauseArg = CAUSE_FROM_ACTOR, origin: Origin | None = None, + fork_source_native_id: str | None = None, + fork_source_turn_id: str | None = None, ) -> dict: """Create a new session with status=created and log the event.""" if backend is None: @@ -246,6 +248,8 @@ async def _create_session( # this insert made the row. created_event=attribution_for(actor, cause), origin=origin, + fork_source_native_id=fork_source_native_id, + fork_source_turn_id=fork_source_turn_id, ) logger.info( "Created session: %s (parent=%s)", session_id, parent_session_id, @@ -787,6 +791,11 @@ async def fork_session( The source's messages up to that point are copied into the fork so the new chat displays exactly the history the agent remembers. + The fork row stores the source's native id and the fork turn + (``fork_source_native_id``, ``fork_source_turn_id``). The engine + forks from them, whatever happens to the source or to the fork's + status, until the fork's first completed turn. + Raises ``ValueError`` when the source doesn't exist, has no native conversation yet, or ``at_message_id`` can't be mapped to a native turn (pre-mapping history — see ``get_native_turn_at_message``). @@ -807,9 +816,8 @@ async def fork_session( ) # Resolve the fork point up front so a bad anchor fails the API - # call, not the fork's first message. The engine re-resolves the - # same mapping when it builds the client (source history is - # immutable, so both resolutions agree). + # call, not the fork's first message. The same turn bounds the + # copied messages and the native fork. fork_turn_id: str | None = None if at_message_id is not None: fork_turn_id = await self.db.get_native_turn_at_message( @@ -837,6 +845,8 @@ async def fork_session( actor=actor, # A person forks a chat in the web UI, whatever started the parent. origin=Origin(TRIGGER_CHAT, {DETAIL_FORK_OF: source_session_id}), + fork_source_native_id=parent["sdk_session_id"], + fork_source_turn_id=fork_turn_id, ) session["title_source"] = "manual" if title else "fork" await self.db.update_session_fields( diff --git a/nerve/db/migrations/v072_session_fork_source.py b/nerve/db/migrations/v072_session_fork_source.py new file mode 100644 index 000000000..32511bba3 --- /dev/null +++ b/nerve/db/migrations/v072_session_fork_source.py @@ -0,0 +1,93 @@ +"""V72: the native conversation that a fork branches from. + +Adds two columns to ``sessions``: + +- ``fork_source_native_id``: the parent's native session id (Claude session + id, Codex thread id) at fork time. +- ``fork_source_turn_id``: the parent's native turn at which the fork + branches (Claude ``--resume-session-at``, Codex ``lastTurnId``). NULL for a + fork of the whole conversation. + +``fork_session`` writes both. The engine forks from them until the fork's +first completed turn records the fork's own native id, and then clears them. + +Existing forks that have not run a turn get the parent's current native id, +and the turn that ``get_native_turn_at_message`` resolves for +``forked_from_message``. A fork that already has a native id keeps it. The +column check makes a second run do nothing. +""" + +from __future__ import annotations + +import logging + +import aiosqlite + +logger = logging.getLogger(__name__) + +PENDING_FORKS = """ +SELECT f.id, p.id, p.sdk_session_id, f.forked_from_message +FROM sessions f JOIN sessions p ON p.id = f.parent_session_id +WHERE f.status = 'created' + AND COALESCE(f.source, '') != 'workflow' + AND f.sdk_session_id IS NULL + AND p.sdk_session_id IS NOT NULL +""" + +# The two lookups of ``MessageStore.get_native_turn_at_message``. +TURN_AT_OR_AFTER = """ +SELECT native_turn_id FROM messages +WHERE session_id = ? AND id >= ? AND native_turn_id IS NOT NULL +ORDER BY id ASC LIMIT 1 +""" +TURN_AT_OR_BEFORE = """ +SELECT native_turn_id FROM messages +WHERE session_id = ? AND id <= ? AND native_turn_id IS NOT NULL +ORDER BY id DESC LIMIT 1 +""" + + +async def _has_column(db: aiosqlite.Connection, table: str, column: str) -> bool: + async with db.execute(f"PRAGMA table_info({table})") as cursor: + rows = await cursor.fetchall() + return any(row[1] == column for row in rows) + + +async def _turn_at_message( + db: aiosqlite.Connection, session_id: str, message_id: str, +) -> str | None: + try: + numeric_id = int(message_id) + except (TypeError, ValueError): + return None + for sql in (TURN_AT_OR_AFTER, TURN_AT_OR_BEFORE): + async with db.execute(sql, (session_id, numeric_id)) as cursor: + row = await cursor.fetchone() + if row: + return str(row[0]) + return None + + +async def up(db: aiosqlite.Connection) -> None: + if await _has_column(db, "sessions", "fork_source_native_id"): + return + await db.execute("ALTER TABLE sessions ADD COLUMN fork_source_native_id TEXT") + await db.execute("ALTER TABLE sessions ADD COLUMN fork_source_turn_id TEXT") + async with db.execute(PENDING_FORKS) as cursor: + pending = await cursor.fetchall() + filled = 0 + for fork_id, parent_id, parent_native_id, at_message in pending: + turn_id = None + if at_message is not None: + turn_id = await _turn_at_message(db, parent_id, at_message) + if turn_id is None: + # fork_session refuses a message without a turn mapping, so + # only a hand-made row gets here. Do not guess a fork point. + continue + await db.execute( + "UPDATE sessions SET fork_source_native_id = ?, fork_source_turn_id = ? " + "WHERE id = ?", + (parent_native_id, turn_id, fork_id), + ) + filled += 1 + logger.info("v072: sessions fork source columns added, %d pending fork(s) filled", filled) diff --git a/nerve/db/sessions.py b/nerve/db/sessions.py index 5e2b7bfd7..8474848f2 100644 --- a/nerve/db/sessions.py +++ b/nerve/db/sessions.py @@ -66,6 +66,8 @@ async def create_session( cause: CauseArg = CAUSE_FROM_ACTOR, created_event: Attribution | None = None, origin: Origin | None = None, + fork_source_native_id: str | None = None, + fork_source_turn_id: str | None = None, ) -> dict: """Insert a session row, recording who caused it to exist. @@ -100,6 +102,9 @@ async def create_session( (:mod:`nerve.session_origin`), written once like the attribution. Without it, both columns stay NULL, as in a row from before the columns existed, and a reader derives the origin. + + ``fork_source_native_id`` and ``fork_source_turn_id`` are the native + conversation and turn that a fork branches from (migration v072). """ now = datetime.now(timezone.utc).isoformat() created_by = actor.actor_id if actor else None @@ -112,12 +117,14 @@ async def create_session( (id, title, source, metadata, status, parent_session_id, forked_from_message, backend, model, cwd, created_by_actor_id, caused_by_actor_id, trigger, origin_detail, + fork_source_native_id, fork_source_turn_id, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", (session_id, title or session_id, source, json.dumps(metadata or {}), status, parent_session_id, forked_from_message, backend, model, cwd, - created_by, caused_by, trigger, origin_detail, now, now), + created_by, caused_by, trigger, origin_detail, + fork_source_native_id, fork_source_turn_id, now, now), ) inserted = cursor.rowcount == 1 await cursor.close() @@ -518,6 +525,7 @@ def _build_session_fields_update( "archived_at", "title", "message_count", "total_cost_usd", "parent_session_id", "forked_from_message", "last_memorized_at", "starred", "model", "backend", "cwd", "title_source", + "fork_source_native_id", "fork_source_turn_id", } set_clauses: list[str] = [] params: list = [] diff --git a/nerve/gateway/routes/sessions.py b/nerve/gateway/routes/sessions.py index d378a3f5c..daf783e9d 100644 --- a/nerve/gateway/routes/sessions.py +++ b/nerve/gateway/routes/sessions.py @@ -609,30 +609,10 @@ async def update_session(session_id: str, req: dict, actor: Actor = Depends(requ raise HTTPException(status_code=400, detail="A session cannot be its own parent") if not await deps.db.get_session(new_parent): raise HTTPException(status_code=404, detail="Parent session not found") - # Fork-window guard: the engine reads parent_session_id to fork a - # brand-new session's first turn ONLY while status == 'created'. - # Refuse to set a parent in that window so a display drag can never - # turn a not-yet-started session into a fork. Anything already run - # (everything draggable in the sidebar) is unaffected. - if session.get("status") == "created": - raise HTTPException( - status_code=409, - detail="Send a first message before nesting this session", - ) if await _would_create_cycle(deps.db, session_id, new_parent): raise HTTPException(status_code=400, detail="That drop would create a cycle") - elif ( - session.get("status") == "created" - and session.get("parent_session_id") - ): - # Symmetric fork-window guard: clearing the parent of a - # not-yet-started fork would silently discard the fork context - # its first turn is about to branch from. - raise HTTPException( - status_code=409, - detail="This fork hasn't started yet — send its first " - "message before un-nesting it", - ) + # Display only: a fork branches from the source that fork_session + # stored on its row, not from its parent. fields["parent_session_id"] = new_parent if not fields: raise HTTPException(status_code=400, detail="No valid fields to update") diff --git a/tests/test_codex_appserver.py b/tests/test_codex_appserver.py index 58d3c2d3e..295c093a3 100644 --- a/tests/test_codex_appserver.py +++ b/tests/test_codex_appserver.py @@ -21,7 +21,9 @@ import pytest -from nerve.agent.backends import BackendDeps, SessionSpec, TransportDiedError +from nerve.agent.backends import ( + BackendDeps, ForkSourceMissingError, SessionSpec, TransportDiedError, +) from nerve.agent.backends import events as ev from nerve.agent.backends.codex import CodexBackend from nerve.agent.backends.base import TurnInput @@ -708,6 +710,17 @@ async def test_fork_uses_selected_native_turn(tmp_path, monkeypatch): await client.disconnect() +@pytest.mark.asyncio +async def test_fork_of_a_missing_thread_reports_a_missing_fork_source(tmp_path, monkeypatch): + # A fork never falls back to a fresh thread: the engine removes the fork + # source and says so (ForkSourceMissingError). + _mode(monkeypatch, "resume_fail") + cfg = _config(tmp_path) + backend = CodexBackend(_deps(cfg)) + with pytest.raises(ForkSourceMissingError, match="th_gone"): + await backend.create_client(_spec(cfg, resume_native_id="th_gone", fork=True)) + + @pytest.mark.asyncio async def test_interrupt_terminates_receive_turn(tmp_path, monkeypatch): _mode(monkeypatch, "interrupt") diff --git a/tests/test_engine.py b/tests/test_engine.py index 681c41040..4e4300b66 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -898,48 +898,6 @@ async def test_engine_fails_turn_when_eof_follows_content(db, tmp_path): assert json.dumps(asst["blocks"] or []).count("half an answer") == 1 -@pytest.mark.asyncio -async def test_workflow_session_never_forks_from_parent(db, tmp_path): - """A source=='workflow' leg carries parent_session_id for sidebar nesting - only — its first turn must NOT fork the parent's native session (that would - contaminate the fresh orchestration prompt, and cross-backend it raises). - A non-workflow child with identical setup still forks: source is the only - discriminator.""" - engine, _made = await _engine_with_scripted_clients( - db, tmp_path, "seed", [[("text", "ok"), ("result", "sdk-ok")]], - ) - # Parent with a resumable native session — the fork target. - await engine.sessions.get_or_create( - "parent", title="t", source="web", backend="claude", actor=None, - ) - await db.update_session_fields("parent", {"sdk_session_id": "sdk-parent"}) - - forks: dict[str, str | None] = {} - served = engine._get_or_create_client - - async def _cap(sid, source, model, **kw): - forks[sid] = kw.get("fork_from") - return await served(sid, source, model, **kw) - - engine._get_or_create_client = _cap - - # Workflow leg nested under the parent, still in 'created' state. - await engine.sessions.get_or_create( - "workflow:wfr-x", title="t", source="workflow", backend="claude", actor=None, - ) - await db.update_session_fields("workflow:wfr-x", {"parent_session_id": "parent"}) - await engine.run("workflow:wfr-x", "go", source="workflow", channel="web", actor=None) - assert forks["workflow:wfr-x"] is None - - # Control: a non-workflow child with the same setup DOES fork. - await engine.sessions.get_or_create( - "web-child", title="t", source="web", backend="claude", actor=None, - ) - await db.update_session_fields("web-child", {"parent_session_id": "parent"}) - await engine.run("web-child", "go", source="web", channel="web", actor=None) - assert forks["web-child"] == "sdk-parent" - - # --------------------------------------------------------------------------- # ClaudeBackend.validate_resume_target # --------------------------------------------------------------------------- diff --git a/tests/test_fork_intent.py b/tests/test_fork_intent.py new file mode 100644 index 000000000..730a5556c --- /dev/null +++ b/tests/test_fork_intent.py @@ -0,0 +1,490 @@ +"""A fork branches from the source stored on its row until its first +completed turn (issue #673). + +``SessionManager.fork_session`` stores the parent's native id and the fork +turn on the fork row. Every client build of the fork reads them, so the +fork branches from the same point whatever happens before its first +completed turn: a crash-retry, a Stop, a status change, or a parent whose +own conversation pointer is gone. The fork row never takes the parent's +native id as its own. + +The backend here is a fake that records each ``SessionSpec``. Its clients +report native ids as the real ones do: a Claude fork has no id until the +CLI sends one, a Codex fork has its new thread id at once. +""" + +from __future__ import annotations + +import asyncio +import contextlib + +import pytest +from claude_agent_sdk import ClaudeAgentOptions + +from nerve.agent.backends.base import ( + BackendCapabilities, + ForkSourceMissingError, + SessionSpec, + TransportDiedError, +) +from nerve.agent.backends.claude import ClaudeClient +from nerve.agent.backends.events import TextDelta, TurnCompleted +from nerve.agent.engine import AgentEngine +from nerve.config import NerveConfig + +PARENT_NATIVE = "sdk-parent" +FORK_OWN = "sdk-fork-own" + + +class _Client: + def __init__(self, spec: SessionSpec, mode: str, *, codex: bool, n: int): + self.spec = spec + self.mode = mode + self.model = "claude-test" + self.n = n + self.entered = asyncio.Event() + if spec.fork: + # Codex: thread/fork returns the new thread. Claude: no id + # until the CLI's first message. + self._native = f"thread-fork-{n}" if codex else None + else: + self._native = spec.resume_native_id + + @property + def native_session_id(self): + return self._native + + async def connect(self): + pass + + async def start_turn(self, turn): + pass + + def _own_id(self) -> str: + if self.spec.fork: + return self._native or FORK_OWN + return self.spec.resume_native_id or f"sdk-fresh-{self.n}" + + async def receive_turn(self): + self.entered.set() + if self.mode == "crash": + raise TransportDiedError("the runtime ended the turn without a result") + if self.mode == "hang": + await asyncio.Event().wait() + # The CLI's first message brings the fork's own id. + self._native = self._own_id() + yield TextDelta(text=f"reply {self.n}") + if self.mode == "content-hang": + await asyncio.Event().wait() + if self.mode == "content-crash": + raise TransportDiedError("the runtime died after the first message") + yield TurnCompleted( + native_session_id=self._native, native_turn_id=f"turn-{self.n}", + total_cost_usd=0.0, + ) + + async def interrupt(self, **kw): + pass + + def is_alive(self): + return True + + async def disconnect(self, timeout: float = 5.0): + pass + + def try_receive_idle_events(self): + return None + + def buffer_used(self): + return 0 + + +class _Backend: + name = "claude" + + def __init__(self, modes, *, codex: bool = False): + self.capabilities = BackendCapabilities( + supports_idle_stream=False, supports_cache_ttl=False, + ) + self.modes = list(modes) + self.codex = codex + self.specs: list[SessionSpec] = [] + self.clients: list[_Client] = [] + # Native ids whose conversation is gone on this host. + self.missing: set[str] = set() + # Fork sources the runtime itself cannot find (Codex thread/read). + self.refuse_fork_of: set[str] = set() + + def default_model(self, source): + return "claude-test" + + def excluded_tools(self): + return set() + + def validate_resume_target(self, native_id, cwd): + return native_id not in self.missing + + async def create_client(self, spec): + self.specs.append(spec) + if spec.fork and spec.resume_native_id in self.refuse_fork_of: + raise ForkSourceMissingError( + f"Cannot fork missing Codex thread {spec.resume_native_id!r}", + ) + mode = self.modes.pop(0) if self.modes else "ok" + client = _Client(spec, mode, codex=self.codex, n=len(self.clients)) + self.clients.append(client) + return client + + +async def _setup(tmp_path, db, modes, *, codex: bool = False, at_turn: bool = False): + """An engine with a parent that has one completed turn, and a fork of it + made by ``fork_session``.""" + ws = tmp_path / "ws" + ws.mkdir(exist_ok=True) + cfg = NerveConfig.from_dict({"workspace": str(ws), "agent": {"backend": "claude"}}) + engine = AgentEngine(cfg, db) + backend = _Backend(modes, codex=codex) + engine._backends["claude"] = backend + await engine.sessions.get_or_create( + "parent", title="p", source="web", backend="claude", actor=None, + ) + await db.update_session_fields("parent", {"sdk_session_id": PARENT_NATIVE}) + first = await db.add_message("parent", "user", "first question", actor=None) + await db.add_message( + "parent", "assistant", "first answer", native_turn_id="p-turn-1", actor=None, + ) + await db.add_message("parent", "user", "second question", actor=None) + await db.add_message( + "parent", "assistant", "second answer", native_turn_id="p-turn-2", actor=None, + ) + fork = await engine.fork_session( + "parent", str(first) if at_turn else None, None, actor=None, + ) + return engine, backend, fork["id"] + + +async def _drop_client(engine: AgentEngine, session_id: str) -> None: + """End the live client, as an idle reap does: the next turn builds anew.""" + client = engine.sessions.remove_client(session_id, reason="idle_timeout") + if client is not None: + await client.disconnect() + + +async def _run(engine, session_id, text="hello"): + return await engine.run(session_id, text, source="web", channel="web", actor=None) + + +async def _stop_when_entered(engine, backend, session_id, client_index): + task = asyncio.create_task(_run(engine, session_id, "first message")) + for _ in range(300): + if len(backend.clients) > client_index and backend.clients[client_index].entered.is_set(): + break + await asyncio.sleep(0.01) + else: + raise AssertionError("the turn did not start") + await asyncio.sleep(0.05) + task.cancel() + with contextlib.suppress(BaseException): # run() returns after a cancel + await task + + +# --------------------------------------------------------------------------- # +# fork_session and the migration # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_fork_session_stores_the_source_and_the_fork_turn(tmp_path, db): + engine, _backend, fork_id = await _setup(tmp_path, db, [], at_turn=True) + row = await db.get_session(fork_id) + assert row["fork_source_native_id"] == PARENT_NATIVE + # The user message maps to the turn that answers it, which also bounds + # the copied messages. + assert row["fork_source_turn_id"] == "p-turn-1" + assert row["sdk_session_id"] is None + contents = [m["content"] for m in await db.get_messages(fork_id)] + assert contents == ["first question", "first answer"] + + whole = await engine.fork_session("parent", None, None, actor=None) + row = await db.get_session(whole["id"]) + assert row["fork_source_native_id"] == PARENT_NATIVE + assert row["fork_source_turn_id"] is None + + +@pytest.mark.asyncio +async def test_v072_fills_the_source_of_forks_that_have_not_run(db): + from nerve.db.migrations.v072_session_fork_source import up + + await db.create_session("parent", actor=None) + await db.update_session_fields("parent", {"sdk_session_id": PARENT_NATIVE}) + anchor = await db.add_message("parent", "user", "q", actor=None) + await db.add_message("parent", "assistant", "a", native_turn_id="p-turn-1", actor=None) + await db.create_session("whole", parent_session_id="parent", actor=None) + await db.create_session( + "at-turn", parent_session_id="parent", forked_from_message=str(anchor), actor=None, + ) + await db.create_session("ran", parent_session_id="parent", actor=None) + await db.update_session_fields("ran", {"sdk_session_id": "sdk-ran", "status": "idle"}) + await db.create_session( + "leg", source="workflow", parent_session_id="parent", actor=None, + ) + await db.db.execute("ALTER TABLE sessions DROP COLUMN fork_source_native_id") + await db.db.execute("ALTER TABLE sessions DROP COLUMN fork_source_turn_id") + + await up(db.db) + await up(db.db) + + expected = { + "whole": (PARENT_NATIVE, None), + "at-turn": (PARENT_NATIVE, "p-turn-1"), + "ran": (None, None), + "leg": (None, None), + "parent": (None, None), + } + for session_id, (native, turn) in expected.items(): + row = await db.get_session(session_id) + assert (row["fork_source_native_id"], row["fork_source_turn_id"]) == (native, turn), session_id + + +# --------------------------------------------------------------------------- # +# The first turn # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_first_turn_forks_and_then_the_fork_resumes_its_own_conversation(tmp_path, db): + engine, backend, fork_id = await _setup(tmp_path, db, ["ok", "ok"], at_turn=True) + + await _run(engine, fork_id) + first = backend.specs[0] + assert (first.fork, first.resume_native_id, first.fork_last_turn_id) == ( + True, PARENT_NATIVE, "p-turn-1", + ) + row = await db.get_session(fork_id) + assert row["sdk_session_id"] == FORK_OWN + assert row["fork_source_native_id"] is None + assert row["fork_source_turn_id"] is None + + await _drop_client(engine, fork_id) + await _run(engine, fork_id, "second message") + second = backend.specs[1] + assert (second.fork, second.resume_native_id, second.fork_last_turn_id) == ( + False, FORK_OWN, None, + ) + + +@pytest.mark.asyncio +async def test_claude_fork_client_has_no_native_id_before_the_cli_sends_one(): + def _client(fork: bool) -> ClaudeClient: + spec = SessionSpec( + session_id="s", source="web", model="claude-test", effort="high", + system_prompt="x", cwd="/tmp", resume_native_id=PARENT_NATIVE, fork=fork, + ) + return ClaudeClient(spec, ClaudeAgentOptions(model="claude-test")) + + assert _client(fork=True).native_session_id is None + assert _client(fork=False).native_session_id == PARENT_NATIVE + + +@pytest.mark.asyncio +async def test_a_crash_retry_forks_again_and_never_stores_the_parent_id(tmp_path, db): + engine, backend, fork_id = await _setup(tmp_path, db, ["crash", "ok"]) + + await _run(engine, fork_id) + first, retry = backend.specs + assert (first.fork, first.resume_native_id) == (True, PARENT_NATIVE) + assert (retry.fork, retry.resume_native_id) == (True, PARENT_NATIVE) + assert (await db.get_session(fork_id))["sdk_session_id"] == FORK_OWN + + +@pytest.mark.asyncio +async def test_a_first_turn_that_fails_keeps_the_fork_source(tmp_path, db): + engine, backend, fork_id = await _setup(tmp_path, db, ["crash", "crash", "ok"]) + + out = await _run(engine, fork_id) + assert out.startswith("Agent error:") + row = await db.get_session(fork_id) + assert row["sdk_session_id"] is None + assert row["fork_source_native_id"] == PARENT_NATIVE + + await _run(engine, fork_id, "again") + assert (backend.specs[2].fork, backend.specs[2].resume_native_id) == (True, PARENT_NATIVE) + + +@pytest.mark.asyncio +async def test_an_error_after_the_first_message_keeps_the_fork_source(tmp_path, db): + """The id that the error path takes from the live client is kept for the + resume, but only a completed turn ends the fork source.""" + engine, backend, fork_id = await _setup(tmp_path, db, ["content-crash", "ok"]) + + out = await _run(engine, fork_id) + assert out.startswith("Agent error:") + row = await db.get_session(fork_id) + assert row["sdk_session_id"] == FORK_OWN + assert row["fork_source_native_id"] == PARENT_NATIVE + + backend.missing.add(FORK_OWN) + await _run(engine, fork_id, "again") + assert (backend.specs[1].fork, backend.specs[1].resume_native_id) == (True, PARENT_NATIVE) + + +@pytest.mark.asyncio +async def test_a_stop_before_the_first_message_keeps_the_fork_source(tmp_path, db): + engine, backend, fork_id = await _setup(tmp_path, db, ["hang", "ok"]) + + # The client build does not write the parent's id into the fork row. + task = asyncio.create_task(_run(engine, fork_id, "first message")) + for _ in range(300): + if backend.clients and backend.clients[0].entered.is_set(): + break + await asyncio.sleep(0.01) + assert (await db.get_session(fork_id))["sdk_session_id"] is None + task.cancel() + with contextlib.suppress(BaseException): + await task + + row = await db.get_session(fork_id) + assert row["sdk_session_id"] is None + assert row["fork_source_native_id"] == PARENT_NATIVE + + await _run(engine, fork_id, "second message") + assert (backend.specs[1].fork, backend.specs[1].resume_native_id) == (True, PARENT_NATIVE) + + +@pytest.mark.asyncio +async def test_a_stop_after_the_first_message_resumes_the_forks_own_conversation(tmp_path, db): + engine, backend, fork_id = await _setup(tmp_path, db, ["content-hang", "ok"]) + + await _stop_when_entered(engine, backend, fork_id, 0) + row = await db.get_session(fork_id) + assert row["sdk_session_id"] == FORK_OWN + # Kept until a turn completes in the fork's own conversation. + assert row["fork_source_native_id"] == PARENT_NATIVE + + await _run(engine, fork_id, "go on") + assert (backend.specs[1].fork, backend.specs[1].resume_native_id) == (False, FORK_OWN) + assert (await db.get_session(fork_id))["fork_source_native_id"] is None + + +@pytest.mark.asyncio +async def test_a_missing_own_conversation_forks_again_from_the_source(tmp_path, db): + engine, backend, fork_id = await _setup(tmp_path, db, ["content-hang", "ok"]) + await _stop_when_entered(engine, backend, fork_id, 0) + assert (await db.get_session(fork_id))["sdk_session_id"] == FORK_OWN + + # The runtime died before it wrote the fork's transcript. + backend.missing.add(FORK_OWN) + await _run(engine, fork_id, "go on") + assert (backend.specs[1].fork, backend.specs[1].resume_native_id) == (True, PARENT_NATIVE) + + +@pytest.mark.asyncio +async def test_a_codex_fork_stopped_before_content_resumes_its_new_thread(tmp_path, db): + engine, backend, fork_id = await _setup(tmp_path, db, ["hang", "ok"], codex=True) + + await _stop_when_entered(engine, backend, fork_id, 0) + assert (await db.get_session(fork_id))["sdk_session_id"] == "thread-fork-0" + + await _run(engine, fork_id, "go on") + assert (backend.specs[1].fork, backend.specs[1].resume_native_id) == (False, "thread-fork-0") + + +@pytest.mark.asyncio +async def test_a_row_that_holds_the_source_id_still_forks(tmp_path, db): + """Defense against a runtime that reports the parent's id for a fork.""" + engine, backend, fork_id = await _setup(tmp_path, db, ["ok"]) + await db.update_session_fields(fork_id, {"sdk_session_id": PARENT_NATIVE}) + + await _run(engine, fork_id) + assert (backend.specs[0].fork, backend.specs[0].resume_native_id) == (True, PARENT_NATIVE) + + +# --------------------------------------------------------------------------- # +# Status changes before the first turn # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_an_unarchived_fork_still_forks_from_its_source(tmp_path, db): + engine, backend, fork_id = await _setup(tmp_path, db, ["ok"]) + + # Archiving the parent archives the nested fork and clears the parent's + # native id. The person brings the fork back. + result = await engine.sessions.archive_session_cascade("parent") + assert set(result["archived"]) == {"parent", fork_id} + assert (await db.get_session("parent"))["sdk_session_id"] is None + assert await engine.sessions.unarchive_session(fork_id) + assert (await db.get_session(fork_id))["status"] == "idle" + + await _run(engine, fork_id) + assert (backend.specs[0].fork, backend.specs[0].resume_native_id) == (True, PARENT_NATIVE) + + +@pytest.mark.asyncio +async def test_resume_session_does_not_fork_a_fork_again(tmp_path, db): + engine, backend, fork_id = await _setup(tmp_path, db, ["ok", "ok"]) + await _run(engine, fork_id) + await _drop_client(engine, fork_id) + await engine.sessions.mark_idle(fork_id) + + resumed = await engine.resume_session(fork_id) + assert resumed["status"] == "created" + await _run(engine, fork_id, "second message") + assert (backend.specs[1].fork, backend.specs[1].resume_native_id) == (False, FORK_OWN) + + +@pytest.mark.asyncio +async def test_only_fork_session_makes_a_fork(tmp_path, db): + """A workflow leg, or a session nested under another one, has a parent + but no fork source: its first turn starts a new conversation.""" + engine, backend, _fork_id = await _setup(tmp_path, db, ["ok", "ok"]) + await engine.sessions.get_or_create( + "workflow:wfr-x", title="t", source="workflow", backend="claude", actor=None, + ) + await db.update_session_fields("workflow:wfr-x", {"parent_session_id": "parent"}) + await engine.sessions.get_or_create( + "nested", title="t", source="web", backend="claude", actor=None, + ) + await db.update_session_fields("nested", {"parent_session_id": "parent"}) + + await engine.run("workflow:wfr-x", "go", source="workflow", channel="web", actor=None) + await _run(engine, "nested", "go") + for spec in backend.specs: + assert (spec.fork, spec.resume_native_id) == (False, None) + + +# --------------------------------------------------------------------------- # +# A missing source # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +@pytest.mark.parametrize("found_by", ["engine", "runtime"]) +async def test_a_missing_source_fails_once_and_says_so(tmp_path, db, found_by): + engine, backend, fork_id = await _setup(tmp_path, db, ["ok"]) + if found_by == "engine": + backend.missing.add(PARENT_NATIVE) # Claude: no transcript file + else: + backend.refuse_fork_of.add(PARENT_NATIVE) # Codex: thread/read fails + + out = await _run(engine, fork_id) + assert out.startswith("Agent error:") + assert "no longer exists" in out + assert "next message starts a new conversation" in out + row = await db.get_session(fork_id) + assert row["fork_source_native_id"] is None + assert row["fork_source_turn_id"] is None + + await _run(engine, fork_id, "again") + last = backend.specs[-1] + assert (last.fork, last.resume_native_id) == (False, None) + + +@pytest.mark.asyncio +async def test_a_fork_across_backends_is_refused(tmp_path, db): + engine, backend, fork_id = await _setup(tmp_path, db, ["ok"]) + await db.update_session_fields("parent", {"backend": "codex"}) + + out = await _run(engine, fork_id) + assert "Cannot fork a native session across agent backends" in out + assert backend.specs == [] diff --git a/tests/test_session_parent_patch.py b/tests/test_session_parent_patch.py index b9db68a0c..0904920f1 100644 --- a/tests/test_session_parent_patch.py +++ b/tests/test_session_parent_patch.py @@ -2,9 +2,8 @@ Covers the write path added for the session-hierarchy feature in ``gateway/routes/sessions.py``: setting a parent, clearing it (null → -top-level), and the guards — self-parent, unknown parent, cycle, and the -fork-window (a session still in ``created`` status can't be nested, so a -display drag never turns a not-yet-started session into a fork). +top-level), and the guards — self-parent, unknown parent, cycle. A drag +never changes what a fork branches from. Harness mirrors TestSidebarListRoutes in test_sidebar_pagination_api.py: a minimal FastAPI app with the real router, auth disabled via an empty jwt_secret. @@ -101,13 +100,22 @@ async def test_cycle_rejected(self, setup): r = setup.client.patch("/api/sessions/a", json={"parent_session_id": "b"}) assert r.status_code == 400 - async def test_created_session_cannot_be_nested(self, setup): - # A brand-new session still in 'created' is fork-eligible; refuse to - # re-parent it so a drag can never turn it into a fork. + async def test_nesting_never_changes_a_fork(self, setup): + # A session forks only from the source that fork_session stored, so a + # drag can nest or un-nest a session that has not run yet. await self._mk(setup, "parent") await self._mk(setup, "fresh", started=False) # stays 'created' r = setup.client.patch("/api/sessions/fresh", json={"parent_session_id": "parent"}) - assert r.status_code == 409 + assert r.status_code == 200 + assert (await setup.db.get_session("fresh"))["fork_source_native_id"] is None + + await setup.db.update_session_fields("parent", {"sdk_session_id": "sdk-parent"}) + fork = await setup.sm.fork_session("parent", actor=None) + r = setup.client.patch(f"/api/sessions/{fork['id']}", json={"parent_session_id": None}) + assert r.status_code == 200 + row = await setup.db.get_session(fork["id"]) + assert row["parent_session_id"] is None + assert row["fork_source_native_id"] == "sdk-parent" async def test_deep_chain_allowed(self, setup): # Arbitrary depth is fine: a → b → c. From d3cf0772bd3e85e037ec1d2dcf280f5c2c9526ca Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 10 Nov 2026 10:18:10 +0000 Subject: [PATCH 2/2] Sessions: close the review findings on fork intent - A whole fork branches at the source's last completed turn (new MessageStore.get_last_native_turn), so a turn that the source completes later is neither copied nor in the fork's context. History without a turn mapping is forked whole. - A Codex fork whose own thread is gone when it connects (resume_dropped) branches from its source again, not from a fresh thread. - The cross-backend check is gone: it read the display parent, which a drag can change. fork_session copies the source's backend to the fork, and the backend stays fixed for the session. - resume_session moves only a stopped, idle or errored session to created (SessionManager.transition takes only_from), so it no longer replaces the status of a running session. - v072 adds each column only when it is missing and writes only rows with no source, so an interrupted run can run again. It finds forks by their fork- id, forked_from_message or fork_of origin, whatever their status. It takes the source from the fork's first started event, the parent's current id, or the parent's last id before the fork. A fork bound to its source's id loses that id. A fork that had its own conversation gets no source. Co-Authored-By: Claude Opus 5.5 --- docs/api.md | 14 +- nerve/agent/engine.py | 53 +++--- nerve/agent/sessions.py | 15 +- nerve/db/messages.py | 12 ++ .../db/migrations/v072_session_fork_source.py | 125 +++++++++++--- tests/test_fork_intent.py | 157 +++++++++++++++--- 6 files changed, 298 insertions(+), 78 deletions(-) diff --git a/docs/api.md b/docs/api.md index 2831b4ea8..66dbf6c02 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1167,9 +1167,12 @@ pre-populated with the source's messages up to the fork point, so the new chat displays exactly the history the agent remembers. The fork stores the source's native id and the fork turn when it is created -(`fork_source_native_id`, `fork_source_turn_id`). Each start of the fork's -runtime branches from them until a turn completes in the fork's own -conversation. A stop, a crash, a restart, an archive, or a later change to +(`fork_source_native_id`, `fork_source_turn_id`). A fork of the whole chat +branches at the source's last completed turn, so a turn that the source +completes later is neither copied nor known to the fork's agent (history +from before per-turn recording is forked whole). Each start of the fork's +runtime branches from the stored source until a turn completes in the fork's +own conversation. A stop, a crash, a restart, an archive, or a later change to the source does not change what the fork branches from. If the source conversation no longer exists, the fork's next turn fails with a message that says so, and the message after it starts a new conversation. @@ -1185,7 +1188,10 @@ Response: { "id": "fork-a1b2c3d4", "title": "My Fork", "source": "web", "status" ``` #### `POST /api/sessions/{id}/resume` -Resume a stopped or idle session (must have a stored `sdk_session_id`). +Resume a stopped or idle session (must have a stored `sdk_session_id`). A +`stopped`, `idle` or `error` session goes to `created`; any other status (for +example `active`, while a turn runs) stays as it is. The next message resumes +the stored native conversation. ```json Response: { "id": "a1b2c3d4", "status": "created", "sdk_session_id": "..." } diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index 7cc172f8e..5e66b2d52 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -22,7 +22,7 @@ import time from collections import OrderedDict from collections.abc import Iterator -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace as dc_replace from datetime import datetime, timedelta, timezone from typing import Any @@ -1679,7 +1679,6 @@ async def _get_or_create_client( is_fork = bool(fork_from) and not sdk_resume_id fork_last_turn_id: str | None = None if is_fork: - await self._check_fork_backend(session or {}, backend.name) if not backend.validate_resume_target(fork_from, session_cwd): raise await self._fork_source_missing(session_id, fork_from) sdk_resume_id = fork_from @@ -1904,12 +1903,15 @@ async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: ), ) - try: - client = await backend.create_client(spec) - except ForkSourceMissingError: - if not is_fork: - raise - raise await self._fork_source_missing(session_id, fork_from) from None + async def _create(spec: SessionSpec) -> AgentClient: + try: + return await backend.create_client(spec) + except ForkSourceMissingError: + if not spec.fork: + raise + raise await self._fork_source_missing(session_id, fork_from) from None + + client = await _create(spec) if getattr(client, "resume_dropped", False): # The backend had to discard the stale native id (codex # resume-miss recovery) — clear the persisted column AND @@ -1921,6 +1923,17 @@ async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: sdk_resume_id = None # The transcript is fresh after all — it needs the preamble. deliver_preamble = static_prompt + if fork_from: + # A fork whose own conversation is gone branches from + # its source again, not from nothing. + await self._safe_disconnect(client) + is_fork = True + fork_last_turn_id = (session or {}).get("fork_source_turn_id") or None + spec = dc_replace( + spec, resume_native_id=fork_from, fork=True, + fork_last_turn_id=fork_last_turn_id, + ) + client = await _create(spec) if deliver_preamble: self._pending_preambles[session_id] = build_session_preamble( session_id, source, recalled_memories or None, @@ -1975,15 +1988,6 @@ async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: ) return client - async def _check_fork_backend(self, session: dict, backend_name: str) -> None: - """Refuse a fork whose parent runs on another backend: a native - conversation cannot move between runtimes. A deleted parent is not - checked (``fork_session`` copied the parent's backend).""" - parent_id = session.get("parent_session_id") - parent = await self.db.get_session(parent_id) if parent_id else None - if parent and parent.get("backend") and parent["backend"] != backend_name: - raise ValueError("Cannot fork a native session across agent backends") - async def _fork_source_missing( self, session_id: str, fork_from: str, ) -> ForkSourceMissingError: @@ -2921,20 +2925,25 @@ async def resume_session( ) -> dict: """Resume a stopped/idle session. - Records a ``created`` status event with reason ``resume`` and - ``attribution``, only when the status changes: a session that is - resumed already stays as it is, so of two resumes at the same time - only one is recorded. + A ``stopped``, ``idle`` or ``error`` session goes to ``created``, with + a ``created`` status event with reason ``resume`` and + ``attribution``. Any other status stays as it is and records + nothing: a session that runs keeps ``active``, and of two resumes at + the same time only one is recorded. The next message resumes the + stored native conversation. """ info = await self.sessions.get_resume_info(session_id) if not info or not info.get("sdk_session_id"): raise ValueError( f"Session {session_id} cannot be resumed (no SDK session)", ) - # Mark as created so the next message will reconnect the client await self.sessions.transition( session_id, SessionStatus.CREATED, {"reason": "resume"}, attribution=attribution, + only_from=( + SessionStatus.STOPPED.value, SessionStatus.IDLE.value, + SessionStatus.ERROR.value, + ), ) session = await self.db.get_session(session_id) return session diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py index 5c323cc22..c3a42bced 100644 --- a/nerve/agent/sessions.py +++ b/nerve/agent/sessions.py @@ -264,6 +264,7 @@ async def transition( self, session_id: str, new_status: SessionStatus, details: dict | None = None, *, attribution: Attribution = UNATTRIBUTED, + only_from: tuple[str, ...] | None = None, ) -> bool: """Atomically transition a session's status and log the event. @@ -271,14 +272,15 @@ async def transition( resume). A transition that records the state of the runtime keeps it unattributed. A session that has ``new_status`` already stays as it is and records nothing, so of two callers that make the same - transition at the same time only one is recorded. Returns ``True`` - when the status changed. + transition at the same time only one is recorded. With + ``only_from``, a session in another status also stays as it is. + Returns ``True`` when the status changed. """ async with self._transition_lock: return await self.db.change_session_status( session_id, new_status.value, event_type=new_status.value, details=details, - attribution=attribution, + attribution=attribution, only_from=only_from, ) # The ``mark_*`` methods record the state of the runtime. They write @@ -817,7 +819,10 @@ async def fork_session( # Resolve the fork point up front so a bad anchor fails the API # call, not the fork's first message. The same turn bounds the - # copied messages and the native fork. + # copied messages and the native fork. A fork of the whole chat + # branches at the source's last completed turn, so turns that the + # source completes later are in neither. With no turn mapping (history + # from before per-turn recording) it takes the whole conversation. fork_turn_id: str | None = None if at_message_id is not None: fork_turn_id = await self.db.get_native_turn_at_message( @@ -829,6 +834,8 @@ async def fork_session( "mapping (recorded per turn since this feature shipped). " "Fork the whole chat instead." ) + else: + fork_turn_id = await self.db.get_last_native_turn(source_session_id) fork_id = f"fork-{str(uuid.uuid4())[:8]}" fork_title = title or f"Fork of {parent.get('title', source_session_id)}" diff --git a/nerve/db/messages.py b/nerve/db/messages.py index a8cc7f84e..cc71e0f7e 100644 --- a/nerve/db/messages.py +++ b/nerve/db/messages.py @@ -121,6 +121,18 @@ async def get_native_turn_at_message( row = await cursor.fetchone() return str(row[0]) if row else None + async def get_last_native_turn(self, session_id: str) -> str | None: + """Return the native turn of the session's last completed turn, or + ``None`` when no message has one.""" + async with self.db.execute( + """SELECT native_turn_id FROM messages + WHERE session_id = ? AND native_turn_id IS NOT NULL + ORDER BY id DESC LIMIT 1""", + (session_id,), + ) as cursor: + row = await cursor.fetchone() + return str(row[0]) if row else None + async def copy_messages_to_session( self, source_session_id: str, diff --git a/nerve/db/migrations/v072_session_fork_source.py b/nerve/db/migrations/v072_session_fork_source.py index 32511bba3..ba421641d 100644 --- a/nerve/db/migrations/v072_session_fork_source.py +++ b/nerve/db/migrations/v072_session_fork_source.py @@ -11,10 +11,26 @@ ``fork_session`` writes both. The engine forks from them until the fork's first completed turn records the fork's own native id, and then clears them. -Existing forks that have not run a turn get the parent's current native id, -and the turn that ``get_native_turn_at_message`` resolves for -``forked_from_message``. A fork that already has a native id keeps it. The -column check makes a second run do nothing. +Existing forks get a source when they never had a native conversation of +their own. A fork is a row with a ``fork-`` id, a ``forked_from_message``, or +a ``fork_of`` origin, whatever its status. The source is, in this order: + +1. The native id of the fork's first ``started`` event, which its first + client build recorded: the parent's id that it branched from (NULL when + that build did not fork). +2. The parent's current native id. +3. The parent's last native id in a ``started`` event before the fork was + created (a parent whose id an archive cleared). + +A fork that recorded any other native id had a conversation of its own and +gets no source. The fork turn is the turn that ``get_native_turn_at_message`` +resolves for ``forked_from_message``, and NULL for a whole fork. A fork whose +own native id is its source (a first turn that did not complete bound it to +the parent's conversation) loses that id, so that it branches again instead +of writing into the parent's conversation. + +Each column is added only when it is missing, and the backfill writes only +rows with no source yet, so a run that was interrupted can run again. """ from __future__ import annotations @@ -25,15 +41,42 @@ logger = logging.getLogger(__name__) -PENDING_FORKS = """ -SELECT f.id, p.id, p.sdk_session_id, f.forked_from_message -FROM sessions f JOIN sessions p ON p.id = f.parent_session_id -WHERE f.status = 'created' +FORKS = """ +SELECT f.id, f.parent_session_id, f.forked_from_message, f.sdk_session_id, + p.sdk_session_id +FROM sessions f LEFT JOIN sessions p ON p.id = f.parent_session_id +WHERE f.parent_session_id IS NOT NULL + AND f.fork_source_native_id IS NULL AND COALESCE(f.source, '') != 'workflow' - AND f.sdk_session_id IS NULL - AND p.sdk_session_id IS NOT NULL + AND (f.id LIKE 'fork-%' + OR f.forked_from_message IS NOT NULL + OR CASE WHEN json_valid(f.origin_detail) + THEN json_extract(f.origin_detail, '$.fork_of') END IS NOT NULL) """ +_STARTED_ID = ( + "CASE WHEN json_valid(details) " + "THEN json_extract(details, '$.sdk_session_id') END" +) +FIRST_STARTED_ID = f""" +SELECT {_STARTED_ID} FROM session_events +WHERE session_id = ? AND event_type = 'started' +ORDER BY id ASC LIMIT 1 +""" +OTHER_STARTED_ID = f""" +SELECT 1 FROM session_events +WHERE session_id = ? AND event_type = 'started' AND {_STARTED_ID} IS NOT NULL + AND {_STARTED_ID} != ? +LIMIT 1 +""" +LAST_STARTED_ID_BEFORE = f""" +SELECT {_STARTED_ID} FROM session_events +WHERE session_id = ? AND event_type = 'started' AND {_STARTED_ID} IS NOT NULL + AND id < ? +ORDER BY id DESC LIMIT 1 +""" +FIRST_EVENT_ID = "SELECT MIN(id) FROM session_events WHERE session_id = ?" + # The two lookups of ``MessageStore.get_native_turn_at_message``. TURN_AT_OR_AFTER = """ SELECT native_turn_id FROM messages @@ -53,6 +96,12 @@ async def _has_column(db: aiosqlite.Connection, table: str, column: str) -> bool return any(row[1] == column for row in rows) +async def _one(db: aiosqlite.Connection, sql: str, params: tuple): + async with db.execute(sql, params) as cursor: + row = await cursor.fetchone() + return row[0] if row else None + + async def _turn_at_message( db: aiosqlite.Connection, session_id: str, message_id: str, ) -> str | None: @@ -61,22 +110,42 @@ async def _turn_at_message( except (TypeError, ValueError): return None for sql in (TURN_AT_OR_AFTER, TURN_AT_OR_BEFORE): - async with db.execute(sql, (session_id, numeric_id)) as cursor: - row = await cursor.fetchone() - if row: - return str(row[0]) + turn = await _one(db, sql, (session_id, numeric_id)) + if turn is not None: + return str(turn) return None +async def _source_of( + db: aiosqlite.Connection, fork_id: str, parent_id: str, + parent_native_id: str | None, +) -> str | None: + recorded = await _one(db, FIRST_STARTED_ID, (fork_id,)) + if recorded: + return str(recorded) + if parent_native_id: + return parent_native_id + created = await _one(db, FIRST_EVENT_ID, (fork_id,)) + before = await _one( + db, LAST_STARTED_ID_BEFORE, + (parent_id, created if created is not None else 2**62), + ) + return str(before) if before else None + + async def up(db: aiosqlite.Connection) -> None: - if await _has_column(db, "sessions", "fork_source_native_id"): - return - await db.execute("ALTER TABLE sessions ADD COLUMN fork_source_native_id TEXT") - await db.execute("ALTER TABLE sessions ADD COLUMN fork_source_turn_id TEXT") - async with db.execute(PENDING_FORKS) as cursor: - pending = await cursor.fetchall() - filled = 0 - for fork_id, parent_id, parent_native_id, at_message in pending: + for column in ("fork_source_native_id", "fork_source_turn_id"): + if not await _has_column(db, "sessions", column): + await db.execute(f"ALTER TABLE sessions ADD COLUMN {column} TEXT") + async with db.execute(FORKS) as cursor: + forks = await cursor.fetchall() + filled = detached = 0 + for fork_id, parent_id, at_message, own_id, parent_native_id in forks: + source = await _source_of(db, fork_id, parent_id, parent_native_id) + if source is None or (own_id and own_id != source): + continue + if await _one(db, OTHER_STARTED_ID, (fork_id, source)): + continue turn_id = None if at_message is not None: turn_id = await _turn_at_message(db, parent_id, at_message) @@ -85,9 +154,13 @@ async def up(db: aiosqlite.Connection) -> None: # only a hand-made row gets here. Do not guess a fork point. continue await db.execute( - "UPDATE sessions SET fork_source_native_id = ?, fork_source_turn_id = ? " - "WHERE id = ?", - (parent_native_id, turn_id, fork_id), + "UPDATE sessions SET fork_source_native_id = ?, fork_source_turn_id = ?," + " sdk_session_id = NULL WHERE id = ?", + (source, turn_id, fork_id), ) filled += 1 - logger.info("v072: sessions fork source columns added, %d pending fork(s) filled", filled) + detached += bool(own_id) + logger.info( + "v072: sessions fork source columns present, %d fork(s) filled " + "(%d bound to their source's conversation)", filled, detached, + ) diff --git a/tests/test_fork_intent.py b/tests/test_fork_intent.py index 730a5556c..ef8d64c89 100644 --- a/tests/test_fork_intent.py +++ b/tests/test_fork_intent.py @@ -43,6 +43,7 @@ def __init__(self, spec: SessionSpec, mode: str, *, codex: bool, n: int): self.model = "claude-test" self.n = n self.entered = asyncio.Event() + self.resume_dropped = False if spec.fork: # Codex: thread/fork returns the new thread. Claude: no id # until the CLI's first message. @@ -114,6 +115,9 @@ def __init__(self, modes, *, codex: bool = False): self.missing: set[str] = set() # Fork sources the runtime itself cannot find (Codex thread/read). self.refuse_fork_of: set[str] = set() + # Resume targets that the runtime finds missing only when it + # connects (Codex): it starts a fresh thread and says so. + self.drop_resume_of: set[str] = set() def default_model(self, source): return "claude-test" @@ -131,7 +135,12 @@ async def create_client(self, spec): f"Cannot fork missing Codex thread {spec.resume_native_id!r}", ) mode = self.modes.pop(0) if self.modes else "ok" - client = _Client(spec, mode, codex=self.codex, n=len(self.clients)) + if not spec.fork and spec.resume_native_id in self.drop_resume_of: + spec = SessionSpec(**{**vars(spec), "resume_native_id": None}) + client = _Client(spec, mode, codex=self.codex, n=len(self.clients)) + client.resume_dropped = True + else: + client = _Client(spec, mode, codex=self.codex, n=len(self.clients)) self.clients.append(client) return client @@ -205,45 +214,117 @@ async def test_fork_session_stores_the_source_and_the_fork_turn(tmp_path, db): contents = [m["content"] for m in await db.get_messages(fork_id)] assert contents == ["first question", "first answer"] + # A whole fork branches at the parent's last completed turn: a turn that + # the parent completes later is neither copied nor in the native fork. whole = await engine.fork_session("parent", None, None, actor=None) row = await db.get_session(whole["id"]) assert row["fork_source_native_id"] == PARENT_NATIVE - assert row["fork_source_turn_id"] is None + assert row["fork_source_turn_id"] == "p-turn-2" + await db.add_message("parent", "user", "later question", actor=None) + await db.add_message( + "parent", "assistant", "later answer", native_turn_id="p-turn-3", actor=None, + ) + await _run(engine, whole["id"]) + spec = engine._backends["claude"].specs[-1] + assert (spec.fork, spec.resume_native_id, spec.fork_last_turn_id) == ( + True, PARENT_NATIVE, "p-turn-2", + ) + contents = [m["content"] for m in await db.get_messages(whole["id"])] + assert "later question" not in contents and "later answer" not in contents @pytest.mark.asyncio -async def test_v072_fills_the_source_of_forks_that_have_not_run(db): +async def test_a_whole_fork_of_history_without_turn_mapping_takes_everything(tmp_path, db): + engine, _backend, _fork_id = await _setup(tmp_path, db, []) + await engine.sessions.get_or_create( + "old", title="o", source="web", backend="claude", actor=None, + ) + await db.update_session_fields("old", {"sdk_session_id": "sdk-old"}) + await db.add_message("old", "user", "q", actor=None) + await db.add_message("old", "assistant", "a", actor=None) + fork = await engine.fork_session("old", None, None, actor=None) + row = await db.get_session(fork["id"]) + assert (row["fork_source_native_id"], row["fork_source_turn_id"]) == ("sdk-old", None) + assert fork["message_count"] == 2 + + +async def _started(db, session_id: str, native_id: str | None) -> None: + await db.log_session_event(session_id, "started", {"sdk_session_id": native_id}) + + +@pytest.mark.asyncio +async def test_v072_gives_a_source_to_forks_that_never_had_their_own_conversation(db): from nerve.db.migrations.v072_session_fork_source import up await db.create_session("parent", actor=None) await db.update_session_fields("parent", {"sdk_session_id": PARENT_NATIVE}) + await _started(db, "parent", PARENT_NATIVE) anchor = await db.add_message("parent", "user", "q", actor=None) await db.add_message("parent", "assistant", "a", native_turn_id="p-turn-1", actor=None) - await db.create_session("whole", parent_session_id="parent", actor=None) - await db.create_session( - "at-turn", parent_session_id="parent", forked_from_message=str(anchor), actor=None, + + async def fork(fork_id: str, parent: str = "parent", **kw) -> None: + await db.create_session(fork_id, parent_session_id=parent, actor=None, **kw) + await db.log_session_event(fork_id, "created", {"parent": parent}) + + # Not run yet, whole and at a message. + await fork("fork-whole") + await fork("fork-at", forked_from_message=str(anchor)) + # Archived and unarchived before its first turn: idle, no own id. + await fork("fork-unarchived") + await db.update_session_fields("fork-unarchived", {"status": "idle"}) + # A first turn that did not complete bound it to the parent's id. + await fork("fork-bound") + await _started(db, "fork-bound", PARENT_NATIVE) + await db.update_session_fields( + "fork-bound", {"sdk_session_id": PARENT_NATIVE, "status": "idle"}, ) - await db.create_session("ran", parent_session_id="parent", actor=None) - await db.update_session_fields("ran", {"sdk_session_id": "sdk-ran", "status": "idle"}) + # Its parent was archived (native id cleared) after the fork. + await db.create_session("archived-parent", actor=None) + await _started(db, "archived-parent", "sdk-old-parent") + await fork("fork-of-archived", parent="archived-parent") + await _started(db, "archived-parent", "sdk-after-the-fork") + # Completed a turn in its own conversation, with or without the id now. + await fork("fork-ran") + await _started(db, "fork-ran", PARENT_NATIVE) + await _started(db, "fork-ran", "sdk-ran") + await db.update_session_fields("fork-ran", {"sdk_session_id": "sdk-ran", "status": "idle"}) + await fork("fork-ran-then-archived") + await _started(db, "fork-ran-then-archived", PARENT_NATIVE) + await _started(db, "fork-ran-then-archived", "sdk-ran-2") + # Its first build did not fork (status had changed): a fresh conversation. + await fork("fork-fresh") + await _started(db, "fork-fresh", None) + await _started(db, "fork-fresh", "sdk-fresh") + # Not forks: a nested session and a workflow leg. + await db.create_session("nested", parent_session_id="parent", actor=None) await db.create_session( - "leg", source="workflow", parent_session_id="parent", actor=None, + "workflow:wfr-x", source="workflow", parent_session_id="parent", actor=None, ) + # An interrupted run: only the first column was added. await db.db.execute("ALTER TABLE sessions DROP COLUMN fork_source_native_id") await db.db.execute("ALTER TABLE sessions DROP COLUMN fork_source_turn_id") + await db.db.execute("ALTER TABLE sessions ADD COLUMN fork_source_native_id TEXT") await up(db.db) await up(db.db) expected = { - "whole": (PARENT_NATIVE, None), - "at-turn": (PARENT_NATIVE, "p-turn-1"), - "ran": (None, None), - "leg": (None, None), - "parent": (None, None), + "fork-whole": (PARENT_NATIVE, None, None), + "fork-at": (PARENT_NATIVE, "p-turn-1", None), + "fork-unarchived": (PARENT_NATIVE, None, None), + "fork-bound": (PARENT_NATIVE, None, None), + "fork-of-archived": ("sdk-old-parent", None, None), + "fork-ran": (None, None, "sdk-ran"), + "fork-ran-then-archived": (None, None, None), + "fork-fresh": (None, None, None), + "nested": (None, None, None), + "workflow:wfr-x": (None, None, None), + "parent": (None, None, PARENT_NATIVE), } - for session_id, (native, turn) in expected.items(): + for session_id, want in expected.items(): row = await db.get_session(session_id) - assert (row["fork_source_native_id"], row["fork_source_turn_id"]) == (native, turn), session_id + got = (row["fork_source_native_id"], row["fork_source_turn_id"], row["sdk_session_id"]) + assert got == want, session_id # --------------------------------------------------------------------------- # @@ -389,6 +470,24 @@ async def test_a_codex_fork_stopped_before_content_resumes_its_new_thread(tmp_pa assert (backend.specs[1].fork, backend.specs[1].resume_native_id) == (False, "thread-fork-0") +@pytest.mark.asyncio +async def test_a_codex_fork_whose_new_thread_is_gone_forks_again(tmp_path, db): + """Codex finds a missing thread only when it connects, and then starts a + fresh one: for a fork, the engine branches from the source instead.""" + engine, backend, fork_id = await _setup(tmp_path, db, ["hang", "ok", "ok"], codex=True) + await _stop_when_entered(engine, backend, fork_id, 0) + assert (await db.get_session(fork_id))["sdk_session_id"] == "thread-fork-0" + + backend.drop_resume_of.add("thread-fork-0") + await _run(engine, fork_id, "go on") + dropped, refork = backend.specs[1:] + assert (dropped.fork, dropped.resume_native_id) == (False, "thread-fork-0") + assert (refork.fork, refork.resume_native_id) == (True, PARENT_NATIVE) + row = await db.get_session(fork_id) + assert row["sdk_session_id"] == "thread-fork-2" + assert row["fork_source_native_id"] is None + + @pytest.mark.asyncio async def test_a_row_that_holds_the_source_id_still_forks(tmp_path, db): """Defense against a runtime that reports the parent's id for a fork.""" @@ -425,7 +524,7 @@ async def test_resume_session_does_not_fork_a_fork_again(tmp_path, db): engine, backend, fork_id = await _setup(tmp_path, db, ["ok", "ok"]) await _run(engine, fork_id) await _drop_client(engine, fork_id) - await engine.sessions.mark_idle(fork_id) + await engine.sessions.mark_stopped(fork_id) resumed = await engine.resume_session(fork_id) assert resumed["status"] == "created" @@ -433,6 +532,16 @@ async def test_resume_session_does_not_fork_a_fork_again(tmp_path, db): assert (backend.specs[1].fork, backend.specs[1].resume_native_id) == (False, FORK_OWN) +@pytest.mark.asyncio +async def test_resume_session_leaves_a_running_session_active(tmp_path, db): + engine, _backend, fork_id = await _setup(tmp_path, db, ["ok"]) + await _run(engine, fork_id) + assert (await db.get_session(fork_id))["status"] == "active" + + resumed = await engine.resume_session(fork_id) + assert resumed["status"] == "active" + + @pytest.mark.asyncio async def test_only_fork_session_makes_a_fork(tmp_path, db): """A workflow leg, or a session nested under another one, has a parent @@ -481,10 +590,14 @@ async def test_a_missing_source_fails_once_and_says_so(tmp_path, db, found_by): @pytest.mark.asyncio -async def test_a_fork_across_backends_is_refused(tmp_path, db): +async def test_a_fork_moved_under_another_session_still_forks(tmp_path, db): + """The parent link is for the sidebar: moving a fork that has not run + under a session of another backend does not change what it forks.""" engine, backend, fork_id = await _setup(tmp_path, db, ["ok"]) - await db.update_session_fields("parent", {"backend": "codex"}) + await engine.sessions.get_or_create( + "other", title="o", source="web", backend="codex", actor=None, + ) + await db.update_session_fields(fork_id, {"parent_session_id": "other"}) - out = await _run(engine, fork_id) - assert "Cannot fork a native session across agent backends" in out - assert backend.specs == [] + await _run(engine, fork_id) + assert (backend.specs[0].fork, backend.specs[0].resume_native_id) == (True, PARENT_NATIVE)