fix(online-agent): persist external runtime completion checkpoint phase

active-runtime/app/aios_online_agent_checkpoint.py

@@ -17,6 +17,7 @@ PHASE_ORDER = {
     "claimed": 10,
     "workspace_prepared": 20,
     "verification_passed": 30,
+    "agent_runtime_done": 40,
     "source_capture_done": 40,
     "commit_done": 40,
     "reconciliation_required": 45,

active-runtime/app/aios_online_agent_worker.py

@@ -1776,7 +1776,7 @@ class WorkerSupervisor:
             # read-only source-capture task has no plan to replace its payload: replay
             # the already validated durable envelope so source signals remain part of
             # the canonical digest and the worker does not invent a different capture.
-            projection_input = execution if persisted.action == "source_capture" else None
+            projection_input = execution if persisted.action in {"source_capture", "agent_runtime"} else None
             status, authoritative = self.bridge.project_execution(task, projection_input)
             if not 200 <= status < 300:
                 raise WorkerError("TASK_PERSISTED_BINDING_UNVERIFIED", phase="projection", details={"api_status": status, "error": authoritative.get("error")})

tests/test_online_agent_agent_runtime_replay.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from pathlib import Path
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "active-runtime/app"))
+
+from aios_online_agent_execution import (  # noqa: E402
+    GovernedExecutionBinding,
+    project_execution_envelope,
+)
+import aios_online_agent_worker as worker  # noqa: F401,E402
+from aios_online_agent_worker import WorkerSupervisor  # noqa: E402
+
+
+BASE = "a" * 40
+
+
+def test_agent_runtime_replays_the_durable_envelope_to_bridge(monkeypatch) -> None:
+    monkeypatch.setattr(
+        worker,
+        "WORKER_CAPABILITIES",
+        tuple(sorted((*worker.WORKER_CAPABILITIES, "agent.runtime.hermes"))),
+    )
+    binding = GovernedExecutionBinding.from_mapping(
+        {
+            "task_id": "r42-agent-runtime-replay-test-20260818",
+            "job_id": "job-agent-runtime-replay-test",
+            "repository": "aios/aios",
+            "base_oid": BASE,
+            "expected_branch_oid": BASE,
+            "target_branch": "agent/agent-runtime-replay-test",
+            "branch_mode": "CREATE_NEW",
+            "generation": 1,
+            "idempotency_key": "agent-runtime-replay-test-v1",
+        }
+    )
+    execution = project_execution_envelope(
+        binding,
+        {
+            "action": "agent_runtime",
+            "adapter": "hermes",
+            "capability_scope": ["agent.runtime.hermes"],
+            "change_set": [],
+            "runtime_input": {
+                "goal": "replay the exact durable runtime envelope",
+                "instructions": "return the bounded sentinel",
+                "success_criteria": [
+                    {
+                        "id": "sentinel",
+                        "description": "return the bounded sentinel",
+                    }
+                ],
+                "constraints": {
+                    "provider_boundary": "provider-credential-only",
+                    "change_set": [],
+                    "writer_lease": False,
+                },
+            },
+        },
+    )
+    task = {
+        **binding.public(),
+        "required_capabilities": ["agent.runtime.hermes"],
+        "metadata": {"execution": execution},
+    }
+
+    class Bridge:
+        def __init__(self) -> None:
+            self.calls: list[object] = []
+
+        def project_execution(self, _task: object, projection_input: object = None):
+            self.calls.append(projection_input)
+            return 200, dict(projection_input or {})
+
+    supervisor = object.__new__(WorkerSupervisor)
+    supervisor.bridge = Bridge()
+    resolved, source = supervisor.resolve_envelope(task)
+
+    assert source == "persisted_verified"
+    assert resolved.action == "agent_runtime"
+    assert resolved.adapter == "hermes"
+    assert resolved.digest == execution["execution_digest"]
+    assert supervisor.bridge.calls == [execution]

tests/test_online_agent_checkpoint_phases.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from aios_online_agent_checkpoint import CheckpointBinding, CheckpointStore
+
+
+def _binding() -> CheckpointBinding:
+    return CheckpointBinding(
+        task_id="checkpoint-agent-runtime-test-20260818",
+        attempt=1,
+        execution_digest="a" * 64,
+        job_id="job-checkpoint-agent-runtime-test-20260818",
+        repository="aios/aios",
+        base_oid="b" * 40,
+        expected_branch_oid="b" * 40,
+        target_branch="agent/checkpoint-agent-runtime-test-20260818",
+        generation=1,
+    )
+
+
+def test_agent_runtime_checkpoint_phase_is_durable_before_result(tmp_path: Path) -> None:
+    store = CheckpointStore(tmp_path / "checkpoints.sqlite3")
+    binding = _binding()
+    phases = [
+        "claimed",
+        "workspace_prepared",
+        "verification_passed",
+        "agent_runtime_done",
+        "result_pending",
+        "result_submitted",
+    ]
+
+    rows = [
+        store.append(binding, phase, {"phase": phase, "task_id": binding.task_id})
+        for phase in phases
+    ]
+
+    assert [row["phase"] for row in rows] == phases
+    assert [int(row["phase_order"]) for row in rows] == [10, 20, 30, 40, 50, 60]
+    latest = store.latest_for_attempt(binding.task_id, binding.attempt)
+    assert latest is not None
+    assert latest["phase"] == "result_submitted"
+    assert latest["payload"] == {"phase": "result_submitted", "task_id": binding.task_id}
+
+
+def test_agent_runtime_checkpoint_append_is_idempotent(tmp_path: Path) -> None:
+    store = CheckpointStore(tmp_path / "checkpoints.sqlite3")
+    binding = _binding()
+    payload = {"phase": "agent_runtime_done", "result": {"status": "SUCCEEDED"}}
+
+    first = store.append(binding, "agent_runtime_done", payload)
+    replay = store.append(binding, "agent_runtime_done", payload)
+
+    assert replay["sequence"] == first["sequence"]
+    assert replay["payload_sha256"] == first["payload_sha256"]