From 066e7b1a33f28bf78515fea2e5396c68d731ad6e Mon Sep 17 00:00:00 2001 From: kina0630 Date: Wed, 5 Aug 2026 10:26:41 +0800 Subject: [PATCH 1/2] fix(checkpoint): reconcile active run timeline Preserve superseded runs for audit while excluding them from active replay, resume, ACP, and result projections. Task: coding-agent/.opencode/prompts/tasks/checkpoint-active-run-timeline.md --- .../tasks/checkpoint-active-run-timeline.md | 66 +++++++ ...-checkpoint-restore-active-run-timeline.md | 85 ++++++++ coding-agent/postmortem/index.yaml | 28 +++ ...ntime-projections-on-checkpoint-restore.md | 45 +++++ coding-agent/src/coding_agent/acp/server.py | 11 +- .../src/coding_agent/cli/local_runtime.py | 5 + coding-agent/src/coding_agent/runs/query.py | 14 +- .../src/coding_agent/server/http_server.py | 4 +- .../coding_agent/server/session_manager.py | 25 ++- .../src/coding_agent/stores/durable_local.py | 23 ++- .../src/coding_agent/stores/durable_pg.py | 25 ++- .../src/coding_agent/stores/runtime_store.py | 98 ++++++++- coding-agent/tests/acp/test_compat_harness.py | 6 + coding-agent/tests/acp/test_server.py | 6 +- coding-agent/tests/cli/test_local_runtime.py | 16 ++ .../coding_agent/test_jsonl_runtime_store.py | 5 + .../coding_agent/test_pg_durable_fencing.py | 60 ++++++ .../coding_agent/test_pg_runtime_store.py | 33 ++++ .../test_runtime_query_service.py | 40 ++++ .../test_sqlite_local_durable_fencing.py | 47 ++++- .../coding_agent/test_sqlite_runtime_store.py | 59 ++++++ coding-agent/tests/ui/test_http_server.py | 186 ++++++++++++++++++ coding-agent/webui/app/src/App.test.tsx | 84 ++++++++ 23 files changed, 943 insertions(+), 28 deletions(-) create mode 100644 coding-agent/.opencode/prompts/tasks/checkpoint-active-run-timeline.md create mode 100644 coding-agent/docs/adr/0075-checkpoint-restore-active-run-timeline.md create mode 100644 coding-agent/postmortem/patterns/PM-0027-reconcile-runtime-projections-on-checkpoint-restore.md 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..aa7edda3e --- /dev/null +++ b/coding-agent/.opencode/prompts/tasks/checkpoint-active-run-timeline.md @@ -0,0 +1,66 @@ +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. +- 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 worker recovery, 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/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..02e96ad70 --- /dev/null +++ b/coding-agent/docs/adr/0075-checkpoint-restore-active-run-timeline.md @@ -0,0 +1,85 @@ +# 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. + +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. + +## 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. +- [ ] `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..4e9dc18fd --- /dev/null +++ b/coding-agent/postmortem/patterns/PM-0027-reconcile-runtime-projections-on-checkpoint-restore.md @@ -0,0 +1,45 @@ +--- +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. +--- + +# 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. 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..80ee41f17 100644 --- a/coding-agent/src/coding_agent/server/session_manager.py +++ b/coding-agent/src/coding_agent/server/session_manager.py @@ -1112,12 +1112,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 +1677,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 +2949,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, diff --git a/coding-agent/src/coding_agent/stores/durable_local.py b/coding-agent/src/coding_agent/stores/durable_local.py index 0e6bfe9ed..c369ed107 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, @@ -1291,6 +1292,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..4b6465dce 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, @@ -916,6 +947,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 +956,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 +1028,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, @@ -1163,6 +1205,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 +1528,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 +1550,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 +1673,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 +1778,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 +1826,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..40e474513 100644 --- a/coding-agent/tests/coding_agent/test_pg_durable_fencing.py +++ b/coding-agent/tests/coding_agent/test_pg_durable_fencing.py @@ -53,6 +53,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 +78,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]) @@ -669,6 +683,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 +856,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..9d94353f2 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 @@ -305,6 +309,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, @@ -607,3 +638,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..433e06419 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, @@ -984,6 +988,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 +1257,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..26ee7cd92 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 @@ -148,3 +149,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/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"); From f02cbce32e60816520006fc7c0ff0d7b991e8e0d Mon Sep 17 00:00:00 2001 From: kina0630 Date: Wed, 5 Aug 2026 10:50:09 +0800 Subject: [PATCH 2/2] fix(checkpoint): close restore control gaps Task packet: coding-agent/.opencode/prompts/tasks/checkpoint-active-run-timeline.md --- .../tasks/checkpoint-active-run-timeline.md | 7 ++- ...-checkpoint-restore-active-run-timeline.md | 11 ++++ ...ntime-projections-on-checkpoint-restore.md | 4 ++ .../coding_agent/server/session_manager.py | 26 ++++++++ .../src/coding_agent/stores/durable_local.py | 1 + .../src/coding_agent/stores/runtime_store.py | 2 + .../coding_agent/test_pg_durable_fencing.py | 48 +++++++++++++++ .../coding_agent/test_pg_runtime_store.py | 48 +++++++++++++++ .../test_sqlite_local_durable_fencing.py | 51 ++++++++++++++++ .../coding_agent/test_sqlite_runtime_store.py | 44 ++++++++++++++ .../tests/ui/test_session_manager_runtime.py | 59 +++++++++++++++++++ 11 files changed, 299 insertions(+), 2 deletions(-) diff --git a/coding-agent/.opencode/prompts/tasks/checkpoint-active-run-timeline.md b/coding-agent/.opencode/prompts/tasks/checkpoint-active-run-timeline.md index aa7edda3e..b901fbc51 100644 --- a/coding-agent/.opencode/prompts/tasks/checkpoint-active-run-timeline.md +++ b/coding-agent/.opencode/prompts/tasks/checkpoint-active-run-timeline.md @@ -9,6 +9,8 @@ Scope: 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/`. @@ -16,8 +18,8 @@ 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 worker recovery, developer-console audit, or direct run-id/event - endpoints to hide superseded records. +- Changing developer-console audit or direct run-id/event endpoints to hide + superseded records. - Redesigning checkpoint snapshot payloads or restore authorization. Context: @@ -47,6 +49,7 @@ Context: 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` 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 index 02e96ad70..7f5806da7 100644 --- a/coding-agent/docs/adr/0075-checkpoint-restore-active-run-timeline.md +++ b/coding-agent/docs/adr/0075-checkpoint-restore-active-run-timeline.md @@ -30,6 +30,11 @@ and PostgreSQL restore, mark previously active runs for the same session whose 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 @@ -56,6 +61,10 @@ durable stores. - 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 @@ -67,6 +76,8 @@ durable stores. - [ ] 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` 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 index 4e9dc18fd..975786e1b 100644 --- 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 @@ -23,6 +23,8 @@ release_checks: - 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 @@ -43,3 +45,5 @@ apply a permanent timestamp filter that would hide new runs after restore. - 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/server/session_manager.py b/coding-agent/src/coding_agent/server/session_manager.py index 80ee41f17..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 @@ -2966,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 c369ed107..d182065e3 100644 --- a/coding-agent/src/coding_agent/stores/durable_local.py +++ b/coding-agent/src/coding_agent/stores/durable_local.py @@ -1061,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 diff --git a/coding-agent/src/coding_agent/stores/runtime_store.py b/coding-agent/src/coding_agent/stores/runtime_store.py index 4b6465dce..6cd68cbd1 100644 --- a/coding-agent/src/coding_agent/stores/runtime_store.py +++ b/coding-agent/src/coding_agent/stores/runtime_store.py @@ -675,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 @@ -1073,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 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 40e474513..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 @@ -381,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() 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 9d94353f2..11752771b 100644 --- a/coding-agent/tests/coding_agent/test_pg_runtime_store.py +++ b/coding-agent/tests/coding_agent/test_pg_runtime_store.py @@ -80,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, @@ -417,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, 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 433e06419..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 @@ -58,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, 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 26ee7cd92..729e0539e 100644 --- a/coding-agent/tests/coding_agent/test_sqlite_runtime_store.py +++ b/coding-agent/tests/coding_agent/test_sqlite_runtime_store.py @@ -14,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, 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: