diff --git a/docs/api.md b/docs/api.md index 70862d3a8..66dbf6c02 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1166,6 +1166,17 @@ 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`). 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. + 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 @@ -1177,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/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..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 @@ -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,19 @@ 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: + 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 +1757,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 +1903,15 @@ async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: ), ) - client = await backend.create_client(spec) + 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 @@ -1889,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, @@ -1922,7 +1967,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 +1988,29 @@ async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: ) return client + 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: @@ -2855,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 @@ -3338,6 +3413,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 +3797,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 +4376,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 +4495,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..c3a42bced 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, @@ -260,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. @@ -267,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 @@ -787,6 +793,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 +818,11 @@ 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. 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( @@ -821,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)}" @@ -837,6 +852,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/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 new file mode 100644 index 000000000..ba421641d --- /dev/null +++ b/nerve/db/migrations/v072_session_fork_source.py @@ -0,0 +1,166 @@ +"""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 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 + +import logging + +import aiosqlite + +logger = logging.getLogger(__name__) + +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.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 +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 _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: + try: + numeric_id = int(message_id) + except (TypeError, ValueError): + return None + for sql in (TURN_AT_OR_AFTER, TURN_AT_OR_BEFORE): + 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: + 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) + 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 = ?," + " sdk_session_id = NULL WHERE id = ?", + (source, turn_id, fork_id), + ) + filled += 1 + 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/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..ef8d64c89 --- /dev/null +++ b/tests/test_fork_intent.py @@ -0,0 +1,603 @@ +"""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() + self.resume_dropped = False + 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() + # 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" + + 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" + 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 + + +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"] + + # 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"] == "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_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) + + 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"}, + ) + # 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( + "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 = { + "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, want in expected.items(): + row = await db.get_session(session_id) + got = (row["fork_source_native_id"], row["fork_source_turn_id"], row["sdk_session_id"]) + assert got == want, 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_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.""" + 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_stopped(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_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 + 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_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 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"}) + + await _run(engine, fork_id) + assert (backend.specs[0].fork, backend.specs[0].resume_native_id) == (True, PARENT_NATIVE) 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.