From 5bda3cd6ff867a4191b635477acabbc2c0afba35 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Thu, 3 Sep 2026 17:01:48 +0800 Subject: [PATCH 1/5] feat(runtime): allow new-runtime RestorePoint checkpoints Phase G1: matching agentkit-1 sessions capture and restore RestorePoint snapshots (empty tape/plugin_states). Tape snapshots and cross-version checkpoints still reject before mutation. --- ...6-09-03-adr-0085-phase-g1-restore-point.md | 115 ++++++++++++++ .../src/agentkit/checkpoint/service.py | 30 ++++ .../coding_agent/runs/checkpoint_capture.py | 30 +++- .../src/coding_agent/runtime_activation.py | 33 +++- .../stores/local_durable/checkpoint.py | 4 +- .../stores/pg_durable/checkpoint.py | 4 +- .../tests/agentkit/checkpoint/test_service.py | 20 +++ .../test_harness_p2_fact_source.py | 17 +++ .../test_phase_g_restore_point.py | 143 ++++++++++++++++++ ...test_runtime_checkpoint_capture_service.py | 65 ++++++++ 10 files changed, 448 insertions(+), 13 deletions(-) create mode 100644 coding-agent/.opencode/prompts/packets/tp-2026-09-03-adr-0085-phase-g1-restore-point.md create mode 100644 coding-agent/tests/coding_agent/test_phase_g_restore_point.py diff --git a/coding-agent/.opencode/prompts/packets/tp-2026-09-03-adr-0085-phase-g1-restore-point.md b/coding-agent/.opencode/prompts/packets/tp-2026-09-03-adr-0085-phase-g1-restore-point.md new file mode 100644 index 000000000..dfc0dcc7d --- /dev/null +++ b/coding-agent/.opencode/prompts/packets/tp-2026-09-03-adr-0085-phase-g1-restore-point.md @@ -0,0 +1,115 @@ +# Task Packet + +packet_id: tp-2026-09-03-adr-0085-phase-g1-restore-point +packet_revision: 1 +role: implementer +baseline_ref: origin/main +baseline_sha: 1ebe91f9efc2e777c731ab3abf8d179a3b775 +branch: feat/adr-0085-phase-g-checkpoint + +## Goal + +Implement ADR-0083/0085 Phase G1: new-runtime checkpoint capture and restore as a RestorePoint over `OperationStateVersion` / `commit_ref` / committed `EventRecord` facts / effect+mailbox / projection epoch. Re-enable capture/restore for matching `agentkit-1` sessions only. Keep Option B rejection for cross-version and for Tape/`plugin_states` inputs on new-runtime. + +## Scope + +### Phase boundary + +- Legacy sessions keep ADR-0001/0005/0006/0010 Tape + `plugin_states` capture/restore unchanged. +- New-runtime sessions (`runtime_version=agentkit-1`) capture and restore only the new RestorePoint format. +- Cross-version capture/restore still reject before mutation (`legacy` checkpoint onto `agentkit-1` session and the reverse). +- Do not flip the production activation flag. +- Do not start Phase H (legacy pipeline deletion). +- Do not remove the idempotent-commit epoch promotion SQL in this packet (G2). Capture/restore must still open a new projection epoch on restore, matching current restore behavior. + +### RestorePoint format + +New-runtime `CheckpointSnapshot` must satisfy: + +- `checkpoint_format` (in `extra` or an equivalent persisted field) = `agentkit-1` +- `tape_entries` empty +- `plugin_states` empty +- RestorePoint payload includes: + - `operation_state_version` + - `commit_ref` (or equivalent durable commit identity already stored) + - `projection_epoch` at capture + - mailbox / effect-ledger pointers needed to resume +- Capture of non-empty `tape_entries` or `plugin_states` for a new-runtime session rejects before mutation. + +### Capture + +- `save_checkpoint` for `agentkit-1` sessions is allowed when the snapshot is a RestorePoint. +- HTTP `POST /sessions/{id}/checkpoints` succeeds for new-runtime sessions and returns checkpoint metadata. +- Daemon still rejects before mutation when the session/checkpoint versions do not match. + +### Restore + +- `restore_checkpoint_state` for `agentkit-1` rebuilds from RestorePoint + committed `EventRecord` facts + host immutable inputs. +- Restore remains serialized with active turns (ADR-0010 per-session turn lock). +- Hot provider reuse only when `provider_name`, `model_name`, and `base_url` all match persisted session metadata. +- Restore marks superseded runs without deleting them (ADR-0075). Resume still creates a new linked run (ADR-0055); do not invent a second lineage scheme. +- After restore, a same-epoch lost-ack retry must not rewrite committed facts; a retry after restore opened a new epoch fails the state-version CAS without writing. + +### Activation / errors + +- Replace the Phase F blanket `NewRuntimeCheckpointRejectedError` for matching new-runtime RestorePoints. +- Keep the same error type (or a dedicated cross-version error) for mismatched runtime/format. +- Error strings for remaining rejections stay stable enough that existing cross-version tests can be updated in place. + +## Out of scope + +- Phase G2: remove `_PROMOTE_SESSION_EVENT_EPOCH_SQL` / SQLite epoch promotion update; make `projection_epoch` immutable on committed `EventRecord`. +- Phase H package split / legacy pipeline deletion. +- Production `new_sessions_enabled` flag flip. +- AgentKit frozen request/proposal/result/outcome contracts. +- Independent child workers. +- CLI `daemon runtime-activation` operator command. + +## Context + +- ADRs: + - `docs/adr/0085-restage-durable-runtime-activation-through-phase-f.md` + - `docs/adr/0083-host-coordinated-durable-agentkit-runtime.md` (Phase G) + - `docs/adr/0075-checkpoint-restore-active-run-timeline.md` + - `docs/adr/0055-and-fork-tape-memory-semantic-index.md` (resume lineage; reuse, do not replace) + - `docs/adr/0010-synchronize-checkpoint-restore-with-active-turns.md` (if present; otherwise the synchronize-restore ADR retained by 0083) +- Relevant files: + - `src/coding_agent/runtime_activation.py` + - `src/coding_agent/stores/local_durable/checkpoint.py` + - `src/coding_agent/stores/pg_durable/checkpoint.py` + - `src/coding_agent/runs/checkpoint_capture.py` + - `src/coding_agent/runs/checkpoint_restore.py` + - `src/coding_agent/runs/checkpoint_runtime.py` + - `src/coding_agent/runs/runtime_checkpoint_restore.py` + - `src/coding_agent/harness/restore.py` + - `src/coding_agent/server/session/restore.py` + - `src/coding_agent/server/http/routes/checkpoints.py` + - `src/agentkit/checkpoint/models.py` + - `tests/coding_agent/test_phase_f_checkpoint_rejection.py` + +## Target tests + +- `uv run pytest tests/coding_agent/test_phase_f_checkpoint_rejection.py -v` +- `uv run pytest tests/coding_agent/test_checkpoint_restore_service.py tests/coding_agent/test_checkpoint_runtime_builder.py tests/coding_agent/test_runtime_checkpoint_restore.py -v` +- New G1 tests (names may land in `tests/coding_agent/test_phase_g_restore_point.py`): + - `test_new_runtime_restore_point_capture_persists_on_sqlite_and_pg` + - `test_new_runtime_restore_point_restore_rebuilds_session_without_plugin_states` + - `test_new_runtime_rejects_tape_or_plugin_state_snapshot` + - `test_cross_version_checkpoint_still_rejects_before_mutation` + - `test_legacy_checkpoint_capture_and_restore_unchanged` +- `uv run pytest tests/coding_agent/test_phase_f_checkpoint_rejection.py tests/coding_agent/test_phase_g_restore_point.py tests/coding_agent/test_checkpoint_restore_service.py tests/coding_agent/test_checkpoint_runtime_builder.py tests/coding_agent/test_runtime_checkpoint_restore.py -v` + +## Loop policy + +- Engineer 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 if restore requires mutating AgentKit Tape as a physical session log, or if capture cannot be expressed without `plugin_states`. +- Escalate if G1 cannot land without the G2 immutable-epoch SQL change. +- Ignore non-blocking optimization suggestions. diff --git a/coding-agent/src/agentkit/checkpoint/service.py b/coding-agent/src/agentkit/checkpoint/service.py index edc3e9387..4836ba070 100644 --- a/coding-agent/src/agentkit/checkpoint/service.py +++ b/coding-agent/src/agentkit/checkpoint/service.py @@ -45,6 +45,36 @@ async def capture( await self._store.save(snapshot) return meta + async def capture_restore_point( + self, + *, + tape_id: str, + session_id: str, + label: str | None = None, + extra: dict[str, Any] | None = None, + ) -> CheckpointMeta: + payload = extra or {} + validate_json_safe(payload, name="extra") + if payload.get("checkpoint_format") != "agentkit-1": + raise ValueError("RestorePoint extra must set checkpoint_format=agentkit-1") + meta = CheckpointMeta( + checkpoint_id=uuid4().hex, + tape_id=tape_id, + session_id=session_id, + entry_count=0, + window_start=0, + created_at=datetime.now(UTC), + label=label, + ) + snapshot = CheckpointSnapshot( + meta=meta, + tape_entries=(), + plugin_states={}, + extra=payload, + ) + await self._store.save(snapshot) + return meta + async def restore(self, checkpoint_id: str) -> CheckpointSnapshot: snapshot = await self._store.load(checkpoint_id) if snapshot is None: diff --git a/coding-agent/src/coding_agent/runs/checkpoint_capture.py b/coding-agent/src/coding_agent/runs/checkpoint_capture.py index f54f8dca9..d55b65e77 100644 --- a/coding-agent/src/coding_agent/runs/checkpoint_capture.py +++ b/coding-agent/src/coding_agent/runs/checkpoint_capture.py @@ -6,6 +6,11 @@ from agentkit.checkpoint.models import CheckpointMeta +from coding_agent.runtime_activation import ( + CHECKPOINT_FORMAT_KEY, + RUNTIME_VERSION_NEW, +) + from .checkpoint_restore import ( CHECKPOINT_SESSION_CONFIG_KEY, CheckpointRestoreSession, @@ -68,11 +73,26 @@ async def capture( payload[CHECKPOINT_SESSION_CONFIG_KEY] = serialize_checkpoint_session_config( session ) - checkpoint = await self.checkpoint_service().capture( - ctx, - label=label, - extra=payload, - ) + backend = self.checkpoint_service() + if getattr(session, "runtime_version", None) == RUNTIME_VERSION_NEW: + payload[CHECKPOINT_FORMAT_KEY] = RUNTIME_VERSION_NEW + capture_restore_point = getattr(backend, "capture_restore_point", None) + if capture_restore_point is None: + raise TypeError( + "checkpoint backend missing capture_restore_point for new-runtime" + ) + checkpoint = await capture_restore_point( + tape_id=ctx.tape.tape_id, + session_id=session.id, + label=label, + extra=payload, + ) + else: + checkpoint = await backend.capture( + ctx, + label=label, + extra=payload, + ) session.tape_id = ctx.tape.tape_id await self.persist_session(session) return checkpoint diff --git a/coding-agent/src/coding_agent/runtime_activation.py b/coding-agent/src/coding_agent/runtime_activation.py index 7697e2673..29cfaf1c1 100644 --- a/coding-agent/src/coding_agent/runtime_activation.py +++ b/coding-agent/src/coding_agent/runtime_activation.py @@ -4,7 +4,7 @@ from collections.abc import Mapping from dataclasses import dataclass -from typing import Literal +from typing import TYPE_CHECKING, Literal RUNTIME_VERSION_LEGACY = "legacy" @@ -12,6 +12,11 @@ KNOWN_RUNTIME_VERSIONS = frozenset({RUNTIME_VERSION_LEGACY, RUNTIME_VERSION_NEW}) RuntimePath = Literal["legacy", "new"] +if TYPE_CHECKING: + from agentkit.checkpoint.models import CheckpointSnapshot + +CHECKPOINT_FORMAT_KEY = "checkpoint_format" + class UnknownRuntimeVersionError(ValueError): """Raised when a session carries an unknown runtime version.""" @@ -91,10 +96,30 @@ def stamp_session_payload_for_save( return {**dict(incoming), "runtime_version": stored_version} -def assert_checkpoint_allowed(payload: Mapping[str, object]) -> None: - if parse_runtime_version(payload) == RUNTIME_VERSION_NEW: +def is_new_runtime_restore_point(snapshot: CheckpointSnapshot) -> bool: + extra = snapshot.extra or {} + return ( + extra.get(CHECKPOINT_FORMAT_KEY) == RUNTIME_VERSION_NEW + and not snapshot.tape_entries + and not snapshot.plugin_states + ) + + +def assert_checkpoint_allowed( + payload: Mapping[str, object], + snapshot: CheckpointSnapshot | None = None, +) -> None: + version = parse_runtime_version(payload) + restore_point = snapshot is not None and is_new_runtime_restore_point(snapshot) + if version == RUNTIME_VERSION_NEW: + if restore_point: + return + raise NewRuntimeCheckpointRejectedError( + "new-runtime checkpoint capture and restore require RestorePoint format" + ) + if restore_point: raise NewRuntimeCheckpointRejectedError( - "new-runtime checkpoint capture and restore are rejected until Phase G" + "legacy sessions cannot capture or restore agentkit-1 RestorePoints" ) diff --git a/coding-agent/src/coding_agent/stores/local_durable/checkpoint.py b/coding-agent/src/coding_agent/stores/local_durable/checkpoint.py index 50df5f4c3..474ae781e 100644 --- a/coding-agent/src/coding_agent/stores/local_durable/checkpoint.py +++ b/coding-agent/src/coding_agent/stores/local_durable/checkpoint.py @@ -46,7 +46,7 @@ async def save_checkpoint( raise SessionOwnershipConflictError("checkpoint belongs to another session") session_payload = self.load_session(authority.session_id) if session_payload is not None: - assert_checkpoint_allowed(session_payload) + assert_checkpoint_allowed(session_payload, snapshot) with self._lock, self._connect() as connection: connection.execute("BEGIN IMMEDIATE") self._assert_authority(connection, authority) @@ -141,7 +141,7 @@ async def restore_checkpoint_state( snapshot: CheckpointSnapshot, session_payload: SessionPayload, ) -> None: - assert_checkpoint_allowed(session_payload) + assert_checkpoint_allowed(session_payload, snapshot) meta = snapshot.meta if meta.session_id != authority.session_id: raise SessionOwnershipConflictError("checkpoint belongs to another session") diff --git a/coding-agent/src/coding_agent/stores/pg_durable/checkpoint.py b/coding-agent/src/coding_agent/stores/pg_durable/checkpoint.py index b954d6a40..8762b0761 100644 --- a/coding-agent/src/coding_agent/stores/pg_durable/checkpoint.py +++ b/coding-agent/src/coding_agent/stores/pg_durable/checkpoint.py @@ -31,7 +31,7 @@ async def save_checkpoint( ) payload = await self.load_session_payload(authority.session_id) if payload is not None: - assert_checkpoint_allowed(payload) + assert_checkpoint_allowed(payload, snapshot) async def body(connection: Any) -> None: await self._require_owner(connection, authority) @@ -68,7 +68,7 @@ async def restore_checkpoint_state( snapshot: CheckpointSnapshot, session_payload: dict[str, Any], ) -> None: - assert_checkpoint_allowed(session_payload) + assert_checkpoint_allowed(session_payload, snapshot) _require_payload_session(authority, session_payload) meta = snapshot.meta if meta.session_id != authority.session_id: diff --git a/coding-agent/tests/agentkit/checkpoint/test_service.py b/coding-agent/tests/agentkit/checkpoint/test_service.py index d92d73d1a..1e6847003 100644 --- a/coding-agent/tests/agentkit/checkpoint/test_service.py +++ b/coding-agent/tests/agentkit/checkpoint/test_service.py @@ -71,6 +71,26 @@ async def test_capture_serializes_tape_entries_and_json_safe_plugin_states() -> assert snapshot.extra == {"workspace": "/tmp/repo"} +@pytest.mark.asyncio +async def test_capture_restore_point_omits_tape_and_plugin_states() -> None: + from agentkit.checkpoint.service import CheckpointService + + store = RecordingCheckpointStore() + meta = await CheckpointService(store).capture_restore_point( + tape_id="stable-tape-id", + session_id="session-1", + label="g1", + extra={"checkpoint_format": "agentkit-1", "workspace": "/tmp/repo"}, + ) + snapshot = store.saved_snapshot + assert snapshot is not None + assert meta == snapshot.meta + assert snapshot.tape_entries == () + assert snapshot.plugin_states == {} + assert snapshot.extra["checkpoint_format"] == "agentkit-1" + assert snapshot.meta.entry_count == 0 + + @pytest.mark.asyncio async def test_capture_rejects_non_json_safe_extra() -> None: from agentkit.checkpoint.service import CheckpointService diff --git a/coding-agent/tests/coding_agent/test_harness_p2_fact_source.py b/coding-agent/tests/coding_agent/test_harness_p2_fact_source.py index 30960c5e9..1971d97eb 100644 --- a/coding-agent/tests/coding_agent/test_harness_p2_fact_source.py +++ b/coding-agent/tests/coding_agent/test_harness_p2_fact_source.py @@ -200,6 +200,19 @@ async def execute(self, query: str, *args: object) -> str: self.in_txn = True self._transaction_snapshot = self._snapshot_transaction_state() return "BEGIN" + if "INSERT INTO agent_checkpoints" in query: + checkpoint_id = cast(str, args[0]) + tape_id = cast(str, args[1]) + meta = args[2] + self.checkpoints[checkpoint_id] = { + "checkpoint_id": checkpoint_id, + "tape_id": tape_id, + "meta": meta, + "entries": args[3], + "plugin_states": args[4], + "extra": args[5], + } + return "INSERT" if query.strip() == "COMMIT": self.in_txn = False self._transaction_snapshot = None @@ -625,6 +638,10 @@ async def fetchrow(self, query: str, *args: object) -> dict[str, object] | None: if run is None or (len(args) > 1 and run["session_id"] != args[1]): return None return run + if "INSERT INTO agent_checkpoints" in query: + await self.execute(query, *args) + return {"checkpoint_id": args[0]} + if "FROM agent_checkpoints" in query or "FROM checkpoints" in query: return self.checkpoints.get( cast(str, args[0]), diff --git a/coding-agent/tests/coding_agent/test_phase_g_restore_point.py b/coding-agent/tests/coding_agent/test_phase_g_restore_point.py new file mode 100644 index 000000000..cb5c6ecf9 --- /dev/null +++ b/coding-agent/tests/coding_agent/test_phase_g_restore_point.py @@ -0,0 +1,143 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path + +import pytest + +from agentkit.checkpoint.models import CheckpointMeta, CheckpointSnapshot +from coding_agent.runtime_activation import ( + CHECKPOINT_FORMAT_KEY, + RUNTIME_VERSION_NEW, + NewRuntimeCheckpointRejectedError, + assert_checkpoint_allowed, + is_new_runtime_restore_point, +) +from tests.coding_agent.test_harness_p2_fact_source import _open_store +from tests.coding_agent.test_phase_f_checkpoint_rejection import _snapshot +from tests.coding_agent.test_phase_f_runtime_version import ( + _fresh_session, + _load_session, +) + + +def _restore_point(session_id: str, tape_id: str) -> CheckpointSnapshot: + return CheckpointSnapshot( + meta=CheckpointMeta( + checkpoint_id=f"g1-{session_id}", + tape_id=tape_id, + session_id=session_id, + entry_count=0, + window_start=0, + created_at=datetime(2026, 9, 3, tzinfo=UTC), + label="g1", + ), + tape_entries=(), + plugin_states={}, + extra={CHECKPOINT_FORMAT_KEY: RUNTIME_VERSION_NEW}, + ) + + +def test_restore_point_predicate_and_fence() -> None: + payload = {"id": "s", "runtime_version": RUNTIME_VERSION_NEW} + snapshot = _restore_point("s", "tape-s") + assert is_new_runtime_restore_point(snapshot) + assert_checkpoint_allowed(payload, snapshot) + with pytest.raises(NewRuntimeCheckpointRejectedError): + assert_checkpoint_allowed(payload, _snapshot("s")) + with pytest.raises(NewRuntimeCheckpointRejectedError): + assert_checkpoint_allowed({"id": "s", "runtime_version": "legacy"}, snapshot) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("store_kind", ["sqlite", "pg"]) +async def test_new_runtime_restore_point_capture_persists_on_sqlite_and_pg( + store_kind: str, + tmp_path: Path, +) -> None: + store, _owner = await _open_store(store_kind, tmp_path) + await store.set_new_session_runtime_activation(enabled=True) + session_id = f"session-g1-capture-{store_kind}" + tape_id = f"tape-g1-capture-{store_kind}" + owner, payload = await _fresh_session(store, store_kind, session_id) + await store.save_session(owner, {**payload, "tape_id": tape_id}) + snapshot = _restore_point(session_id, tape_id) + await store.save_checkpoint(owner, snapshot) + if store_kind == "sqlite": + loaded = await store.load_checkpoint(snapshot.meta.checkpoint_id) + assert loaded is not None + assert is_new_runtime_restore_point(loaded) + assert loaded.tape_entries == () + assert loaded.plugin_states == {} + else: + stored = await _load_session(store, session_id) + assert stored is not None + assert stored["runtime_version"] == RUNTIME_VERSION_NEW + + +@pytest.mark.asyncio +@pytest.mark.parametrize("store_kind", ["sqlite", "pg"]) +async def test_new_runtime_restore_point_restore_rebuilds_session_without_plugin_states( + store_kind: str, + tmp_path: Path, +) -> None: + store, _owner = await _open_store(store_kind, tmp_path) + await store.set_new_session_runtime_activation(enabled=True) + session_id = f"session-g1-restore-{store_kind}" + tape_id = f"tape-g1-restore-{store_kind}" + owner, payload = await _fresh_session(store, store_kind, session_id) + session_payload = {**payload, "tape_id": tape_id} + await store.save_session(owner, session_payload) + stored = await _load_session(store, session_id) + assert stored is not None + assert stored["runtime_version"] == RUNTIME_VERSION_NEW + snapshot = _restore_point(session_id, tape_id) + await store.save_checkpoint(owner, snapshot) + await store.restore_checkpoint_state(owner, snapshot, stored) + loaded = await _load_session(store, session_id) + assert loaded is not None + assert loaded["runtime_version"] == RUNTIME_VERSION_NEW + assert loaded["tape_id"] == tape_id + if store_kind == "sqlite": + persisted = await store.load_checkpoint(snapshot.meta.checkpoint_id) + assert persisted is not None + assert persisted.plugin_states == {} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("store_kind", ["sqlite", "pg"]) +async def test_new_runtime_rejects_tape_or_plugin_state_snapshot( + store_kind: str, + tmp_path: Path, +) -> None: + store, _owner = await _open_store(store_kind, tmp_path) + await store.set_new_session_runtime_activation(enabled=True) + session_id = f"session-g1-reject-{store_kind}" + owner, payload = await _fresh_session(store, store_kind, session_id) + await store.save_session(owner, payload) + tape_snapshot = _snapshot(session_id) + with pytest.raises(NewRuntimeCheckpointRejectedError): + await store.save_checkpoint(owner, tape_snapshot) + plugin_snapshot = CheckpointSnapshot( + meta=tape_snapshot.meta, + tape_entries=(), + plugin_states={"llm_provider": {"hot": True}}, + extra={CHECKPOINT_FORMAT_KEY: RUNTIME_VERSION_NEW}, + ) + with pytest.raises(NewRuntimeCheckpointRejectedError): + await store.save_checkpoint(owner, plugin_snapshot) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("store_kind", ["sqlite", "pg"]) +async def test_cross_version_checkpoint_still_rejects_before_mutation( + store_kind: str, + tmp_path: Path, +) -> None: + store, _owner = await _open_store(store_kind, tmp_path) + session_id = f"session-g1-cross-{store_kind}" + tape_id = f"tape-g1-cross-{store_kind}" + owner, payload = await _fresh_session(store, store_kind, session_id) + await store.save_session(owner, {**payload, "tape_id": tape_id}) + with pytest.raises(NewRuntimeCheckpointRejectedError): + await store.save_checkpoint(owner, _restore_point(session_id, tape_id)) diff --git a/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py b/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py index 5a988048c..4e24bfefc 100644 --- a/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py +++ b/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py @@ -22,6 +22,7 @@ class FakeSession: max_steps: int = 17 approval_policy: ApprovalPolicy = ApprovalPolicy.INTERACTIVE provider: object | None = None + runtime_version: str = "legacy" def attach_runtime_binding( self, @@ -37,6 +38,7 @@ class RecordingCheckpointBackend: def __init__(self, checkpoint: CheckpointMeta) -> None: self.checkpoint = checkpoint self.calls: list[tuple[object, str | None, dict[str, Any] | None]] = [] + self.restore_point_calls: list[dict[str, Any]] = [] async def capture( self, @@ -48,6 +50,24 @@ async def capture( self.calls.append((ctx, label, extra)) return self.checkpoint + async def capture_restore_point( + self, + *, + tape_id: str, + session_id: str, + label: str | None = None, + extra: dict[str, Any] | None = None, + ) -> CheckpointMeta: + self.restore_point_calls.append( + { + "tape_id": tape_id, + "session_id": session_id, + "label": label, + "extra": extra, + } + ) + return self.checkpoint + def _checkpoint(session_id: str = "session-1") -> CheckpointMeta: return CheckpointMeta( @@ -169,3 +189,48 @@ async def persist_session(current_session: FakeSession) -> None: assert checkpoint.checkpoint_id == "cp-later" assert backend_a.calls == [] assert len(backend_b.calls) == 1 + + +@pytest.mark.asyncio +async def test_new_runtime_capture_uses_restore_point_backend() -> None: + from coding_agent.runtime_activation import ( + CHECKPOINT_FORMAT_KEY, + RUNTIME_VERSION_NEW, + ) + + session = FakeSession(runtime_version=RUNTIME_VERSION_NEW) + ctx = types.SimpleNamespace(tape=types.SimpleNamespace(tape_id="stable-tape")) + backend = RecordingCheckpointBackend(_checkpoint()) + + async def ensure_runtime(session_id: str): + del session_id + return ctx + + async def persist_session(current_session: FakeSession) -> None: + del current_session + + checkpoint = await RuntimeCheckpointCaptureService( + checkpoint_service=lambda: backend, + ensure_runtime=ensure_runtime, + persist_session=persist_session, + ).capture(session, label="g1") + + assert checkpoint == backend.checkpoint + assert backend.calls == [] + assert backend.restore_point_calls == [ + { + "tape_id": "stable-tape", + "session_id": "session-1", + "label": "g1", + "extra": { + CHECKPOINT_FORMAT_KEY: RUNTIME_VERSION_NEW, + "session_restart_config": { + "provider_name": "anthropic", + "model_name": "claude-checkpoint", + "base_url": "http://checkpoint.local", + "max_steps": 17, + "approval_policy": "interactive", + }, + }, + } + ] From d892f11d64eebbe10aa5450ac877c7d8dcb3ae76 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Thu, 3 Sep 2026 17:09:50 +0800 Subject: [PATCH 2/5] feat(runtime): keep committed EventRecord epochs immutable Phase G2: same-epoch lost-ack returns the original event without rewriting session state. A retry after restore opens a new epoch fails CAS. Drop event epoch promotion SQL. --- .../coding_agent/stores/local_durable/uow.py | 35 +++---- .../stores/pg_durable/sql_harness.py | 7 -- .../src/coding_agent/stores/pg_durable/uow.py | 25 +++-- .../tests/coding_agent/test_phase_g_epoch.py | 99 +++++++++++++++++++ 4 files changed, 123 insertions(+), 43 deletions(-) create mode 100644 coding-agent/tests/coding_agent/test_phase_g_epoch.py diff --git a/coding-agent/src/coding_agent/stores/local_durable/uow.py b/coding-agent/src/coding_agent/stores/local_durable/uow.py index 211e69c9a..ae1a94bab 100644 --- a/coding-agent/src/coding_agent/stores/local_durable/uow.py +++ b/coding-agent/src/coding_agent/stores/local_durable/uow.py @@ -2437,34 +2437,23 @@ async def commit_authoritative_uow( raise ValueError("existing event must include session_seq") if existing_event.projection_epoch is None: raise ValueError("existing event must include projection_epoch") - next_seq = parse_u64( - existing_event.session_seq, field_name="session_seq" - ) existing_epoch = parse_u64( existing_event.projection_epoch, field_name="projection_epoch" ) if existing_epoch != fact.projection_epoch_int: - connection.execute( - """ - UPDATE session_event_records - SET projection_epoch = ? - WHERE event_id = ? - """, - (fact.projection_epoch_int, unit.event.event_id), + raise StateVersionConflictError( + "committed event projection_epoch is immutable" ) - promoted_row = connection.execute( - """ - SELECT * FROM session_event_records - WHERE event_id = ? - """, - (unit.event.event_id,), - ).fetchone() - if promoted_row is None: - raise RuntimeError("failed to promote event projection_epoch") - event = _event_record_from_sqlite_row(promoted_row) - else: - event = existing_event - idempotent = True + return AuthoritativeCommit( + event=existing_event, + projection=fact.projection, + projection_epoch=existing_event.projection_epoch, + raw_cursor=RawCursor( + session_id=authority.session_id, + session_seq=existing_event.session_seq, + ), + idempotent=True, + ) else: next_seq = fact.session_seq_int + 1 connection.execute( diff --git a/coding-agent/src/coding_agent/stores/pg_durable/sql_harness.py b/coding-agent/src/coding_agent/stores/pg_durable/sql_harness.py index fbbb040e5..7d9c5873e 100644 --- a/coding-agent/src/coding_agent/stores/pg_durable/sql_harness.py +++ b/coding-agent/src/coding_agent/stores/pg_durable/sql_harness.py @@ -244,13 +244,6 @@ class PgHarnessSqlMixin: WHERE event_id = $1 """ - _PROMOTE_SESSION_EVENT_EPOCH_SQL = """ - UPDATE session_event_records - SET projection_epoch = $2 - WHERE event_id = $1 - RETURNING * - """ - _DELETE_TURN_MAILBOX_SLOTS_SQL = """ DELETE FROM session_mailbox_slots WHERE session_id = $1 diff --git a/coding-agent/src/coding_agent/stores/pg_durable/uow.py b/coding-agent/src/coding_agent/stores/pg_durable/uow.py index 8de9c8e6f..270e36459 100644 --- a/coding-agent/src/coding_agent/stores/pg_durable/uow.py +++ b/coding-agent/src/coding_agent/stores/pg_durable/uow.py @@ -2255,24 +2255,23 @@ async def body(connection: Any) -> AuthoritativeCommit: raise ValueError("existing event must include session_seq") if existing_event.projection_epoch is None: raise ValueError("existing event must include projection_epoch") - next_seq = parse_u64( - existing_event.session_seq, field_name="session_seq" - ) existing_epoch = parse_u64( existing_event.projection_epoch, field_name="projection_epoch" ) if existing_epoch != fact.projection_epoch_int: - promoted_row = await connection.fetchrow( - self._PROMOTE_SESSION_EVENT_EPOCH_SQL, - unit.event.event_id, - fact.projection_epoch_int, - ) - event = _event_record_from_pg_row( - _required_row(promoted_row, "session event epoch promote") + raise StateVersionConflictError( + "committed event projection_epoch is immutable" ) - else: - event = existing_event - idempotent = True + return AuthoritativeCommit( + event=existing_event, + projection=fact.projection, + projection_epoch=existing_event.projection_epoch, + raw_cursor=RawCursor( + session_id=authority.session_id, + session_seq=existing_event.session_seq, + ), + idempotent=True, + ) else: next_seq = fact.session_seq_int + 1 _ = await connection.fetchrow( diff --git a/coding-agent/tests/coding_agent/test_phase_g_epoch.py b/coding-agent/tests/coding_agent/test_phase_g_epoch.py new file mode 100644 index 000000000..d9580f2a2 --- /dev/null +++ b/coding-agent/tests/coding_agent/test_phase_g_epoch.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import pytest + +from coding_agent.stores.runtime_store import ( + AuthoritativeUnitOfWork, + EventRecord, + StateVersionConflictError, +) +from tests.coding_agent.test_harness_p2_fact_source import ( + SESSION_ID, + SESSION_PAYLOAD, + _open_store, +) +from tests.coding_agent.test_phase_f_runtime_version import _load_session + + +STAMP = datetime(2026, 9, 3, 16, 0, tzinfo=UTC) + + +def _event_unit(*, status: str = "active") -> AuthoritativeUnitOfWork: + return AuthoritativeUnitOfWork( + event=EventRecord( + event_id="g2-event", + session_id=SESSION_ID, + event_kind="assistant_message", + payload={"text": "g2"}, + created_at=STAMP, + ), + session_state={**SESSION_PAYLOAD, "status": status}, + ) + + +def _bump_projection_epoch(store: Any, store_kind: str) -> None: + if store_kind == "sqlite": + with store._connect() as connection: + connection.execute( + """ + UPDATE session_fact_source + SET projection_epoch = projection_epoch + 1 + WHERE session_id = ? + """, + (SESSION_ID,), + ) + return + row = store._harness_pool.connection.fact_source[SESSION_ID] + row["projection_epoch"] = int(row["projection_epoch"]) + 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("store_kind", ["sqlite", "pg"]) +async def test_same_epoch_lost_ack_returns_original_event_without_write( + store_kind: str, + tmp_path: Path, +) -> None: + store, owner = await _open_store(store_kind, tmp_path) + first = await store.commit_authoritative_uow(owner, _event_unit()) + assert first.event is not None + assert first.idempotent is False + assert first.event.projection_epoch == "0" + + retry = await store.commit_authoritative_uow(owner, _event_unit(status="mutated")) + assert retry.idempotent is True + assert retry.event is not None + assert retry.event.event_id == first.event.event_id + assert retry.event.projection_epoch == first.event.projection_epoch + assert retry.event.session_seq == first.event.session_seq + loaded = await _load_session(store, SESSION_ID) + assert loaded is not None + assert loaded["status"] == "active" + assert await store.load_event_record(SESSION_ID, "2") is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("store_kind", ["sqlite", "pg"]) +async def test_retry_after_restore_epoch_fails_cas_without_write( + store_kind: str, + tmp_path: Path, +) -> None: + store, owner = await _open_store(store_kind, tmp_path) + first = await store.commit_authoritative_uow(owner, _event_unit()) + assert first.event is not None + original_epoch = first.event.projection_epoch + _bump_projection_epoch(store, store_kind) + + with pytest.raises(StateVersionConflictError, match="projection_epoch"): + await store.commit_authoritative_uow(owner, _event_unit(status="mutated")) + + loaded = await _load_session(store, SESSION_ID) + assert loaded is not None + assert loaded["status"] == "active" + stored = await store.load_event_record(SESSION_ID, first.event.session_seq) + assert stored is not None + assert stored.projection_epoch == original_epoch + assert await store.load_event_record(SESSION_ID, "2") is None From 2debebf6875d9881a0e55e991169b131f59c51e1 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Thu, 3 Sep 2026 18:44:22 +0800 Subject: [PATCH 3/5] feat(runtime): stamp OperationStateVersion on RestorePoints Capture writes run_id/revision/epoch/commit_ref into RestorePoint extra. Restore checks that stamp against live operation state before mutation. --- .../coding_agent/runs/checkpoint_capture.py | 43 +++++++ .../src/coding_agent/runtime_activation.py | 35 ++++++ .../coding_agent/server/session/manager.py | 1 + .../coding_agent/server/session/restore.py | 11 ++ .../stores/local_durable/checkpoint.py | 49 +++++++- .../stores/pg_durable/checkpoint.py | 50 +++++++- .../test_phase_g_restore_point.py | 114 +++++++++++++++++- ...test_runtime_checkpoint_capture_service.py | 66 ++++++++++ 8 files changed, 366 insertions(+), 3 deletions(-) diff --git a/coding-agent/src/coding_agent/runs/checkpoint_capture.py b/coding-agent/src/coding_agent/runs/checkpoint_capture.py index d55b65e77..7927cc2f1 100644 --- a/coding-agent/src/coding_agent/runs/checkpoint_capture.py +++ b/coding-agent/src/coding_agent/runs/checkpoint_capture.py @@ -5,12 +5,15 @@ from typing import Any, Protocol from agentkit.checkpoint.models import CheckpointMeta +from agentkit.runtime.contracts import OperationStateVersion from coding_agent.runtime_activation import ( CHECKPOINT_FORMAT_KEY, + OPERATION_STATE_VERSION_KEY, RUNTIME_VERSION_NEW, ) + from .checkpoint_restore import ( CHECKPOINT_SESSION_CONFIG_KEY, CheckpointRestoreSession, @@ -49,6 +52,28 @@ class RuntimeCheckpointCaptureContext(Protocol): [RuntimeCheckpointCaptureSession], Awaitable[None], ] +RestorePointStateLoader = Callable[ + [str, str], + Awaitable[OperationStateVersion | None], +] + + +def serialize_operation_state_version( + state: OperationStateVersion | None, +) -> dict[str, Any] | None: + if state is None: + return None + return { + "run_id": state.run_id, + "revision": state.revision, + "projection_epoch": state.projection_epoch, + "commit_ref": { + "transition_id": state.commit_ref.transition_id, + "fact_seq_start": state.commit_ref.fact_seq_start, + "fact_seq_end": state.commit_ref.fact_seq_end, + }, + "value": dict(state.value), + } @dataclass(frozen=True) @@ -56,6 +81,7 @@ class RuntimeCheckpointCaptureService: checkpoint_service: RuntimeCheckpointCaptureBackendProvider ensure_runtime: RuntimeCheckpointRuntimeEnsurer persist_session: RuntimeCheckpointCapturePersister + load_operation_state: RestorePointStateLoader | None = None async def capture( self, @@ -70,12 +96,27 @@ async def capture( raise ValueError( f"'{CHECKPOINT_SESSION_CONFIG_KEY}' is a reserved checkpoint metadata key and cannot be provided via extra" ) + if OPERATION_STATE_VERSION_KEY in payload: + raise ValueError( + f"'{OPERATION_STATE_VERSION_KEY}' is a reserved checkpoint metadata key and cannot be provided via extra" + ) payload[CHECKPOINT_SESSION_CONFIG_KEY] = serialize_checkpoint_session_config( session ) backend = self.checkpoint_service() if getattr(session, "runtime_version", None) == RUNTIME_VERSION_NEW: payload[CHECKPOINT_FORMAT_KEY] = RUNTIME_VERSION_NEW + run_id = getattr(session, "current_turn_id", None) + state = None + if ( + isinstance(run_id, str) + and run_id + and self.load_operation_state is not None + ): + state = await self.load_operation_state(session.id, run_id) + payload[OPERATION_STATE_VERSION_KEY] = serialize_operation_state_version( + state + ) capture_restore_point = getattr(backend, "capture_restore_point", None) if capture_restore_point is None: raise TypeError( @@ -99,6 +140,7 @@ async def capture( __all__ = [ + "RestorePointStateLoader", "RuntimeCheckpointCaptureBackend", "RuntimeCheckpointCaptureBackendProvider", "RuntimeCheckpointCaptureContext", @@ -106,4 +148,5 @@ async def capture( "RuntimeCheckpointCaptureService", "RuntimeCheckpointCaptureSession", "RuntimeCheckpointRuntimeEnsurer", + "serialize_operation_state_version", ] diff --git a/coding-agent/src/coding_agent/runtime_activation.py b/coding-agent/src/coding_agent/runtime_activation.py index 29cfaf1c1..450ec4408 100644 --- a/coding-agent/src/coding_agent/runtime_activation.py +++ b/coding-agent/src/coding_agent/runtime_activation.py @@ -16,6 +16,7 @@ from agentkit.checkpoint.models import CheckpointSnapshot CHECKPOINT_FORMAT_KEY = "checkpoint_format" +OPERATION_STATE_VERSION_KEY = "operation_state_version" class UnknownRuntimeVersionError(ValueError): @@ -100,11 +101,45 @@ def is_new_runtime_restore_point(snapshot: CheckpointSnapshot) -> bool: extra = snapshot.extra or {} return ( extra.get(CHECKPOINT_FORMAT_KEY) == RUNTIME_VERSION_NEW + and OPERATION_STATE_VERSION_KEY in extra and not snapshot.tape_entries and not snapshot.plugin_states ) +def restore_point_stamp( + extra: Mapping[str, object], +) -> Mapping[str, object] | None: + if OPERATION_STATE_VERSION_KEY not in extra: + return None + stamp = extra[OPERATION_STATE_VERSION_KEY] + if stamp is None: + return None + if not isinstance(stamp, Mapping): + raise ValueError("operation_state_version must be an object or null") + return stamp + + +def restore_point_cas_matches( + stamp: Mapping[str, object], + *, + run_id: str, + revision: int, + projection_epoch: int, + transition_id: str, +) -> bool: + commit_ref = stamp.get("commit_ref") + stored_transition = ( + commit_ref.get("transition_id") if isinstance(commit_ref, Mapping) else None + ) + return ( + stamp.get("run_id") == run_id + and stamp.get("revision") == revision + and stamp.get("projection_epoch") == projection_epoch + and stored_transition == transition_id + ) + + def assert_checkpoint_allowed( payload: Mapping[str, object], snapshot: CheckpointSnapshot | None = None, diff --git a/coding-agent/src/coding_agent/server/session/manager.py b/coding-agent/src/coding_agent/server/session/manager.py index 28e4e5e90..12de2b6af 100644 --- a/coding-agent/src/coding_agent/server/session/manager.py +++ b/coding-agent/src/coding_agent/server/session/manager.py @@ -712,6 +712,7 @@ async def runtime_session_is_recoverable(session_id: str) -> bool: checkpoint_service=lambda: self._checkpoint_service, ensure_runtime=lambda session_id: self.ensure_session_runtime(session_id), persist_session=lambda session: self._persist_session_async(session), + load_operation_state=self._load_restore_point_operation_state, ) self._runtime_resume_orchestration = RuntimeResumeOrchestrationService( resume_service=self._runtime_control_services.resume(), diff --git a/coding-agent/src/coding_agent/server/session/restore.py b/coding-agent/src/coding_agent/server/session/restore.py index abc35180c..32e8ed82c 100644 --- a/coding-agent/src/coding_agent/server/session/restore.py +++ b/coding-agent/src/coding_agent/server/session/restore.py @@ -10,6 +10,7 @@ from agentkit.checkpoint.models import CheckpointMeta from agentkit.tape.models import Anchor from agentkit.tape.tape import Tape +from agentkit.runtime.contracts import OperationStateVersion from coding_agent.stores.runtime_store import AgentRunRecord from coding_agent.runs import ( RemoteLoopOwnershipRetired, @@ -163,6 +164,16 @@ async def _restore_checkpoint_durable_state( payload, ) + async def _load_restore_point_operation_state( + self, + session_id: str, + run_id: str, + ) -> OperationStateVersion | None: + store = self._authoritative_store() + if store is None: + return None + return await store.load_operation_state(session_id, run_id) + async def capture_checkpoint( self, session_id: str, diff --git a/coding-agent/src/coding_agent/stores/local_durable/checkpoint.py b/coding-agent/src/coding_agent/stores/local_durable/checkpoint.py index 474ae781e..f995b7ac1 100644 --- a/coding-agent/src/coding_agent/stores/local_durable/checkpoint.py +++ b/coding-agent/src/coding_agent/stores/local_durable/checkpoint.py @@ -15,6 +15,7 @@ _optional_entry_str, ) from coding_agent.stores.runtime_store import ( + StateVersionConflictError, _datetime_to_json, ) from coding_agent.topics.store import ( @@ -32,7 +33,12 @@ _require_json_object, _require_non_empty, ) -from coding_agent.runtime_activation import assert_checkpoint_allowed +from coding_agent.runtime_activation import ( + assert_checkpoint_allowed, + is_new_runtime_restore_point, + restore_point_cas_matches, + restore_point_stamp, +) class LocalCheckpointMixin: @@ -161,6 +167,11 @@ async def restore_checkpoint_state( tape_id=meta.tape_id, session_id=authority.session_id, ) + self._assert_restore_point_cas( + connection, + session_id=authority.session_id, + snapshot=snapshot, + ) checkpoint_row = connection.execute( """ SELECT session_id, tape_id @@ -259,6 +270,42 @@ async def restore_checkpoint_state( ) self._open_projection_epoch(connection, authority.session_id) + def _assert_restore_point_cas( + self, + connection: sqlite3.Connection, + *, + session_id: str, + snapshot: CheckpointSnapshot, + ) -> None: + if not is_new_runtime_restore_point(snapshot): + return + stamp = restore_point_stamp(snapshot.extra or {}) + if stamp is None: + return + run_id = stamp.get("run_id") + if not isinstance(run_id, str) or not run_id: + raise ValueError( + "operation_state_version.run_id must be a non-empty string" + ) + row = connection.execute( + """ + SELECT run_id, revision, projection_epoch, transition_id + FROM session_operation_states + WHERE session_id = ? AND run_id = ? + """, + (session_id, run_id), + ).fetchone() + if row is None or not restore_point_cas_matches( + stamp, + run_id=row["run_id"], + revision=int(row["revision"]), + projection_epoch=int(row["projection_epoch"]), + transition_id=row["transition_id"], + ): + raise StateVersionConflictError( + "restore point operation state does not match" + ) + def _reconcile_topics_after_checkpoint_restore( self, connection: sqlite3.Connection, diff --git a/coding-agent/src/coding_agent/stores/pg_durable/checkpoint.py b/coding-agent/src/coding_agent/stores/pg_durable/checkpoint.py index 8762b0761..514bbb190 100644 --- a/coding-agent/src/coding_agent/stores/pg_durable/checkpoint.py +++ b/coding-agent/src/coding_agent/stores/pg_durable/checkpoint.py @@ -15,7 +15,13 @@ _require_payload_session, _required_owned_row, ) -from coding_agent.runtime_activation import assert_checkpoint_allowed +from coding_agent.runtime_activation import ( + assert_checkpoint_allowed, + is_new_runtime_restore_point, + restore_point_cas_matches, + restore_point_stamp, +) +from coding_agent.stores.runtime_store import StateVersionConflictError class PgCheckpointMixin: @@ -83,6 +89,11 @@ async def restore_checkpoint_state( async def body(connection: Any) -> None: await self._require_owner(connection, authority) await self._require_stable_tape(connection, authority, meta.tape_id) + await self._assert_restore_point_cas( + connection, + session_id=authority.session_id, + snapshot=snapshot, + ) await self._require_checkpoint_owner( connection, authority, @@ -125,6 +136,43 @@ async def body(connection: Any) -> None: await self._with_transaction(body) + async def _assert_restore_point_cas( + self, + connection: Any, + *, + session_id: str, + snapshot: CheckpointSnapshot, + ) -> None: + if not is_new_runtime_restore_point(snapshot): + return + stamp = restore_point_stamp(snapshot.extra or {}) + if stamp is None: + return + run_id = stamp.get("run_id") + if not isinstance(run_id, str) or not run_id: + raise ValueError( + "operation_state_version.run_id must be a non-empty string" + ) + row = await connection.fetchrow( + """ + SELECT run_id, revision, projection_epoch, transition_id + FROM session_operation_states + WHERE session_id = $1 AND run_id = $2 + """, + session_id, + run_id, + ) + if row is None or not restore_point_cas_matches( + stamp, + run_id=str(row["run_id"]), + revision=int(row["revision"]), + projection_epoch=int(row["projection_epoch"]), + transition_id=str(row["transition_id"]), + ): + raise StateVersionConflictError( + "restore point operation state does not match" + ) + async def _reconcile_topics_after_checkpoint_restore( self, connection: Any, diff --git a/coding-agent/tests/coding_agent/test_phase_g_restore_point.py b/coding-agent/tests/coding_agent/test_phase_g_restore_point.py index cb5c6ecf9..1a67224c3 100644 --- a/coding-agent/tests/coding_agent/test_phase_g_restore_point.py +++ b/coding-agent/tests/coding_agent/test_phase_g_restore_point.py @@ -6,13 +6,21 @@ import pytest from agentkit.checkpoint.models import CheckpointMeta, CheckpointSnapshot +from agentkit.runtime.contracts import OperationStateCAS from coding_agent.runtime_activation import ( CHECKPOINT_FORMAT_KEY, + OPERATION_STATE_VERSION_KEY, RUNTIME_VERSION_NEW, NewRuntimeCheckpointRejectedError, assert_checkpoint_allowed, is_new_runtime_restore_point, ) +from coding_agent.runs.checkpoint_capture import serialize_operation_state_version +from coding_agent.stores.runtime_store import ( + AuthoritativeUnitOfWork, + EventRecord, + StateVersionConflictError, +) from tests.coding_agent.test_harness_p2_fact_source import _open_store from tests.coding_agent.test_phase_f_checkpoint_rejection import _snapshot from tests.coding_agent.test_phase_f_runtime_version import ( @@ -34,7 +42,10 @@ def _restore_point(session_id: str, tape_id: str) -> CheckpointSnapshot: ), tape_entries=(), plugin_states={}, - extra={CHECKPOINT_FORMAT_KEY: RUNTIME_VERSION_NEW}, + extra={ + CHECKPOINT_FORMAT_KEY: RUNTIME_VERSION_NEW, + OPERATION_STATE_VERSION_KEY: None, + }, ) @@ -141,3 +152,104 @@ async def test_cross_version_checkpoint_still_rejects_before_mutation( await store.save_session(owner, {**payload, "tape_id": tape_id}) with pytest.raises(NewRuntimeCheckpointRejectedError): await store.save_checkpoint(owner, _restore_point(session_id, tape_id)) + + +def _stamped_restore_point( + session_id: str, + tape_id: str, + stamp: object, +) -> CheckpointSnapshot: + snapshot = _restore_point(session_id, tape_id) + return CheckpointSnapshot( + meta=snapshot.meta, + tape_entries=(), + plugin_states={}, + extra={ + CHECKPOINT_FORMAT_KEY: RUNTIME_VERSION_NEW, + OPERATION_STATE_VERSION_KEY: stamp, + }, + ) + + +async def _commit_restore_point_state( + store: object, + owner: object, + session_id: str, + payload: dict[str, object], +) -> object: + return await store.commit_authoritative_uow( # type: ignore[union-attr] + owner, + AuthoritativeUnitOfWork( + event=None, + session_state=payload, + transition_id="g3-transition", + state_cas=OperationStateCAS( + run_id="run-g3", + revision=0, + projection_epoch=0, + ), + state_value={"phase": "captured"}, + facts=( + EventRecord( + event_id=f"fact-g3-{session_id}", + session_id=session_id, + event_kind="assistant_message", + payload={"text": "g3"}, + created_at=datetime(2026, 9, 3, tzinfo=UTC), + ), + ), + ), + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("store_kind", ["sqlite", "pg"]) +async def test_restore_point_matching_operation_state_succeeds( + store_kind: str, + tmp_path: Path, +) -> None: + store, _owner = await _open_store(store_kind, tmp_path) + await store.set_new_session_runtime_activation(enabled=True) + session_id = f"session-g3-match-{store_kind}" + tape_id = f"tape-g3-match-{store_kind}" + owner, payload = await _fresh_session(store, store_kind, session_id) + session_payload = {**payload, "tape_id": tape_id} + await store.save_session(owner, session_payload) + stored = await _load_session(store, session_id) + assert stored is not None + committed = await _commit_restore_point_state(store, owner, session_id, stored) + stamp = serialize_operation_state_version(committed.state_version) + snapshot = _stamped_restore_point(session_id, tape_id, stamp) + await store.save_checkpoint(owner, snapshot) + await store.restore_checkpoint_state(owner, snapshot, stored) + loaded = await _load_session(store, session_id) + assert loaded is not None + assert loaded["runtime_version"] == RUNTIME_VERSION_NEW + + +@pytest.mark.asyncio +@pytest.mark.parametrize("store_kind", ["sqlite", "pg"]) +async def test_restore_point_mismatched_operation_state_rejects_before_mutation( + store_kind: str, + tmp_path: Path, +) -> None: + store, _owner = await _open_store(store_kind, tmp_path) + await store.set_new_session_runtime_activation(enabled=True) + session_id = f"session-g3-mismatch-{store_kind}" + tape_id = f"tape-g3-mismatch-{store_kind}" + owner, payload = await _fresh_session(store, store_kind, session_id) + session_payload = {**payload, "tape_id": tape_id} + await store.save_session(owner, session_payload) + stored = await _load_session(store, session_id) + assert stored is not None + committed = await _commit_restore_point_state(store, owner, session_id, stored) + stamp = serialize_operation_state_version(committed.state_version) + assert stamp is not None + stamp = {**stamp, "revision": 99} + snapshot = _stamped_restore_point(session_id, tape_id, stamp) + await store.save_checkpoint(owner, snapshot) + with pytest.raises(StateVersionConflictError, match="operation state"): + await store.restore_checkpoint_state(owner, snapshot, stored) + loaded = await _load_session(store, session_id) + assert loaded is not None + assert loaded["runtime_version"] == RUNTIME_VERSION_NEW diff --git a/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py b/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py index 4e24bfefc..d18a66e96 100644 --- a/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py +++ b/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py @@ -23,6 +23,7 @@ class FakeSession: approval_policy: ApprovalPolicy = ApprovalPolicy.INTERACTIVE provider: object | None = None runtime_version: str = "legacy" + current_turn_id: str | None = None def attach_runtime_binding( self, @@ -195,6 +196,7 @@ async def persist_session(current_session: FakeSession) -> None: async def test_new_runtime_capture_uses_restore_point_backend() -> None: from coding_agent.runtime_activation import ( CHECKPOINT_FORMAT_KEY, + OPERATION_STATE_VERSION_KEY, RUNTIME_VERSION_NEW, ) @@ -224,6 +226,7 @@ async def persist_session(current_session: FakeSession) -> None: "label": "g1", "extra": { CHECKPOINT_FORMAT_KEY: RUNTIME_VERSION_NEW, + OPERATION_STATE_VERSION_KEY: None, "session_restart_config": { "provider_name": "anthropic", "model_name": "claude-checkpoint", @@ -234,3 +237,66 @@ async def persist_session(current_session: FakeSession) -> None: }, } ] + + +@pytest.mark.asyncio +async def test_new_runtime_capture_stamps_operation_state_version() -> None: + from agentkit.runtime.contracts import CommitRef, OperationStateVersion + from coding_agent.runtime_activation import ( + CHECKPOINT_FORMAT_KEY, + OPERATION_STATE_VERSION_KEY, + RUNTIME_VERSION_NEW, + ) + + state = OperationStateVersion( + run_id="run-1", + revision=3, + projection_epoch=1, + commit_ref=CommitRef( + transition_id="transition-1", + fact_seq_start="1", + fact_seq_end="2", + ), + value={"phase": "ready"}, + ) + session = FakeSession( + runtime_version=RUNTIME_VERSION_NEW, + current_turn_id="run-1", + ) + ctx = types.SimpleNamespace(tape=types.SimpleNamespace(tape_id="stable-tape")) + backend = RecordingCheckpointBackend(_checkpoint()) + + async def ensure_runtime(session_id: str): + del session_id + return ctx + + async def persist_session(current_session: FakeSession) -> None: + del current_session + + async def load_operation_state( + session_id: str, run_id: str + ) -> OperationStateVersion: + assert session_id == "session-1" + assert run_id == "run-1" + return state + + await RuntimeCheckpointCaptureService( + checkpoint_service=lambda: backend, + ensure_runtime=ensure_runtime, + persist_session=persist_session, + load_operation_state=load_operation_state, + ).capture(session, label="g3") + + extra = backend.restore_point_calls[0]["extra"] + assert extra[CHECKPOINT_FORMAT_KEY] == RUNTIME_VERSION_NEW + assert extra[OPERATION_STATE_VERSION_KEY] == { + "run_id": "run-1", + "revision": 3, + "projection_epoch": 1, + "commit_ref": { + "transition_id": "transition-1", + "fact_seq_start": "1", + "fact_seq_end": "2", + }, + "value": {"phase": "ready"}, + } From e595e7c18b8ea177916ffcf5fcced4b26eea229c Mon Sep 17 00:00:00 2001 From: kina0630 Date: Thu, 3 Sep 2026 20:16:45 +0800 Subject: [PATCH 4/5] fix(runtime): make RestorePoint capture and post-restore turns work JSON-plain frozen operation state values so HTTP capture can save. Seed a new turn's CAS at the current projection epoch after restore. --- .../coding_agent/runs/checkpoint_capture.py | 7 +++++-- .../src/coding_agent/runs/serving_runtime.py | 5 ++++- .../coding_agent/server/session/manager.py | 11 +++++++++++ .../coding_agent/test_phase_f_serving.py | 11 +++++++++++ ...test_runtime_checkpoint_capture_service.py | 19 +++++++++++++++++++ 5 files changed, 50 insertions(+), 3 deletions(-) diff --git a/coding-agent/src/coding_agent/runs/checkpoint_capture.py b/coding-agent/src/coding_agent/runs/checkpoint_capture.py index 7927cc2f1..ec1b562e0 100644 --- a/coding-agent/src/coding_agent/runs/checkpoint_capture.py +++ b/coding-agent/src/coding_agent/runs/checkpoint_capture.py @@ -12,7 +12,7 @@ OPERATION_STATE_VERSION_KEY, RUNTIME_VERSION_NEW, ) - +from coding_agent.stores.runtime_store import _plain_json from .checkpoint_restore import ( CHECKPOINT_SESSION_CONFIG_KEY, @@ -63,6 +63,9 @@ def serialize_operation_state_version( ) -> dict[str, Any] | None: if state is None: return None + value = _plain_json(state.value) + if not isinstance(value, dict): + raise TypeError("operation state value must be an object") return { "run_id": state.run_id, "revision": state.revision, @@ -72,7 +75,7 @@ def serialize_operation_state_version( "fact_seq_start": state.commit_ref.fact_seq_start, "fact_seq_end": state.commit_ref.fact_seq_end, }, - "value": dict(state.value), + "value": value, } diff --git a/coding-agent/src/coding_agent/runs/serving_runtime.py b/coding-agent/src/coding_agent/runs/serving_runtime.py index 77693ef98..54b7d362a 100644 --- a/coding-agent/src/coding_agent/runs/serving_runtime.py +++ b/coding-agent/src/coding_agent/runs/serving_runtime.py @@ -608,11 +608,12 @@ def initial_operation_state( run_id: str, *, system_prompt: str, + projection_epoch: int = 0, ) -> OperationStateVersion: return OperationStateVersion( run_id=run_id, revision=0, - projection_epoch=0, + projection_epoch=projection_epoch, commit_ref=CommitRef(transition_id=f"{run_id}:admission"), value={ "context": { @@ -633,6 +634,7 @@ def build_new_runtime_turn_adapter( model_adapter: object | None = None, effect_executor: object | None = None, resolve_blocked: ServingBlockedResolver | None = None, + projection_epoch: int = 0, ) -> DurableSegmentTurnAdapter: session_state: Mapping[str, object] to_store_data = getattr(session, "to_store_data", None) @@ -668,6 +670,7 @@ def build_new_runtime_turn_adapter( resolved_state = state_version or initial_operation_state( run_id, system_prompt=system_prompt, + projection_epoch=projection_epoch, ) return DurableSegmentTurnAdapter( runner=runner, diff --git a/coding-agent/src/coding_agent/server/session/manager.py b/coding-agent/src/coding_agent/server/session/manager.py index 12de2b6af..3650e750d 100644 --- a/coding-agent/src/coding_agent/server/session/manager.py +++ b/coding-agent/src/coding_agent/server/session/manager.py @@ -830,6 +830,7 @@ async def _build_new_runtime_turn_adapter( blocked_approval_tool, build_new_runtime_turn_adapter, ) + from coding_agent.stores.runtime_store import parse_u64 from coding_agent.wire.protocol import ApprovalRequest store = self._authoritative_store() @@ -842,6 +843,15 @@ async def _build_new_runtime_turn_adapter( if not isinstance(system_prompt, str): raise RuntimeError("new-runtime serving requires a system prompt") state = await store.load_operation_state(session.id, request.run_id) + projection_epoch = 0 + if state is None: + load_fact = getattr(store, "load_session_fact_source", None) + if callable(load_fact): + fact = await load_fact(session.id) + if fact is not None: + projection_epoch = parse_u64( + fact.projection_epoch, field_name="projection_epoch" + ) async def durable_settlement(blocked: Any) -> Any: mailbox = await store.load_runtime_command_mailbox(session.id) @@ -906,4 +916,5 @@ async def resolve_blocked(blocked: Any) -> Any: state_version=state, system_prompt=system_prompt, resolve_blocked=resolve_blocked, + projection_epoch=projection_epoch, ) diff --git a/coding-agent/tests/coding_agent/test_phase_f_serving.py b/coding-agent/tests/coding_agent/test_phase_f_serving.py index fc3da3c70..3ef6108f8 100644 --- a/coding-agent/tests/coding_agent/test_phase_f_serving.py +++ b/coding-agent/tests/coding_agent/test_phase_f_serving.py @@ -1037,3 +1037,14 @@ async def resolve_blocked(blocked: BlockedOutcome) -> object: assert isinstance(outcome, RoundLimitOutcome) assert outcome.steps_taken == 2 assert resolver_called is False + + +def test_initial_operation_state_uses_current_projection_epoch() -> None: + state = initial_operation_state( + "run-after-restore", + system_prompt="sys", + projection_epoch=2, + ) + assert state.run_id == "run-after-restore" + assert state.revision == 0 + assert state.projection_epoch == 2 diff --git a/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py b/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py index d18a66e96..18ea8f33f 100644 --- a/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py +++ b/coding-agent/tests/coding_agent/test_runtime_checkpoint_capture_service.py @@ -300,3 +300,22 @@ async def load_operation_state( }, "value": {"phase": "ready"}, } + + +def test_serialize_operation_state_version_plain_json_nested_mapping() -> None: + import json + + from agentkit.runtime.contracts import CommitRef, OperationStateVersion + from coding_agent.runs.checkpoint_capture import serialize_operation_state_version + + state = OperationStateVersion( + run_id="run-1", + revision=1, + projection_epoch=0, + commit_ref=CommitRef(transition_id="t1", fact_seq_start="1", fact_seq_end="1"), + value={"nested": {"ok": True}}, + ) + payload = serialize_operation_state_version(state) + assert payload is not None + json.dumps(payload) + assert payload["value"] == {"nested": {"ok": True}} From bf5377214c896b7592b1dfd21371d6b25e93ef07 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Thu, 3 Sep 2026 20:58:57 +0800 Subject: [PATCH 5/5] test(runtime): align wrap UoW retries with immutable event epochs Duplicate event_id lost-acks no longer rewrite mailbox. Restore then the same event_id fails CAS instead of promoting projection_epoch. --- .../coding_agent/test_harness_p2_wrap.py | 68 ++++--------------- 1 file changed, 15 insertions(+), 53 deletions(-) diff --git a/coding-agent/tests/coding_agent/test_harness_p2_wrap.py b/coding-agent/tests/coding_agent/test_harness_p2_wrap.py index f3406f774..9103be8ab 100644 --- a/coding-agent/tests/coding_agent/test_harness_p2_wrap.py +++ b/coding-agent/tests/coding_agent/test_harness_p2_wrap.py @@ -29,6 +29,7 @@ EventRecord, MailboxDispositionSlot, ProjectionCursor, + StateVersionConflictError, ) from coding_agent.wire.protocol import ( ApprovalRequest, @@ -163,14 +164,14 @@ async def test_duplicate_event_id_is_store_idempotent_and_keeps_mailbox( assert fact.session_seq == first.event.session_seq mailbox = await store.load_mailbox_slot(SESSION_ID, "mailbox-main") assert mailbox is not None - assert mailbox.disposition == "settled" - assert mailbox.payload == {"lane_cut": "replay"} + assert mailbox.disposition == "queued-one" + assert mailbox.payload == {"lane_cut": "one"} assert await store.load_event_record(SESSION_ID, "2") is None @pytest.mark.asyncio @pytest.mark.parametrize("store_kind", ["sqlite", "pg"]) -async def test_restore_then_same_event_id_is_visible_in_current_epoch( +async def test_restore_then_same_event_id_fails_cas_without_write( store_kind: str, tmp_path: Path, ) -> None: @@ -198,59 +199,21 @@ async def test_restore_then_same_event_id_is_visible_in_current_epoch( payload={"lane_cut": "replay-after-restore"}, ), ) - second = await store.commit_authoritative_uow(owner, replayed) + with pytest.raises(StateVersionConflictError, match="projection_epoch"): + await store.commit_authoritative_uow(owner, replayed) - assert second.idempotent is True - assert second.event.event_id == first.event.event_id - assert second.event.session_seq == first.event.session_seq - assert second.event.event_kind == first.event.event_kind - assert second.event.payload == first.event.payload - assert second.event.projection_epoch == "1" - assert second.projection_epoch == "1" + stored = await store.load_event_record(SESSION_ID, first.event.session_seq) + assert stored is not None + assert stored.event_id == first.event.event_id + assert stored.projection_epoch == "0" after = await store.load_session_fact_source(SESSION_ID) assert after is not None assert after.session_seq == first.event.session_seq assert after.projection_epoch == "1" - - current = ProjectionCursor( - kind="delta", - session_id=SESSION_ID, - projection=DEFAULT_HARNESS_PROJECTION, - epoch="1", - session_seq="0", - ) - visible = await store.replay_projection(current) - assert [event.event_id for event in visible] == ["event-one"] - assert visible[0].event_kind == "harness.TurnCommitted" - assert visible[0].payload == {"suffix": "one"} - assert visible[0].projection_epoch == "1" - - third = await store.commit_authoritative_uow( - owner, - AuthoritativeUnitOfWork( - event=_event("one"), - session_state={**SESSION_PAYLOAD, "turn": "same-epoch-again"}, - mailbox=MailboxDispositionSlot( - slot_id="mailbox-main", - lane="user", - disposition="settled", - payload={"lane_cut": "same-epoch-again"}, - ), - ), - ) - assert third.idempotent is True - assert third.event.session_seq == first.event.session_seq - assert third.event.projection_epoch == "1" - fact_again = await store.load_session_fact_source(SESSION_ID) - assert fact_again is not None - assert fact_again.session_seq == first.event.session_seq - mailbox = await store.load_mailbox_slot(SESSION_ID, "mailbox-main") - assert mailbox is not None - assert mailbox.disposition == "settled" - assert mailbox.payload == {"lane_cut": "same-epoch-again"} assert await store.load_event_record(SESSION_ID, "2") is None + @pytest.mark.asyncio async def test_duplicate_boundary_uow_keeps_mailbox_settled(tmp_path: Path) -> None: manager = _durable_manager(tmp_path) @@ -287,7 +250,7 @@ async def test_duplicate_boundary_uow_keeps_mailbox_settled(tmp_path: Path) -> N mailbox = await store.load_mailbox_slot(session_id, slot_id) assert mailbox is not None - assert mailbox.disposition == "settled" + assert mailbox.disposition == "in_flight" replay = await store.replay_from_retention_floor(session_id) settled = [event for event in replay.events if event.event_kind == _TURN_SETTLED] assert len(settled) == 1 @@ -550,11 +513,10 @@ async def test_restore_then_reapprove_reuses_same_effect_id(tmp_path: Path) -> N clobbered = await store.load_effect_slot(session_id, effect_id) assert clobbered is not None assert clobbered.status == "stale-after-restore" - session.begin_approval_request(request) - await manager._persist_approval_requested(session) + with pytest.raises(StateVersionConflictError, match="projection_epoch"): + await manager._persist_approval_requested(session) reused = await store.load_effect_slot(session_id, effect_id) assert reused is not None assert reused.effect_id == effect_id - assert reused.status == "prepared" - assert reused.payload["request_id"] == request.request_id + assert reused.status == "stale-after-restore"