diff --git a/coding-agent/.opencode/prompts/tasks/checkpoint-active-run-timeline.md b/coding-agent/.opencode/prompts/tasks/checkpoint-active-run-timeline.md new file mode 100644 index 000000000..b901fbc51 --- /dev/null +++ b/coding-agent/.opencode/prompts/tasks/checkpoint-active-run-timeline.md @@ -0,0 +1,69 @@ +Goal: +Restore a checkpoint without replaying or resuming runtime runs that occurred +after that checkpoint, while retaining those runs and events for audit. + +Scope: +- Add durable, backwards-compatible run supersession fields to JSONL, SQLite, + and PostgreSQL runtime records. +- Mark post-checkpoint active runs inside SQLite and PostgreSQL durable restore + transactions. +- Add an explicit active-run query contract and use it for session history, + latest-run/resume state, result fallback, and ACP session load. +- Exclude superseded runs from executor claim paths and persist the reconciled + current turn in session metadata. +- Add focused persistence, restore, query, HTTP, ACP, and WebUI regressions. +- Record the recurring projection-reconciliation failure in `postmortem/`. + +Out of scope: +- Deleting runtime runs, events, snapshots, or interactions. +- Atomic runtime-projection reconciliation for legacy JSONL checkpoint restore. +- Production deployment, o6n mutation, GitOps updates, or live smoke tests. +- Changing developer-console audit or direct run-id/event endpoints to hide + superseded records. +- Redesigning checkpoint snapshot payloads or restore authorization. + +Context: +- ADRs: + - `docs/adr/0005-checkpoint-restore-uses-truncate-rollback.md` + - `docs/adr/0010-synchronize-checkpoint-restore-with-active-turns.md` + - `docs/adr/0068-local-sqlite-transactional-durable-fencing.md` + - `docs/adr/0075-checkpoint-restore-active-run-timeline.md` +- Relevant files: + - `src/coding_agent/stores/runtime_store.py` + - `src/coding_agent/stores/durable_local.py` + - `src/coding_agent/stores/durable_pg.py` + - `src/coding_agent/runs/query.py` + - `src/coding_agent/server/http_server.py` + - `src/coding_agent/acp/server.py` + - `webui/app/src/App.test.tsx` +- Existing behavior: + - Checkpoint capture runs under exclusive runtime maintenance admission. + - Durable restore already reconciles tape, session, topics, and checkpoints in + one SQLite or PostgreSQL transaction. + - `list_agent_runs()` is an audit/control query and must remain unfiltered. + - A test command that selects zero tests is a failure, not verification. +- Postmortem: + - `PM-0026` matches `runtime_store.py`; run its focused adapter/lifecycle + release checks and verify no new blank exception-to-run-error path. + +Target tests: +- `uv run pytest tests/coding_agent/test_sqlite_runtime_store.py tests/coding_agent/test_pg_runtime_store.py tests/coding_agent/test_sqlite_local_durable_fencing.py tests/coding_agent/test_pg_durable_fencing.py tests/coding_agent/test_runtime_query_service.py -v` +- `uv run pytest tests/ui/test_http_server.py tests/acp/test_server.py -k "checkpoint or runtime_run or session_load or result" -v` +- `uv run pytest tests/ui/test_session_manager_runtime.py -k "restore_durable_state" -v` +- `uv run pytest tests/coding_agent/test_adapter_types.py tests/coding_agent/test_pipeline_adapter.py tests/coding_agent/test_run_lifecycle.py -v` +- `pnpm --dir webui/app --ignore-workspace run test -- App.test.tsx` +- `rg -n "str\\(exc\\)" src/coding_agent/stores/runtime_store.py src/coding_agent/stores/durable_local.py src/coding_agent/stores/durable_pg.py` + +Loop policy: +- Engineer writes focused failing tests first, confirms the intended failures, + implements the smallest correct change, and runs the target tests. +- Reviewer reviews only the resulting diff and affected tests. +- Reviewer reports only P1/P2 findings. +- Engineer fixes only accepted P1/P2 findings and reruns the same target tests. +- Verifier reruns the exact target tests and reports pass/fail only. + +Stop conditions: +- At most one review/fix/retest cycle. +- Escalate architectural redirection or scope expansion to the human. +- Ignore non-blocking optimization suggestions. +- Do not push, open a PR, deploy, or mutate o6n without explicit authorization. diff --git a/coding-agent/docs/adr/0075-checkpoint-restore-active-run-timeline.md b/coding-agent/docs/adr/0075-checkpoint-restore-active-run-timeline.md new file mode 100644 index 000000000..7f5806da7 --- /dev/null +++ b/coding-agent/docs/adr/0075-checkpoint-restore-active-run-timeline.md @@ -0,0 +1,96 @@ +# ADR-0075: Reconcile the active run timeline on checkpoint restore + +**Status**: Proposed +**Date**: 2026-08-05 + +## Context + +Checkpoint restore replaces the stable tape and session configuration with the +selected snapshot, but runtime run records and their events remain durable. The +session history, resume metadata, result fallback, and ACP session load currently +list every durable run. A run created after the restored checkpoint can therefore +reappear as an apparently current assistant response even though its tape entries +were rolled back. + +Runtime records also serve audit, worker recovery, and developer diagnostics. +Deleting those records would repair the visible timeline at the cost of losing +evidence. Filtering by the checkpoint timestamp forever is also incorrect because +new runs created after a successful restore belong to the new active timeline. + +Checkpoint capture is admitted through the session's exclusive runtime +maintenance boundary. A run cannot legitimately overlap capture, so the +checkpoint creation time is the durable boundary available to existing snapshots. + +## Decision + +Preserve every runtime run and event for audit, and add nullable run fields that +record when and by which checkpoint a run was superseded. During durable SQLite +and PostgreSQL restore, mark previously active runs for the same session whose +`started_at` is later than the checkpoint's `created_at`. This mutation is part of +the same fenced transaction that restores tape, session state, topics, and the +checkpoint set. Existing supersession markers are not overwritten. + +Before the restore transaction persists the session payload, recompute +`current_turn_id` from the latest active run at or before the checkpoint boundary. +Persist that optional value in session metadata so restart hydration remains +aligned with the restored timeline. + +Keep `list_agent_runs()` and direct run/event lookup as full audit queries. Add an +explicit active-timeline query in the run query service that excludes superseded +runs. Session history, latest-run selection, resume metadata, result fallback, and +ACP session load use the active query; worker recovery, developer diagnostics, and +direct run/event endpoints keep using the audit query. + +New runs created after restore are active because supersession is persisted only +for rows present when restore executes. Existing databases are upgraded +idempotently so the nullable fields do not require an offline migration. + +JSONL records round trip the optional fields for compatibility, but JSONL restore +does not gain projection atomicity in this change. JSONL is a legacy/debug format; +the production contract applies to the authoritative SQLite and PostgreSQL +durable stores. + +## Alternatives Rejected + +- Delete post-checkpoint runs and events: fixes replay but destroys audit and + incident evidence. +- Filter permanently by checkpoint creation time: also hides legitimate runs + created after restore. +- Filter only in WebUI: leaves ACP load, resume, session summary, and result + fallback inconsistent. +- Change `list_agent_runs()` to hide superseded rows everywhere: breaks audit, + startup recovery, worker state, and developer-console consumers. + +Executor claim paths are control-plane consumers rather than audit queries. They +exclude superseded requested or expired rows so rolled-back work cannot execute +after restore. + +## Acceptance Criteria + +- [ ] SQLite and PostgreSQL runtime schemas upgrade existing databases and round + trip supersession fields. +- [ ] SQLite and PostgreSQL durable restore atomically mark only active runs that + started after the restored checkpoint. +- [ ] Active run queries, latest-run selection, session summary, result fallback, + and ACP load ignore superseded runs. +- [ ] Full run listing and direct run/event/display-event lookup retain + superseded records for audit. +- [ ] A run created after restore remains visible on the active timeline. +- [ ] Superseded requested or expired runs cannot be claimed by an executor. +- [ ] Restart hydration retains the reconciled `current_turn_id`. +- [ ] `uv run pytest tests/coding_agent/test_sqlite_runtime_store.py tests/coding_agent/test_pg_runtime_store.py tests/coding_agent/test_sqlite_local_durable_fencing.py tests/coding_agent/test_pg_durable_fencing.py tests/coding_agent/test_runtime_query_service.py -v` +- [ ] `uv run pytest tests/ui/test_http_server.py tests/acp/test_server.py -k "checkpoint or runtime_run or session_load or result" -v` +- [ ] `pnpm --dir webui/app --ignore-workspace run test -- App.test.tsx` + +## References + +- `docs/adr/0005-checkpoint-restore-uses-truncate-rollback.md` +- `docs/adr/0010-synchronize-checkpoint-restore-with-active-turns.md` +- `docs/adr/0068-local-sqlite-transactional-durable-fencing.md` +- `src/coding_agent/stores/runtime_store.py` +- `src/coding_agent/stores/durable_local.py` +- `src/coding_agent/stores/durable_pg.py` +- `src/coding_agent/runs/query.py` +- `src/coding_agent/server/http_server.py` +- `src/coding_agent/acp/server.py` +- `webui/app/src/lib/api.ts` diff --git a/coding-agent/postmortem/index.yaml b/coding-agent/postmortem/index.yaml index c09e713bc..9956e172f 100644 --- a/coding-agent/postmortem/index.yaml +++ b/coding-agent/postmortem/index.yaml @@ -669,3 +669,31 @@ patterns: - message - run - persistence +- id: PM-0027 + title: Reconcile runtime projections on checkpoint restore + path: patterns/PM-0027-reconcile-runtime-projections-on-checkpoint-restore.md + severity: high + confidence: high + subsystems: + - checkpoint + - runtime + - ui + related_files: + - src/coding_agent/stores/durable_local.py + - src/coding_agent/stores/durable_pg.py + - src/coding_agent/stores/runtime_store.py + - src/coding_agent/runs/query.py + - src/coding_agent/server/http_server.py + - src/coding_agent/acp/server.py + - webui/app/src/App.tsx + - webui/app/src/lib/api.ts + related_commits: [] + keywords: + - checkpoint + - restore + - runtime + - projection + - superseded + - timeline + - replay + - audit diff --git a/coding-agent/postmortem/patterns/PM-0027-reconcile-runtime-projections-on-checkpoint-restore.md b/coding-agent/postmortem/patterns/PM-0027-reconcile-runtime-projections-on-checkpoint-restore.md new file mode 100644 index 000000000..975786e1b --- /dev/null +++ b/coding-agent/postmortem/patterns/PM-0027-reconcile-runtime-projections-on-checkpoint-restore.md @@ -0,0 +1,49 @@ +--- +id: PM-0027 +title: Reconcile runtime projections on checkpoint restore +status: active +severity: high +confidence: high +subsystems: + - checkpoint + - runtime + - ui +related_commits: [] +related_files: + - src/coding_agent/stores/durable_local.py + - src/coding_agent/stores/durable_pg.py + - src/coding_agent/stores/runtime_store.py + - src/coding_agent/runs/query.py + - src/coding_agent/server/http_server.py + - src/coding_agent/acp/server.py + - webui/app/src/App.tsx + - webui/app/src/lib/api.ts +release_checks: + - Verify SQLite and PostgreSQL restore reconcile the active run timeline atomically. + - Verify session history, resume, result fallback, ACP, and WebUI use active runs only. + - Verify direct run and event lookup still exposes superseded records for audit. + - Verify a run created after restore remains active. + - Verify executor claim paths reject superseded requested and expired runs. + - Verify the reconciled current turn survives session reload. +--- + +# Summary + +Restoring tape entries without reconciling durable runtime projections can make +post-checkpoint assistant replies reappear after a successful restore. The tape +is the conversation source of truth, but run records and display events remain +valid audit evidence and must not be deleted. + +Persist an explicit supersession marker during the restore transaction. Product +timeline consumers must query active runs, while audit and control consumers keep +the full run set. Do not infer the active timeline only in a frontend, and do not +apply a permanent timestamp filter that would hide new runs after restore. + +# Release Review Checklist + +- Run focused SQLite and PostgreSQL durable restore tests. +- Run runtime-query, HTTP result/session-history, ACP load, and WebUI restore tests. +- Confirm superseded run events remain addressable by direct run id. +- Confirm post-restore runs are not hidden. +- Confirm superseded queued runs cannot be claimed and executed. +- Confirm restored session persistence reloads the latest active turn id. diff --git a/coding-agent/src/coding_agent/acp/server.py b/coding-agent/src/coding_agent/acp/server.py index 829ca2da1..81b69c7f4 100644 --- a/coding-agent/src/coding_agent/acp/server.py +++ b/coding-agent/src/coding_agent/acp/server.py @@ -96,6 +96,8 @@ async def replay_display_events( async def list_runtime_runs(self, session_id: str) -> list[Any]: ... + async def list_active_runtime_runs(self, session_id: str) -> list[Any]: ... + async def list_sessions_async(self) -> list[str]: ... async def close_session(self, session_id: str) -> None: ... @@ -151,9 +153,7 @@ def __init__( self._model_name = model_name self._base_url = base_url self._max_steps = max_steps - self._modes = ( - modes if modes is not None else default_acp_modes() - ) + self._modes = modes if modes is not None else default_acp_modes() self._active_prompt_sessions: set[str] = set() self._cancelled_prompt_sessions: set[str] = set() @@ -389,7 +389,7 @@ async def _session_load(self, params: JSONObject) -> None: session_id, additional_directories, ) - runs = await self._session_manager.list_runtime_runs(session_id) + runs = await self._session_manager.list_active_runtime_runs(session_id) for run in _sort_runtime_runs(runs): run_id = getattr(run, "run_id", None) if not isinstance(run_id, str) or not run_id: @@ -872,7 +872,8 @@ def _session_mode_state(modes: Iterable[AcpMode] | None = None) -> JSONObject: { "id": m.id, "name": m.name, - "description": m.description or f"Provider: {m.provider or 'any'}, Model: {m.model or 'any'}", + "description": m.description + or f"Provider: {m.provider or 'any'}, Model: {m.model or 'any'}", } for m in (modes or default_acp_modes()) ] diff --git a/coding-agent/src/coding_agent/cli/local_runtime.py b/coding-agent/src/coding_agent/cli/local_runtime.py index 0788940e0..25ec773e8 100644 --- a/coding-agent/src/coding_agent/cli/local_runtime.py +++ b/coding-agent/src/coding_agent/cli/local_runtime.py @@ -81,6 +81,8 @@ async def update_session_mcp_servers( async def list_runtime_runs(self, session_id: str) -> Any: ... + async def list_active_runtime_runs(self, session_id: str) -> Any: ... + async def list_checkpoints(self, session_id: str) -> Any: ... async def session_resume_metadata(self, session_id: str) -> dict[str, object]: ... @@ -233,6 +235,9 @@ async def update_session_additional_directories( async def list_runtime_runs(self, session_id: str) -> Any: return await self._delegate.list_runtime_runs(session_id) + async def list_active_runtime_runs(self, session_id: str) -> Any: + return await self._delegate.list_active_runtime_runs(session_id) + async def list_checkpoints(self, session_id: str) -> Any: return await self._delegate.list_checkpoints(session_id) diff --git a/coding-agent/src/coding_agent/runs/query.py b/coding-agent/src/coding_agent/runs/query.py index 3f45e279b..ab5737805 100644 --- a/coding-agent/src/coding_agent/runs/query.py +++ b/coding-agent/src/coding_agent/runs/query.py @@ -93,8 +93,18 @@ async def load_runtime_run(self, run_id: str) -> AgentRunRecord: async def list_runtime_runs(self, session_id: str) -> list[AgentRunRecord]: return await self._require_store().list_agent_runs(session_id) + async def list_active_runtime_runs( + self, + session_id: str, + ) -> list[AgentRunRecord]: + return [ + run + for run in await self.list_runtime_runs(session_id) + if run.superseded_at is None + ] + async def latest_runtime_run(self, session_id: str) -> AgentRunRecord | None: - runs = await self.list_runtime_runs(session_id) + runs = await self.list_active_runtime_runs(session_id) if not runs: return None return max(runs, key=lambda run: (run.started_at, run.run_id)) @@ -194,7 +204,7 @@ async def session_resume_metadata( ) interrupted_runs = [ run - for run in await self.list_runtime_runs(session.id) + for run in await self.list_active_runtime_runs(session.id) if run.status == "interrupted" ] if interrupted_runs: diff --git a/coding-agent/src/coding_agent/server/http_server.py b/coding-agent/src/coding_agent/server/http_server.py index 4b4ccd702..6d6dc4c06 100644 --- a/coding-agent/src/coding_agent/server/http_server.py +++ b/coding-agent/src/coding_agent/server/http_server.py @@ -4195,7 +4195,7 @@ async def list_session_runtime_runs( del request _ = await _get_visible_session(session_id, auth_context) try: - records = await session_manager.list_runtime_runs(session_id) + records = await session_manager.list_active_runtime_runs(session_id) except RuntimeError as exc: raise HTTPException( status_code=404, @@ -6491,7 +6491,7 @@ async def _session_result_runtime_run_id(session: Session) -> str | None: if session.current_turn_id is not None: return session.current_turn_id try: - runs = await session_manager.list_runtime_runs(session.id) + runs = await session_manager.list_active_runtime_runs(session.id) except RuntimeError: return None if not runs: diff --git a/coding-agent/src/coding_agent/server/session_manager.py b/coding-agent/src/coding_agent/server/session_manager.py index 914447eaa..f66557b62 100644 --- a/coding-agent/src/coding_agent/server/session_manager.py +++ b/coding-agent/src/coding_agent/server/session_manager.py @@ -402,6 +402,7 @@ class SessionRecord: additional_directories: list[str] tape_id: str | None last_failure_details: str | None + current_turn_id: str | None = None def to_store_data(self) -> dict[str, Any]: return { @@ -422,6 +423,7 @@ def to_store_data(self) -> dict[str, Any]: "additional_directories": list(self.additional_directories), "tape_id": self.tape_id, "last_failure_details": self.last_failure_details, + "turn_id": self.current_turn_id, } @classmethod @@ -459,6 +461,9 @@ def from_store_data(cls, data: dict[str, Any]) -> SessionRecord: last_failure_details_raw, str ): raise TypeError("session metadata has invalid last_failure_details") + current_turn_id_raw = data.get("turn_id") + if current_turn_id_raw is not None and not isinstance(current_turn_id_raw, str): + raise TypeError("session metadata has invalid turn_id") mcp_servers_raw = data.get("mcp_servers", {}) if not isinstance(mcp_servers_raw, dict): raise TypeError("session metadata has invalid mcp_servers") @@ -505,6 +510,7 @@ def from_store_data(cls, data: dict[str, Any]) -> SessionRecord: additional_directories=additional_directories, tape_id=tape_id_raw, last_failure_details=last_failure_details_raw, + current_turn_id=current_turn_id_raw, ) def to_session(self) -> Session: @@ -525,6 +531,7 @@ def to_session(self) -> Session: additional_directories=list(self.additional_directories), tape_id=self.tape_id, last_failure_details=self.last_failure_details, + current_turn_id=self.current_turn_id, ) @@ -762,6 +769,7 @@ def to_record(self) -> SessionRecord: additional_directories=list(self.additional_directories), tape_id=self.tape_id, last_failure_details=self.last_failure_details, + current_turn_id=self.current_turn_id, ) @classmethod @@ -1112,12 +1120,7 @@ async def runtime_session_is_recoverable(session_id: str) -> bool: self._runtime_checkpoint_restore_orchestration = ( RuntimeCheckpointRestoreOrchestrationService( admission=self._runtime_maintenance_admission, - restore=lambda session, checkpoint_id: ( - self._runtime_checkpoint_restore_service.restore( - session, - checkpoint_id, - ) - ), + restore=self._restore_checkpoint, ) ) self._runtime_checkpoint_query_service = RuntimeCheckpointQueryService( @@ -1682,6 +1685,14 @@ async def list_runtime_runs(self, session_id: str) -> list[AgentRunRecord]: session_id ) + async def list_active_runtime_runs( + self, + session_id: str, + ) -> list[AgentRunRecord]: + return await self._runtime_control_services.queries().list_active_runtime_runs( + session_id + ) + async def session_resume_metadata(self, session_id: str) -> dict[str, Any]: session = await self.get_session_async(session_id) return await self._runtime_control_services.queries().session_resume_metadata( @@ -2946,6 +2957,16 @@ async def _request_approval(req: ApprovalRequest) -> ApprovalResponse: async def _restore_checkpoint(self, session: Session, checkpoint_id: str) -> None: await self._runtime_checkpoint_restore_service.restore(session, checkpoint_id) + latest_active_run = ( + None + if self._runtime_store is None + else await self._runtime_control_services.queries().latest_runtime_run( + session.id + ) + ) + session.current_turn_id = ( + None if latest_active_run is None else latest_active_run.run_id + ) async def _restore_checkpoint_durable_state( self, @@ -2953,6 +2974,24 @@ async def _restore_checkpoint_durable_state( snapshot: Any, ) -> None: typed_session = cast(Session, session) + active_runs = ( + [] + if self._runtime_store is None + else await self._runtime_control_services.queries().list_active_runtime_runs( + typed_session.id + ) + ) + runs_at_checkpoint = [ + run for run in active_runs if run.started_at <= snapshot.meta.created_at + ] + latest_run = ( + None + if not runs_at_checkpoint + else max(runs_at_checkpoint, key=lambda run: (run.started_at, run.run_id)) + ) + typed_session.current_turn_id = ( + None if latest_run is None else latest_run.run_id + ) if self._local_durable_store is None and self._pg_durable_store is None: if typed_session.tape_id is None: raise ValueError("session has no stable tape id") diff --git a/coding-agent/src/coding_agent/stores/durable_local.py b/coding-agent/src/coding_agent/stores/durable_local.py index 0e6bfe9ed..d182065e3 100644 --- a/coding-agent/src/coding_agent/stores/durable_local.py +++ b/coding-agent/src/coding_agent/stores/durable_local.py @@ -96,7 +96,7 @@ def __init__(self, path: Path) -> None: connection.executescript(SQLiteSessionStore._CREATE_TABLE_SQL) connection.executescript(SQLiteTapeStore._CREATE_SCHEMA_SQL) connection.executescript(SQLiteCheckpointStore._CREATE_SCHEMA_SQL) - connection.executescript(SQLiteRuntimeStore._CREATE_SCHEMA_SQL) + SQLiteRuntimeStore._ensure_schema(connection) connection.executescript(SQLiteTopicStore._CREATE_SCHEMA_SQL) connection.executescript(self._SESSION_TAPES_SCHEMA_SQL) @@ -724,9 +724,10 @@ async def create_agent_run( """ INSERT INTO agent_runs ( run_id, session_id, tape_id, parent_run_id, agent_id, status, - started_at, ended_at, metadata, result, error + started_at, ended_at, metadata, result, error, + superseded_by_checkpoint_id, superseded_at ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(run_id) DO UPDATE SET tape_id = excluded.tape_id, @@ -1060,6 +1061,7 @@ async def claim_attached_executor_run( if run.metadata.get("executor_ref_kind") in {"external_worker", "local_attached"} and run.metadata.get("executor_kind") == executor_kind + and run.superseded_at is None ] selected: AgentRunRecord | None = None authority: OwnerAuthority | None = None @@ -1291,6 +1293,22 @@ async def restore_checkpoint_state( """, (authority.session_id, json.dumps(session_payload, sort_keys=True)), ) + connection.execute( + """ + UPDATE agent_runs + SET superseded_by_checkpoint_id = ?, + superseded_at = ? + WHERE session_id = ? + AND julianday(started_at) > julianday(?) + AND superseded_at IS NULL + """, + ( + meta.checkpoint_id, + now, + authority.session_id, + _datetime_to_json(meta.created_at), + ), + ) self._reconcile_topics_after_checkpoint_restore( connection, tape_id=meta.tape_id, diff --git a/coding-agent/src/coding_agent/stores/durable_pg.py b/coding-agent/src/coding_agent/stores/durable_pg.py index 1e24e78e2..8529a7045 100644 --- a/coding-agent/src/coding_agent/stores/durable_pg.py +++ b/coding-agent/src/coding_agent/stores/durable_pg.py @@ -56,6 +56,15 @@ class PGDurableStore: DO UPDATE SET payload = EXCLUDED.payload, updated_at = NOW() """ _DELETE_SESSION_SQL = "DELETE FROM agent_http_sessions WHERE session_id = $1" + _SUPERSEDE_RUNS_AFTER_CHECKPOINT_SQL = """ + UPDATE agent_runs + SET superseded_by_checkpoint_id = $3, + superseded_at = NOW(), + updated_at = NOW() + WHERE session_id = $1 + AND started_at > $2 + AND superseded_at IS NULL + """ _SELECT_SESSION_FOR_UPDATE_SQL = """ SELECT payload FROM agent_http_sessions @@ -131,9 +140,13 @@ class PGDurableStore: ended_at, metadata, result, - error + error, + superseded_by_checkpoint_id, + superseded_at + ) + VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13 ) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10::jsonb, $11) ON CONFLICT (run_id) DO UPDATE SET tape_id = EXCLUDED.tape_id, @@ -446,6 +459,8 @@ async def body(connection: Any) -> AgentRunRecord: record.metadata, record.result, record.error, + record.superseded_by_checkpoint_id, + record.superseded_at, ) return _agent_run_from_row( _required_owned_row(row, "run target belongs to another owner") @@ -715,6 +730,12 @@ async def body(connection: Any) -> None: authority.session_id, session_payload, ) + await connection.execute( + self._SUPERSEDE_RUNS_AFTER_CHECKPOINT_SQL, + authority.session_id, + meta.created_at, + meta.checkpoint_id, + ) await self._reconcile_topics_after_checkpoint_restore( connection, tape_id=meta.tape_id, diff --git a/coding-agent/src/coding_agent/stores/runtime_store.py b/coding-agent/src/coding_agent/stores/runtime_store.py index 14ba0f329..6cd68cbd1 100644 --- a/coding-agent/src/coding_agent/stores/runtime_store.py +++ b/coding-agent/src/coding_agent/stores/runtime_store.py @@ -30,6 +30,8 @@ class AgentRunRecord: metadata: JSONObject = field(default_factory=dict) result: JSONObject = field(default_factory=dict) error: str | None = None + superseded_by_checkpoint_id: str | None = None + superseded_at: datetime | None = None def __post_init__(self) -> None: _require_non_empty("run_id", self.run_id) @@ -48,6 +50,13 @@ def __post_init__(self) -> None: _require_json_object("result", self.result) if self.error is not None: _require_non_empty("error", self.error) + if self.superseded_by_checkpoint_id is not None: + _require_non_empty( + "superseded_by_checkpoint_id", + self.superseded_by_checkpoint_id, + ) + if self.superseded_at is not None: + _require_datetime("superseded_at", self.superseded_at) @dataclass(frozen=True) @@ -157,6 +166,8 @@ async def update_agent_run( metadata=metadata, result=result, error=error, + superseded_by_checkpoint_id=existing.superseded_by_checkpoint_id, + superseded_at=existing.superseded_at, ) await self._append_jsonl("runs.jsonl", _agent_run_to_payload(updated)) return updated @@ -213,6 +224,8 @@ async def claim_attached_executor_run( metadata=cast(JSONObject, metadata), result=selected.result, error=selected.error, + superseded_by_checkpoint_id=selected.superseded_by_checkpoint_id, + superseded_at=selected.superseded_at, ) self._append_jsonl_sync("runs.jsonl", _agent_run_to_payload(claimed)) return claimed @@ -456,7 +469,9 @@ class SQLiteRuntimeStore: ended_at TEXT, metadata TEXT NOT NULL, result TEXT NOT NULL, - error TEXT + error TEXT, + superseded_by_checkpoint_id TEXT, + superseded_at TEXT ); CREATE INDEX IF NOT EXISTS agent_runs_session_id_idx @@ -506,7 +521,22 @@ def __init__(self, path: Path) -> None: self._lock = threading.Lock() self._path.parent.mkdir(parents=True, exist_ok=True) with self._connect() as connection: - connection.executescript(self._CREATE_SCHEMA_SQL) + self._ensure_schema(connection) + + @classmethod + def _ensure_schema(cls, connection: sqlite3.Connection) -> None: + connection.executescript(cls._CREATE_SCHEMA_SQL) + connection.execute("BEGIN IMMEDIATE") + columns = { + str(row[1]) + for row in connection.execute("PRAGMA table_info(agent_runs)").fetchall() + } + if "superseded_by_checkpoint_id" not in columns: + connection.execute( + "ALTER TABLE agent_runs ADD COLUMN superseded_by_checkpoint_id TEXT" + ) + if "superseded_at" not in columns: + connection.execute("ALTER TABLE agent_runs ADD COLUMN superseded_at TEXT") def _connect(self) -> sqlite3.Connection: connection = sqlite3.connect(self._path) @@ -519,9 +549,10 @@ async def create_agent_run(self, record: AgentRunRecord) -> AgentRunRecord: """ INSERT INTO agent_runs ( run_id, session_id, tape_id, parent_run_id, agent_id, status, - started_at, ended_at, metadata, result, error + started_at, ended_at, metadata, result, error, + superseded_by_checkpoint_id, superseded_at ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(run_id) DO UPDATE SET session_id = excluded.session_id, @@ -644,6 +675,7 @@ async def claim_attached_executor_run( if run.metadata.get("executor_ref_kind") in {"external_worker", "local_attached"} and run.metadata.get("executor_kind") == executor_kind + and run.superseded_at is None ] if not candidates: return None @@ -916,6 +948,8 @@ class PGRuntimeStore: metadata JSONB NOT NULL DEFAULT '{}'::jsonb, result JSONB NOT NULL DEFAULT '{}'::jsonb, error TEXT, + superseded_by_checkpoint_id TEXT, + superseded_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); @@ -923,6 +957,11 @@ class PGRuntimeStore: CREATE INDEX IF NOT EXISTS agent_runs_session_id_idx ON agent_runs (session_id, started_at, run_id); + ALTER TABLE agent_runs + ADD COLUMN IF NOT EXISTS superseded_by_checkpoint_id TEXT; + ALTER TABLE agent_runs + ADD COLUMN IF NOT EXISTS superseded_at TIMESTAMPTZ; + CREATE INDEX IF NOT EXISTS agent_runs_tape_id_idx ON agent_runs (tape_id, started_at, run_id) WHERE tape_id IS NOT NULL; @@ -990,9 +1029,13 @@ class PGRuntimeStore: ended_at, metadata, result, - error + error, + superseded_by_checkpoint_id, + superseded_at + ) + VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13 ) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10::jsonb, $11) ON CONFLICT (run_id) DO UPDATE SET session_id = EXCLUDED.session_id, @@ -1031,6 +1074,7 @@ class PGRuntimeStore: AND metadata->>'executor_ref_kind' IN ('external_worker', 'local_attached') AND metadata->>'executor_kind' = $2 AND ($1::text IS NULL OR session_id = $1) + AND superseded_at IS NULL ORDER BY started_at, run_id LIMIT 1 FOR UPDATE SKIP LOCKED @@ -1163,6 +1207,8 @@ async def create_agent_run(self, record: AgentRunRecord) -> AgentRunRecord: record.metadata, record.result, record.error, + record.superseded_by_checkpoint_id, + record.superseded_at, ) return _agent_run_from_row(_required_row(row, "agent run insert")) @@ -1484,6 +1530,12 @@ def _agent_run_sqlite_values(record: AgentRunRecord) -> tuple[object, ...]: _json_to_sql(record.metadata), _json_to_sql(record.result), record.error, + record.superseded_by_checkpoint_id, + ( + None + if record.superseded_at is None + else _datetime_to_json(record.superseded_at) + ), ) @@ -1500,6 +1552,16 @@ def _agent_run_from_sqlite_row(row: sqlite3.Row) -> AgentRunRecord: metadata=_json_object_from_sql(row["metadata"], context="agent run metadata"), result=_json_object_from_sql(row["result"], context="agent run result"), error=_sqlite_optional_str(row, "error", context="agent run"), + superseded_by_checkpoint_id=_sqlite_optional_str( + row, + "superseded_by_checkpoint_id", + context="agent run", + ), + superseded_at=_sqlite_optional_datetime( + row, + "superseded_at", + context="agent run", + ), ) @@ -1613,6 +1675,16 @@ def _agent_run_from_row(row: dict[str, object]) -> AgentRunRecord: metadata=_required_json_object(row, "metadata", context="agent run row"), result=_required_json_object(row, "result", context="agent run row"), error=_optional_str(row, "error", context="agent run row"), + superseded_by_checkpoint_id=_optional_str( + row, + "superseded_by_checkpoint_id", + context="agent run row", + ), + superseded_at=_optional_datetime( + row, + "superseded_at", + context="agent run row", + ), ) @@ -1708,6 +1780,12 @@ def _agent_run_to_payload(record: AgentRunRecord) -> JSONObject: "metadata": record.metadata, "result": record.result, "error": record.error, + "superseded_by_checkpoint_id": record.superseded_by_checkpoint_id, + "superseded_at": ( + None + if record.superseded_at is None + else _datetime_to_json(record.superseded_at) + ), } @@ -1750,6 +1828,16 @@ def _agent_run_from_payload(payload: JSONObject) -> AgentRunRecord: context="agent run payload", ), error=_optional_payload_str(payload, "error", context="agent run payload"), + superseded_by_checkpoint_id=_optional_payload_str( + payload, + "superseded_by_checkpoint_id", + context="agent run payload", + ), + superseded_at=_optional_payload_datetime( + payload, + "superseded_at", + context="agent run payload", + ), ) diff --git a/coding-agent/tests/acp/test_compat_harness.py b/coding-agent/tests/acp/test_compat_harness.py index ab3d27fc0..89f2434e6 100644 --- a/coding-agent/tests/acp/test_compat_harness.py +++ b/coding-agent/tests/acp/test_compat_harness.py @@ -231,6 +231,12 @@ async def list_runtime_runs(self, session_id: str) -> list[Any]: SimpleNamespace(run_id="run-1", started_at=datetime(2026, 1, 1, tzinfo=UTC)) ] + async def list_active_runtime_runs(self, session_id: str) -> list[Any]: + self.calls.append(("list_active_runtime_runs", session_id)) + return [ + SimpleNamespace(run_id="run-1", started_at=datetime(2026, 1, 1, tzinfo=UTC)) + ] + async def list_sessions_async(self) -> list[str]: self.calls.append(("list_sessions_async", None)) return ["sess-compat"] diff --git a/coding-agent/tests/acp/test_server.py b/coding-agent/tests/acp/test_server.py index c0ded91ad..17480888b 100644 --- a/coding-agent/tests/acp/test_server.py +++ b/coding-agent/tests/acp/test_server.py @@ -168,6 +168,10 @@ async def list_runtime_runs(self, session_id: str) -> list[Any]: self.calls.append(("list_runtime_runs", session_id)) return list(self.runtime_runs) + async def list_active_runtime_runs(self, session_id: str) -> list[Any]: + self.calls.append(("list_active_runtime_runs", session_id)) + return list(self.runtime_runs) + async def replay_display_events( self, run_id: str, @@ -1264,7 +1268,7 @@ async def test_session_load_replays_display_events_before_response( "update_session_additional_directories", {"session_id": "sess-1", "additional_directories": []}, ), - ("list_runtime_runs", "sess-1"), + ("list_active_runtime_runs", "sess-1"), ( "replay_display_events", {"run_id": "run-1", "last_event_id": None, "limit": 1000}, diff --git a/coding-agent/tests/cli/test_local_runtime.py b/coding-agent/tests/cli/test_local_runtime.py index a4b6bf46d..ac2e68eb8 100644 --- a/coding-agent/tests/cli/test_local_runtime.py +++ b/coding-agent/tests/cli/test_local_runtime.py @@ -28,6 +28,7 @@ def __init__(self) -> None: self.renew_owner_leases_event = asyncio.Event() self.release_owned_sessions_called = False self.close_called = False + self.active_runtime_run_requests = [] async def _persist_session_async(self, session) -> None: self.persisted.append(session) @@ -52,6 +53,10 @@ async def renew_owner_leases(self) -> None: async def close(self) -> None: self.close_called = True + async def list_active_runtime_runs(self, session_id: str): + self.active_runtime_run_requests.append(session_id) + return [SimpleNamespace(run_id="run-active")] + async def test_server_backed_local_cli_attach_runtime_uses_session_binding_delegate() -> ( None @@ -101,6 +106,17 @@ async def test_server_backed_local_cli_delegates_owner_acquire() -> None: assert delegate.owner_acquires == ["session-1"] +async def test_server_backed_local_cli_delegates_active_runtime_runs() -> None: + manager = object.__new__(ServerBackedLocalCliSessionManager) + delegate = FakeDelegate() + manager._delegate = delegate + + runs = await manager.list_active_runtime_runs("session-1") + + assert [run.run_id for run in runs] == ["run-active"] + assert delegate.active_runtime_run_requests == ["session-1"] + + async def test_server_backed_local_cli_releases_owner_leases_before_close() -> None: manager = object.__new__(ServerBackedLocalCliSessionManager) delegate = FakeDelegate() diff --git a/coding-agent/tests/coding_agent/test_jsonl_runtime_store.py b/coding-agent/tests/coding_agent/test_jsonl_runtime_store.py index 58f3f82d4..785591f1c 100644 --- a/coding-agent/tests/coding_agent/test_jsonl_runtime_store.py +++ b/coding-agent/tests/coding_agent/test_jsonl_runtime_store.py @@ -18,6 +18,7 @@ async def test_jsonl_runtime_store_persists_runs_and_events_across_instances( root = tmp_path / "runtime" first = JSONLRuntimeStore(root) started_at = datetime(2026, 6, 1, 12, 0, tzinfo=UTC) + superseded_at = datetime(2026, 6, 1, 12, 2, tzinfo=UTC) await first.create_agent_run( AgentRunRecord( @@ -31,6 +32,8 @@ async def test_jsonl_runtime_store_persists_runs_and_events_across_instances( metadata={"execution_placement": "server_embedded"}, result={}, error=None, + superseded_by_checkpoint_id="checkpoint-1", + superseded_at=superseded_at, ) ) await first.update_agent_run( @@ -61,6 +64,8 @@ async def test_jsonl_runtime_store_persists_runs_and_events_across_instances( assert loaded is not None assert loaded.status == "completed" assert loaded.metadata["resume_from_run_id"] == "run-0" + assert loaded.superseded_by_checkpoint_id == "checkpoint-1" + assert loaded.superseded_at == superseded_at assert event.sequence == 1 assert [item.event_id for item in events] == ["event-1"] assert events[0].payload["resume_from_run_id"] == "run-0" diff --git a/coding-agent/tests/coding_agent/test_pg_durable_fencing.py b/coding-agent/tests/coding_agent/test_pg_durable_fencing.py index 2522c963d..6be64f608 100644 --- a/coding-agent/tests/coding_agent/test_pg_durable_fencing.py +++ b/coding-agent/tests/coding_agent/test_pg_durable_fencing.py @@ -1,6 +1,7 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta +from collections.abc import Mapping from typing import Any, cast from unittest.mock import patch @@ -53,6 +54,7 @@ def __init__(self) -> None: self.checkpoint_meta_row: dict[str, object] | None = { "meta": {"session_id": "session-a"} } + self.agent_runs: dict[str, dict[str, object]] = {} self.topics: dict[str, dict[str, object]] = {} self.topic_anchors: list[dict[str, object]] = [] self.topic_recall_links: list[dict[str, object]] = [] @@ -77,6 +79,19 @@ async def execute(self, query: str, *args: object) -> str: "fencing_token": args[3], } return "UPDATE 1" + if "UPDATE agent_runs" in query and "superseded_at IS NULL" in query: + session_id = cast(str, args[0]) + checkpoint_created_at = cast(datetime, args[1]) + checkpoint_id = cast(str, args[2]) + for run in self.agent_runs.values(): + if ( + run["session_id"] == session_id + and cast(datetime, run["started_at"]) > checkpoint_created_at + and run["superseded_at"] is None + ): + run["superseded_by_checkpoint_id"] = checkpoint_id + run["superseded_at"] = datetime.now(UTC) + return "UPDATE" if "INSERT INTO session_tapes" in query: session_id = cast(str, args[0]) tape_id = cast(str, args[1]) @@ -367,6 +382,53 @@ def __init__(self, *, pool: FakePGPool) -> None: self.pool = pool +@pytest.mark.asyncio +async def test_fenced_pg_runtime_claim_uses_durable_active_claim_path() -> None: + pool = FakePGPool() + authority = OwnerAuthority("session-a", "owner-a", 7) + expected = AgentRunRecord( + run_id="run-active", + session_id="session-a", + tape_id="tape-a", + parent_run_id=None, + agent_id=None, + status="claimed", + started_at=datetime(2026, 6, 1, 12, 0, tzinfo=UTC), + ) + calls: list[tuple[Mapping[str, OwnerAuthority], str | None, str]] = [] + + class DurableStore: + async def claim_attached_executor_run( + self, + authorities, + *, + session_id, + executor_kind, + claim_metadata, + ): + del claim_metadata + calls.append((authorities, session_id, executor_kind)) + return expected + + store = FencedPGRuntimeStore( + durable_store=cast(Any, DurableStore()), + pool=cast(Any, pool), + authority_for_session=lambda _: authority, + authorities=lambda: {authority.session_id: authority}, + ) + + claimed = await store.claim_attached_executor_run( + session_id=authority.session_id, + executor_kind="local_cli", + claim_metadata={"worker_id": "worker-1"}, + ) + + assert claimed == expected + assert calls == [ + ({authority.session_id: authority}, authority.session_id, "local_cli") + ] + + @pytest.mark.asyncio async def test_pg_owner_store_acquire_authority_allocates_db_epoch() -> None: pool = FakePGPool() @@ -669,6 +731,38 @@ async def test_pg_durable_restore_reconciles_topic_projection_with_checkpoint() pool.connection.session_tape_by_session["session-a"] = "tape-a" pool.connection.session_by_tape["tape-a"] = "session-a" pool.connection.checkpoint_meta_row = {"meta": {"session_id": "session-a"}} + pool.connection.agent_runs = { + "run-before": { + "session_id": "session-a", + "started_at": checkpoint_created_at - timedelta(seconds=1), + "superseded_by_checkpoint_id": None, + "superseded_at": None, + }, + "run-at-boundary": { + "session_id": "session-a", + "started_at": checkpoint_created_at, + "superseded_by_checkpoint_id": None, + "superseded_at": None, + }, + "run-after": { + "session_id": "session-a", + "started_at": checkpoint_created_at + timedelta(seconds=1), + "superseded_by_checkpoint_id": None, + "superseded_at": None, + }, + "run-already-superseded": { + "session_id": "session-a", + "started_at": checkpoint_created_at + timedelta(seconds=2), + "superseded_by_checkpoint_id": "checkpoint-original", + "superseded_at": checkpoint_created_at + timedelta(seconds=3), + }, + "run-other-session": { + "session_id": "session-b", + "started_at": checkpoint_created_at + timedelta(seconds=1), + "superseded_by_checkpoint_id": None, + "superseded_at": None, + }, + } pool.connection.topics = { "topic-keep": { "topic_id": "topic-keep", @@ -810,6 +904,20 @@ async def test_pg_durable_restore_reconciles_topic_projection_with_checkpoint() } ] assert set(pool.connection.topic_costs) == {"topic-other"} + assert pool.connection.agent_runs["run-before"]["superseded_at"] is None + assert pool.connection.agent_runs["run-at-boundary"]["superseded_at"] is None + assert ( + pool.connection.agent_runs["run-after"]["superseded_by_checkpoint_id"] + == "checkpoint-keep" + ) + assert pool.connection.agent_runs["run-after"]["superseded_at"] is not None + assert ( + pool.connection.agent_runs["run-already-superseded"][ + "superseded_by_checkpoint_id" + ] + == "checkpoint-original" + ) + assert pool.connection.agent_runs["run-other-session"]["superseded_at"] is None @pytest.mark.asyncio diff --git a/coding-agent/tests/coding_agent/test_pg_runtime_store.py b/coding-agent/tests/coding_agent/test_pg_runtime_store.py index 476fba9c0..11752771b 100644 --- a/coding-agent/tests/coding_agent/test_pg_runtime_store.py +++ b/coding-agent/tests/coding_agent/test_pg_runtime_store.py @@ -45,6 +45,8 @@ async def fetchrow(self, query: str, *args: object) -> dict[str, object] | None: metadata, result, error, + superseded_by_checkpoint_id, + superseded_at, ) = args row = { "run_id": run_id, @@ -58,6 +60,8 @@ async def fetchrow(self, query: str, *args: object) -> dict[str, object] | None: "metadata": metadata, "result": result, "error": error, + "superseded_by_checkpoint_id": superseded_by_checkpoint_id, + "superseded_at": superseded_at, } self.agent_runs[cast(str, run_id)] = row return row @@ -76,6 +80,11 @@ async def fetchrow(self, query: str, *args: object) -> dict[str, object] | None: continue if metadata.get("executor_kind") != executor_kind: continue + if ( + "superseded_at IS NULL" in query + and row["superseded_at"] is not None + ): + continue row["status"] = "claimed" row["metadata"] = { **metadata, @@ -305,6 +314,33 @@ async def test_create_update_load_and_list_agent_runs( assert [run.run_id for run in listed] == ["run-1", "run-2"] +@pytest.mark.asyncio +async def test_load_agent_run_round_trips_supersession_fields( + store: PGRuntimeStore, +) -> None: + superseded_at = _dt(10) + await store.create_agent_run( + AgentRunRecord( + run_id="run-superseded", + session_id="session-1", + tape_id="tape-1", + parent_run_id=None, + agent_id=None, + status="completed", + started_at=_dt(9), + ended_at=_dt(9, 30), + superseded_by_checkpoint_id="checkpoint-1", + superseded_at=superseded_at, + ) + ) + + loaded = await store.load_agent_run("run-superseded") + + assert loaded is not None + assert loaded.superseded_by_checkpoint_id == "checkpoint-1" + assert loaded.superseded_at == superseded_at + + @pytest.mark.asyncio async def test_claim_external_worker_run_marks_requested_run_claimed( store: PGRuntimeStore, @@ -386,6 +422,49 @@ async def test_claim_attached_executor_run_accepts_local_attached_binding( assert claimed.metadata["worker_id"] == "worker-1" +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["requested", "expired"]) +async def test_claim_attached_executor_run_skips_superseded_run( + store: PGRuntimeStore, + status: str, +) -> None: + started_at = _dt(9) + for run_id, run_started_at, superseded_at in ( + ("run-superseded", started_at, _dt(10)), + ("run-active", _dt(9, 1), None), + ): + await store.create_agent_run( + AgentRunRecord( + run_id=run_id, + session_id="session-local-attached", + tape_id=None, + parent_run_id=None, + agent_id=None, + status=status, + started_at=run_started_at, + metadata={ + "executor_ref_kind": "local_attached", + "executor_kind": "local_cli", + }, + result={}, + error=None, + superseded_by_checkpoint_id=( + "checkpoint-1" if superseded_at is not None else None + ), + superseded_at=superseded_at, + ) + ) + + claimed = await store.claim_attached_executor_run( + session_id="session-local-attached", + executor_kind="local_cli", + claim_metadata={"worker_id": "worker-1"}, + ) + + assert claimed is not None + assert claimed.run_id == "run-active" + + @pytest.mark.asyncio async def test_append_runtime_event_replays_in_sequence_order( store: PGRuntimeStore, @@ -607,3 +686,5 @@ async def test_pg_runtime_store_schema_initialization_is_idempotent( assert "CREATE TABLE IF NOT EXISTS runtime_events" in schema_calls[0] assert "CREATE TABLE IF NOT EXISTS run_message_snapshots" in schema_calls[0] assert "CREATE TABLE IF NOT EXISTS agent_interactions" in schema_calls[0] + assert "ADD COLUMN IF NOT EXISTS superseded_by_checkpoint_id" in schema_calls[0] + assert "ADD COLUMN IF NOT EXISTS superseded_at" in schema_calls[0] diff --git a/coding-agent/tests/coding_agent/test_runtime_query_service.py b/coding-agent/tests/coding_agent/test_runtime_query_service.py index d25f84f91..988d2ec62 100644 --- a/coding-agent/tests/coding_agent/test_runtime_query_service.py +++ b/coding-agent/tests/coding_agent/test_runtime_query_service.py @@ -179,6 +179,42 @@ async def test_runtime_query_service_reports_resume_metadata() -> None: } +@pytest.mark.asyncio +async def test_runtime_query_service_separates_active_timeline_from_audit_runs() -> ( + None +): + store = RecordingRuntimeQueryStore() + store.runs.extend( + [ + _run( + "run-active", + status="completed", + started_at=datetime(2026, 5, 19, 11, 0, tzinfo=UTC), + ), + _run( + "run-superseded", + status="completed", + started_at=datetime(2026, 5, 19, 12, 0, tzinfo=UTC), + superseded_by_checkpoint_id="cp-restore", + superseded_at=datetime(2026, 5, 19, 13, 0, tzinfo=UTC), + ), + ] + ) + + service = RuntimeQueryService(store) + + assert [run.run_id for run in await service.list_runtime_runs("session-1")] == [ + "run-active", + "run-superseded", + ] + assert [ + run.run_id for run in await service.list_active_runtime_runs("session-1") + ] == ["run-active"] + latest = await service.latest_runtime_run("session-1") + assert latest is not None + assert latest.run_id == "run-active" + + @pytest.mark.asyncio async def test_runtime_query_service_marks_active_run_not_resumable() -> None: store = RecordingRuntimeQueryStore() @@ -272,6 +308,8 @@ def _run( *, status: str, started_at: datetime, + superseded_by_checkpoint_id: str | None = None, + superseded_at: datetime | None = None, ) -> AgentRunRecord: return AgentRunRecord( run_id=run_id, @@ -283,6 +321,8 @@ def _run( started_at=started_at, metadata={}, result={}, + superseded_by_checkpoint_id=superseded_by_checkpoint_id, + superseded_at=superseded_at, ) diff --git a/coding-agent/tests/coding_agent/test_sqlite_local_durable_fencing.py b/coding-agent/tests/coding_agent/test_sqlite_local_durable_fencing.py index 3ecfeca40..043edf2e0 100644 --- a/coding-agent/tests/coding_agent/test_sqlite_local_durable_fencing.py +++ b/coding-agent/tests/coding_agent/test_sqlite_local_durable_fencing.py @@ -12,7 +12,11 @@ from agentkit.errors import ConfigError from coding_agent.stores.durable_local import SQLiteLocalDurableStore from coding_agent.stores.local import local_sqlite_path, local_sqlite_storage_config -from coding_agent.stores.runtime_store import AgentRunRecord, RuntimeEventRecord +from coding_agent.stores.runtime_store import ( + AgentRunRecord, + RuntimeEventRecord, + SQLiteRuntimeStore, +) from coding_agent.server.session_manager import SessionManager from coding_agent.server.stores.session_owner_store import ( SQLiteSessionOwnerStore, @@ -54,6 +58,57 @@ async def load(self, tape_id: str) -> list[dict[str, object]]: return [] +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["requested", "expired"]) +async def test_sqlite_durable_claim_skips_superseded_run( + tmp_path: Path, + status: str, +) -> None: + store = SQLiteLocalDurableStore(tmp_path / "local.sqlite3") + owner = await store.acquire_owner("session-a", "owner-a") + await store.save_session( + owner, + {"id": "session-a", "session_id": "session-a", "tape_id": "tape-a"}, + ) + started_at = datetime(2026, 6, 1, 12, 0, tzinfo=UTC) + for run_id, run_started_at, superseded_at in ( + ("run-superseded", started_at, started_at), + ("run-active", started_at + timedelta(minutes=1), None), + ): + await store.create_agent_run( + owner, + AgentRunRecord( + run_id=run_id, + session_id="session-a", + tape_id="tape-a", + parent_run_id=None, + agent_id=None, + status=status, + started_at=run_started_at, + metadata={ + "executor_ref_kind": "local_attached", + "executor_kind": "local_cli", + }, + result={}, + error=None, + superseded_by_checkpoint_id=( + "checkpoint-1" if superseded_at is not None else None + ), + superseded_at=superseded_at, + ), + ) + + claimed = await store.claim_attached_executor_run( + {"session-a": owner}, + session_id="session-a", + executor_kind="local_cli", + claim_metadata={"worker_id": "worker-1"}, + ) + + assert claimed is not None + assert claimed.run_id == "run-active" + + @pytest.mark.asyncio async def test_sqlite_owner_authority_renew_does_not_advance_epoch( tmp_path: Path, @@ -984,6 +1039,35 @@ async def test_sqlite_local_durable_store_restores_checkpoint_state_atomically( plugin_states={}, ), ) + for run_id, started_at in ( + ("run-before", created_at - timedelta(seconds=1)), + ("run-at-boundary", created_at), + ("run-after", created_at + timedelta(seconds=1)), + ): + await store.create_agent_run( + owner, + AgentRunRecord( + run_id=run_id, + session_id="session-a", + tape_id="tape-a", + parent_run_id=None, + agent_id=None, + status="completed", + started_at=started_at, + metadata={}, + result={"content": run_id}, + ), + ) + await store.append_runtime_event( + owner, + RuntimeEventRecord( + event_id="event-after", + run_id="run-after", + event_kind="wire.StreamDelta", + payload={"content": "must remain available for audit"}, + created_at=created_at + timedelta(seconds=2), + ), + ) await store.save_checkpoint( owner, CheckpointSnapshot( @@ -1224,6 +1308,18 @@ async def test_sqlite_local_durable_store_restores_checkpoint_state_atomically( } assert await store.load_checkpoint("checkpoint-keep") is not None assert await store.load_checkpoint("checkpoint-drop") is None + runtime_reader = SQLiteRuntimeStore(tmp_path / "local.sqlite3") + runs = { + run.run_id: run for run in await runtime_reader.list_agent_runs("session-a") + } + assert runs["run-before"].superseded_at is None + assert runs["run-at-boundary"].superseded_at is None + assert runs["run-after"].superseded_by_checkpoint_id == "checkpoint-keep" + assert runs["run-after"].superseded_at is not None + assert [ + event.event_id + for event in await runtime_reader.replay_runtime_events("run-after") + ] == ["event-after"] topic_reader = SQLiteTopicStore(tmp_path / "local.sqlite3") restored_topics = await topic_reader.list_topics(tape_id="tape-a") topics_by_id = {topic.topic_id: topic for topic in restored_topics} diff --git a/coding-agent/tests/coding_agent/test_sqlite_runtime_store.py b/coding-agent/tests/coding_agent/test_sqlite_runtime_store.py index 038f63784..729e0539e 100644 --- a/coding-agent/tests/coding_agent/test_sqlite_runtime_store.py +++ b/coding-agent/tests/coding_agent/test_sqlite_runtime_store.py @@ -1,5 +1,6 @@ from __future__ import annotations +import sqlite3 from datetime import UTC, datetime import pytest @@ -13,6 +14,50 @@ ) +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["requested", "expired"]) +async def test_sqlite_runtime_store_claim_skips_superseded_run( + tmp_path, + status: str, +) -> None: + store = SQLiteRuntimeStore(tmp_path / "runtime.sqlite3") + started_at = datetime(2026, 6, 1, 12, 0, tzinfo=UTC) + for run_id, run_started_at, superseded_at in ( + ("run-superseded", started_at, started_at), + ("run-active", started_at.replace(minute=1), None), + ): + await store.create_agent_run( + AgentRunRecord( + run_id=run_id, + session_id="session-1", + tape_id="tape-1", + parent_run_id=None, + agent_id=None, + status=status, + started_at=run_started_at, + metadata={ + "executor_ref_kind": "local_attached", + "executor_kind": "local_cli", + }, + result={}, + error=None, + superseded_by_checkpoint_id=( + "checkpoint-1" if superseded_at is not None else None + ), + superseded_at=superseded_at, + ) + ) + + claimed = await store.claim_attached_executor_run( + session_id="session-1", + executor_kind="local_cli", + claim_metadata={"executor_id": "executor-1"}, + ) + + assert claimed is not None + assert claimed.run_id == "run-active" + + @pytest.mark.asyncio async def test_sqlite_runtime_store_persists_runtime_records_across_instances( tmp_path, @@ -148,3 +193,61 @@ async def test_sqlite_runtime_store_normalizes_blank_error_to_none(tmp_path) -> loaded = await store.load_agent_run("run-1") assert loaded is not None assert loaded.error is None + + +@pytest.mark.asyncio +async def test_sqlite_runtime_store_round_trips_run_supersession(tmp_path) -> None: + path = tmp_path / "runtime.sqlite3" + store = SQLiteRuntimeStore(path) + superseded_at = datetime(2026, 6, 1, 13, 0, tzinfo=UTC) + + await store.create_agent_run( + AgentRunRecord( + run_id="run-superseded", + session_id="session-1", + tape_id="tape-1", + parent_run_id=None, + agent_id=None, + status="completed", + started_at=datetime(2026, 6, 1, 12, 0, tzinfo=UTC), + superseded_by_checkpoint_id="checkpoint-1", + superseded_at=superseded_at, + ) + ) + + loaded = await SQLiteRuntimeStore(path).load_agent_run("run-superseded") + + assert loaded is not None + assert loaded.superseded_by_checkpoint_id == "checkpoint-1" + assert loaded.superseded_at == superseded_at + + +def test_sqlite_runtime_store_migrates_legacy_agent_runs_schema(tmp_path) -> None: + path = tmp_path / "runtime.sqlite3" + with sqlite3.connect(path) as connection: + connection.execute( + """ + CREATE TABLE agent_runs ( + run_id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + tape_id TEXT, + parent_run_id TEXT, + agent_id TEXT, + status TEXT NOT NULL, + started_at TEXT NOT NULL, + ended_at TEXT, + metadata TEXT NOT NULL, + result TEXT NOT NULL, + error TEXT + ) + """ + ) + + SQLiteRuntimeStore(path) + + with sqlite3.connect(path) as connection: + columns = { + str(row[1]) + for row in connection.execute("PRAGMA table_info(agent_runs)").fetchall() + } + assert {"superseded_by_checkpoint_id", "superseded_at"} <= columns diff --git a/coding-agent/tests/ui/test_http_server.py b/coding-agent/tests/ui/test_http_server.py index 19c3304a3..c7d1118d5 100644 --- a/coding-agent/tests/ui/test_http_server.py +++ b/coding-agent/tests/ui/test_http_server.py @@ -4769,8 +4769,17 @@ async def test_external_worker_session_runs_endpoint_lists_runs(self, client): lease_seconds=30, ) assert claim is not None + superseded_run = replace( + run, + run_id="superseded-run", + started_at=run.started_at + timedelta(seconds=1), + superseded_by_checkpoint_id="checkpoint-1", + superseded_at=run.started_at + timedelta(seconds=2), + ) + store.runs[superseded_run.run_id] = superseded_run response = await client.get(f"/sessions/{session_id}/runs") + audit_response = await client.get(f"/runs/{superseded_run.run_id}") assert response.status_code == 200 payload = response.json() @@ -4778,6 +4787,8 @@ async def test_external_worker_session_runs_endpoint_lists_runs(self, client): assert [item["run_id"] for item in payload["runs"]] == [run.run_id] assert payload["runs"][0]["metadata"]["worker_id"] == "worker-1" assert "claim_token_hash" not in payload["runs"][0]["metadata"] + assert audit_response.status_code == 200 + assert audit_response.json()["run_id"] == superseded_run.run_id async def test_external_worker_workers_endpoint_reports_running_and_stale_workers( self, client, tmp_path: Path @@ -7700,6 +7711,71 @@ async def test_session_result_restores_from_persisted_runtime_events_when_tape_i == "Tool activity: shell_command: uv run pytest tests/foo.py -q" ) + async def test_session_result_ignores_superseded_runtime_run( + self, + client: AsyncClient, + ) -> None: + session_id = "result-active-timeline" + register_session( + session_id, + default_run_target=_local_run_target(Path.cwd()), + tape_id="empty-active-timeline-tape", + ) + store = FakeExternalWorkerRuntimeStore() + active_run = AgentRunRecord( + run_id="run-active", + session_id=session_id, + tape_id="empty-active-timeline-tape", + parent_run_id=None, + agent_id=None, + status="completed", + started_at=datetime(2026, 1, 2, 3, 4, 5), + ended_at=datetime(2026, 1, 2, 3, 5, 0), + ) + superseded_run = replace( + active_run, + run_id="run-superseded", + started_at=datetime(2026, 1, 2, 3, 6, 0), + ended_at=datetime(2026, 1, 2, 3, 7, 0), + superseded_by_checkpoint_id="checkpoint-1", + superseded_at=datetime(2026, 1, 2, 3, 8, 0), + ) + store.runs = { + active_run.run_id: active_run, + superseded_run.run_id: superseded_run, + } + store.events = [ + RuntimeEventRecord( + sequence=1, + event_id="event-active-text", + run_id=active_run.run_id, + event_kind="wire.StreamDelta", + payload={ + "message_type": "StreamDelta", + "message": {"content": "Restored answer.", "role": "assistant"}, + }, + created_at=datetime(2026, 1, 2, 3, 4, 30), + ), + RuntimeEventRecord( + sequence=2, + event_id="event-superseded-text", + run_id=superseded_run.run_id, + event_kind="wire.StreamDelta", + payload={ + "message_type": "StreamDelta", + "message": {"content": "Rolled back answer.", "role": "assistant"}, + }, + created_at=datetime(2026, 1, 2, 3, 6, 30), + ), + ] + session_manager.configure_runtime_store(store) + + response = await client.get(f"/sessions/{session_id}/result") + + assert response.status_code == 200 + assert response.json()["turn_id"] == active_run.run_id + assert response.json()["final_answer"] == "Restored answer." + async def test_session_result_includes_recorded_failure_details( self, client: AsyncClient ) -> None: @@ -10467,6 +10543,116 @@ async def fake_restore_checkpoint( "checkpoint_id": "cp-http-restore", } + async def test_restore_checkpoint_reconciles_summary_and_result_active_run( + self, + client: AsyncClient, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + session_id = "restore-active-run-session" + session = register_session(session_id, tape_id="restore-active-run-tape") + active_run = AgentRunRecord( + run_id="run-active", + session_id=session_id, + tape_id=session.tape_id, + parent_run_id=None, + agent_id=None, + status="completed", + started_at=datetime(2026, 1, 2, 3, 4, 0), + ended_at=datetime(2026, 1, 2, 3, 5, 0), + ) + superseded_run = replace( + active_run, + run_id="run-superseded", + started_at=datetime(2026, 1, 2, 3, 6, 0), + ended_at=datetime(2026, 1, 2, 3, 7, 0), + superseded_by_checkpoint_id="checkpoint-1", + superseded_at=datetime(2026, 1, 2, 3, 8, 0), + ) + runtime_store = FakeExternalWorkerRuntimeStore() + runtime_store.runs = { + active_run.run_id: active_run, + superseded_run.run_id: superseded_run, + } + runtime_store.events = [ + RuntimeEventRecord( + sequence=1, + event_id="event-active", + run_id=active_run.run_id, + event_kind="wire.StreamDelta", + payload={ + "message_type": "StreamDelta", + "message": {"content": "Restored answer.", "role": "assistant"}, + }, + created_at=datetime(2026, 1, 2, 3, 4, 30), + ), + RuntimeEventRecord( + sequence=2, + event_id="event-superseded", + run_id=superseded_run.run_id, + event_kind="wire.StreamDelta", + payload={ + "message_type": "StreamDelta", + "message": { + "content": "Rolled back answer.", + "role": "assistant", + }, + }, + created_at=datetime(2026, 1, 2, 3, 6, 30), + ), + ] + session.current_turn_id = superseded_run.run_id + session_manager.configure_runtime_store(runtime_store) + + async def fake_restore(restored_session, checkpoint_id: str) -> None: + assert restored_session is session + assert checkpoint_id == "checkpoint-1" + + monkeypatch.setattr( + session_manager, + "_runtime_checkpoint_restore_service", + SimpleNamespace(restore=fake_restore), + ) + + restore_response = await client.post( + f"/sessions/{session_id}/checkpoints/checkpoint-1/restore" + ) + summary_response = await client.get(f"/sessions/{session_id}") + result_response = await client.get(f"/sessions/{session_id}/result") + + assert restore_response.status_code == 200 + assert summary_response.json()["turn_id"] == active_run.run_id + assert result_response.json()["turn_id"] == active_run.run_id + assert result_response.json()["final_answer"] == "Restored answer." + + new_run = replace( + active_run, + run_id="run-after-restore", + started_at=datetime(2026, 1, 2, 3, 9, 0), + ended_at=datetime(2026, 1, 2, 3, 10, 0), + ) + runtime_store.runs[new_run.run_id] = new_run + runtime_store.events.append( + RuntimeEventRecord( + sequence=3, + event_id="event-after-restore", + run_id=new_run.run_id, + event_kind="wire.StreamDelta", + payload={ + "message_type": "StreamDelta", + "message": {"content": "New answer.", "role": "assistant"}, + }, + created_at=datetime(2026, 1, 2, 3, 9, 30), + ) + ) + session.current_turn_id = new_run.run_id + + new_summary = await client.get(f"/sessions/{session_id}") + new_result = await client.get(f"/sessions/{session_id}/result") + + assert new_summary.json()["turn_id"] == new_run.run_id + assert new_result.json()["turn_id"] == new_run.run_id + assert new_result.json()["final_answer"] == "New answer." + async def test_restore_checkpoint_returns_409_for_active_turn(self, client): session_id = "restore-busy-session" register_session(session_id) diff --git a/coding-agent/tests/ui/test_session_manager_runtime.py b/coding-agent/tests/ui/test_session_manager_runtime.py index 36ab2748d..a7ecd2f98 100644 --- a/coding-agent/tests/ui/test_session_manager_runtime.py +++ b/coding-agent/tests/ui/test_session_manager_runtime.py @@ -118,6 +118,65 @@ def _external_worker_run_target() -> RunTarget: ) +@pytest.mark.asyncio +async def test_restore_durable_state_persists_reconciled_current_turn() -> None: + manager = SessionManager(store=InMemorySessionStore()) + session_id = await manager.create_session() + session = manager.get_session(session_id) + session.tape_id = "tape-1" + checkpoint_created_at = datetime(2026, 6, 1, 12, 0, tzinfo=UTC) + active_run = AgentRunRecord( + run_id="run-active", + session_id=session_id, + tape_id=session.tape_id, + parent_run_id=None, + agent_id=None, + status="completed", + started_at=checkpoint_created_at, + ) + stale_run = AgentRunRecord( + run_id="run-stale", + session_id=session_id, + tape_id=session.tape_id, + parent_run_id=None, + agent_id=None, + status="completed", + started_at=checkpoint_created_at + timedelta(seconds=1), + ) + + class RuntimeStore: + async def list_agent_runs(self, requested_session_id: str): + assert requested_session_id == session_id + return [active_run, stale_run] + + persisted_payloads: list[dict[str, Any]] = [] + + class DurableStore: + async def restore_checkpoint_state(self, authority, snapshot, payload): + del authority, snapshot + persisted_payloads.append(payload) + + manager.configure_runtime_store(cast(Any, RuntimeStore())) + manager._local_durable_store = cast(Any, DurableStore()) + manager._owner_authority_for_session = cast(Any, lambda _: object()) + session.current_turn_id = stale_run.run_id + + await manager._restore_checkpoint_durable_state( + session, + types.SimpleNamespace( + meta=types.SimpleNamespace(created_at=checkpoint_created_at) + ), + ) + + assert session.current_turn_id == active_run.run_id + assert persisted_payloads[0]["turn_id"] == active_run.run_id + restored_session = type(session).from_store_data(persisted_payloads[0]) + assert restored_session.current_turn_id == active_run.run_id + legacy_payload = dict(persisted_payloads[0]) + legacy_payload.pop("turn_id") + assert type(session).from_store_data(legacy_payload).current_turn_id is None + + def test_selected_topic_store_returns_fenced_sqlite_store_for_local_durable_bundle( tmp_path: Path, ) -> None: diff --git a/coding-agent/webui/app/src/App.test.tsx b/coding-agent/webui/app/src/App.test.tsx index 28e12dde8..6a82b85a3 100644 --- a/coding-agent/webui/app/src/App.test.tsx +++ b/coding-agent/webui/app/src/App.test.tsx @@ -1958,6 +1958,90 @@ describe("P2 race guards", () => { vi.restoreAllMocks(); }); + it("rebuilds messages from the active run list after checkpoint restore", async () => { + const active = session("session-restore-active-0001", "model-restore", { + checkpoint_count: 1, + }); + const checkpoint = { + checkpoint_id: "cp-active", + tape_id: "tape-active", + session_id: active.session_id, + entry_count: 1, + window_start: 0, + created_at: "2026-06-12T00:00:00Z", + label: "active-boundary", + }; + let runsCalls = 0; + + const displayEvent = (runId: string, content: string) => ({ + source_event_id: `${runId}-event`, + run_id: runId, + sequence: 1, + display_kind: "assistant_text_delta", + payload: { content, role: "assistant" }, + created_at: "2026-06-12T00:01:00Z", + }); + + vi.stubGlobal( + "fetch", + vi.fn((input: RequestInfo | URL) => { + const url = String(input); + if (url.endsWith("/sessions")) { + return Promise.resolve(jsonResponse({ sessions: [active] })); + } + if (url.endsWith(`/sessions/${active.session_id}/checkpoints`)) { + return Promise.resolve(jsonResponse({ checkpoints: [checkpoint] })); + } + if (url.endsWith(`/sessions/${active.session_id}/checkpoints/cp-active/restore`)) { + return Promise.resolve(jsonResponse({})); + } + if (url.endsWith(`/sessions/${active.session_id}`)) { + return Promise.resolve(jsonResponse(active)); + } + if (url.endsWith(`/sessions/${active.session_id}/runs`)) { + runsCalls += 1; + return Promise.resolve( + jsonResponse({ + session_id: active.session_id, + runs: runsCalls === 1 + ? [{ run_id: "run-active" }, { run_id: "run-rolled-back" }] + : [{ run_id: "run-active" }], + }), + ); + } + if (url.includes("/runs/run-active/display-events?")) { + return Promise.resolve( + jsonResponse({ + run_id: "run-active", + events: [displayEvent("run-active", "Restored answer")], + }), + ); + } + if (url.includes("/runs/run-rolled-back/display-events?")) { + return Promise.resolve( + jsonResponse({ + run_id: "run-rolled-back", + events: [displayEvent("run-rolled-back", "Rolled back answer")], + }), + ); + } + throw new Error(`unexpected fetch ${url}`); + }), + ); + + render(); + + fireEvent.click(await screen.findByRole("button", { name: /model-restore/i })); + expect(await screen.findByText("Rolled back answer")).toBeTruthy(); + fireEvent.click(screen.getByRole("button", { name: "Toggle checkpoints panel" })); + fireEvent.click(await screen.findByRole("button", { name: "Restore" })); + fireEvent.click(screen.getByRole("button", { name: "Confirm restore" })); + + await waitFor(() => expect(runsCalls).toBe(2)); + await waitFor(() => expect(screen.queryByText("Rolled back answer")).toBeNull()); + expect(screen.getByText("Restored answer")).toBeTruthy(); + }); + it("applies only the active session's diff when fetches resolve out of order", async () => { const first = session("session-diff-0001", "model-diff-a"); const second = session("session-diff-0002", "model-diff-b");