refactor(runtime): share snapshot eligibility across Python and TypeScript

loopx/control_plane/effect_runtime.py

@@ -43,33 +43,10 @@
 MAX_RESPONSE_BYTES = 8 * 1024 * 1024
 MAX_REQUEST_BYTES = 8 * 1024 * 1024
 MAX_LOCAL_SNAPSHOT_BYTES = 64 * 1024 * 1024
-LOCAL_SNAPSHOT_METHODS = frozenset({
-    "todo.read",
-    "todo.quota_planning.project",
-    "goal.vision_wait.project",
-    "collaboration.inbox.read",
-    "collaboration.delegation.inventory_page",
-    "benchmark.trajectory.codex-usage",
-    "coordination.local_authority.todo_source",
-    "coordination.local_authority.todo_snapshot",
-    "todo.context.page",
-    "work_item.context.project",
-    "work_item.interaction_reads.project",
-    "scheduler.hint.project",
-    "work_item.task_graph.goal_topology",
-    "work_item.replan_semantics.project",
-    "quota.settlement.read",
-    "quota.spend.commit",
-    "quota.turn_envelope.evaluate",
-    "turn.controller",
-    "goal.checkpoint_read_context.resolve",
-    "goal.checkpoint_read_context.commit",
-    "goal.checkpoint_read_context.inspect_replay",
-    "performance_diagnosis.inspect",
-    "state.recovery.archive.verify",
-    "state.recovery.manifest.verify",
-    "runtime.checkpoint_artifacts",
-})
+# Shared packaged transport eligibility; business authority stays with each method.
+LOCAL_SNAPSHOT_METHODS = frozenset(json.loads(
+    Path(__file__).with_name("effect_runtime_snapshot_methods.json").read_text(encoding="utf-8"),
+))
 MAX_STARTUP_DIAGNOSTIC_BYTES = 8 * 1024
 STARTUP_LOCK_TIMEOUT_SECONDS = 15.0
 STARTUP_POLL_SECONDS = 0.025

loopx/control_plane/effect_runtime_server.ts

@@ -3,6 +3,7 @@ import { createServer, type Socket } from "node:net";
 import { realpath, type FileHandle } from "node:fs/promises";
 
 import type { JsonObject } from "./effect_program.ts";
+import localSnapshotMethods from "./effect_runtime_snapshot_methods.json" with { type: "json" };
 import {
   createEffectRuntimeHandlers,
   dispatchEffectRuntimeMethod,
@@ -27,33 +28,7 @@ const STARTUP_ERROR_SCHEMA = "loopx_effect_runtime_startup_error_v0";
 const MAX_REQUEST_BYTES = 8 * 1024 * 1024;
 const MAX_INLINE_RESPONSE_BYTES = 8 * 1024 * 1024;
 // Explicit opt-in: ordinary effects retain a bounded request/response wire.
-const LOCAL_SNAPSHOT_METHODS = new Set([
-  "todo.read",
-  "todo.quota_planning.project",
-  "goal.vision_wait.project",
-  "collaboration.inbox.read",
-  "collaboration.delegation.inventory_page",
-  "benchmark.trajectory.codex-usage",
-  "coordination.local_authority.todo_source",
-  "coordination.local_authority.todo_snapshot",
-  "todo.context.page",
-  "work_item.context.project",
-  "work_item.interaction_reads.project",
-  "scheduler.hint.project",
-  "work_item.task_graph.goal_topology",
-  "work_item.replan_semantics.project",
-  "quota.settlement.read",
-  "quota.spend.commit",
-  "quota.turn_envelope.evaluate",
-  "turn.controller",
-  "goal.checkpoint_read_context.resolve",
-  "goal.checkpoint_read_context.commit",
-  "goal.checkpoint_read_context.inspect_replay",
-  "performance_diagnosis.inspect",
-  "state.recovery.archive.verify",
-  "state.recovery.manifest.verify",
-  "runtime.checkpoint_artifacts",
-]);
+const LOCAL_SNAPSHOT_METHODS = new Set(localSnapshotMethods);
 const DEFAULT_IDLE_MS = 5 * 60 * 1_000;
 // Bounds for LOOPX_EFFECT_RUNTIME_IDLE_MS. The upper bound is the largest
 // delay `setTimeout` accepts: a larger delay overflows and fires immediately,

loopx/control_plane/effect_runtime_snapshot_methods.json

@@ -0,0 +1,27 @@
+[
+  "todo.read",
+  "todo.quota_planning.project",
+  "goal.vision_wait.project",
+  "collaboration.inbox.read",
+  "collaboration.delegation.inventory_page",
+  "benchmark.trajectory.codex-usage",
+  "coordination.local_authority.todo_source",
+  "coordination.local_authority.todo_snapshot",
+  "todo.context.page",
+  "work_item.context.project",
+  "work_item.interaction_reads.project",
+  "scheduler.hint.project",
+  "work_item.task_graph.goal_topology",
+  "work_item.replan_semantics.project",
+  "quota.settlement.read",
+  "quota.spend.commit",
+  "quota.turn_envelope.evaluate",
+  "turn.controller",
+  "goal.checkpoint_read_context.resolve",
+  "goal.checkpoint_read_context.commit",
+  "goal.checkpoint_read_context.inspect_replay",
+  "performance_diagnosis.inspect",
+  "state.recovery.archive.verify",
+  "state.recovery.manifest.verify",
+  "runtime.checkpoint_artifacts"
+]

tests/control_plane/test_effect_runtime_integration.py

@@ -754,6 +754,34 @@ def test_runtime_decode_change_rotates_identity_and_starts_replacement(
         raise
 
 
+def test_snapshot_method_declaration_rotates_runtime_and_changes_eligibility(
+    tmp_path: Path, monkeypatch, request,
+) -> None:
+    staged_root, _ = observe_copied_runtime(tmp_path, monkeypatch, request)
+    monkeypatch.setattr(effect_runtime, "_runtime_dir", lambda: tmp_path / "runtime")
+    source = tmp_path / "candidate"
+    source.write_bytes(b"synthetic candidate")
+    params = {"action": "probe", "components": [{"id": "code", "source_path": str(source)}]}
+    original_fingerprint = effect_runtime._runtime_fingerprint()
+    original = effect_runtime.effect_runtime_result("runtime.ping", {})
+    observed = effect_runtime.effect_runtime_result(
+        "runtime.checkpoint_artifacts", params, large_local_snapshot=True,
+    )
+    assert observed["source_witness"]["code"][0]["sha256"] == hashlib.sha256(source.read_bytes()).hexdigest()
+
+    declaration = staged_root / "effect_runtime_snapshot_methods.json"
+    methods = json.loads(declaration.read_text(encoding="utf-8"))
+    methods.remove("runtime.checkpoint_artifacts")
+    declaration.write_text(json.dumps(methods) + "\n", encoding="utf-8")
+    assert effect_runtime._runtime_fingerprint() != original_fingerprint
+    with pytest.raises(effect_runtime.EffectRuntimeRejected, match="snapshot transport is unavailable"):
+        effect_runtime.effect_runtime_result(
+            "runtime.checkpoint_artifacts", params, large_local_snapshot=True,
+        )
+    replacement = effect_runtime.effect_runtime_result("runtime.ping", {})
+    assert replacement["pid"] != original["pid"]
+
+
 def test_runtime_retirement_trace_filters_private_stderr() -> None:
     raw = b'private path/token\nloopx_fixture_retirement:{"event":{}}\nloopx_fixture_retirement:{"event":"lock_release_error","code":"EPERM","token":"private","path":"private","pid":12,"ms":1}\n'
     assert trace_events(raw) == [{"event": "lock_release_error", "pid": 12, "ms": 1, "code": "EPERM"}]

tests/control_plane/test_monitor_target_transport.py

@@ -15,6 +15,20 @@
 from loopx.control_plane.quota.monitor_poll import resolve_due_monitor_candidate
 from loopx.control_plane.scheduler import monitor_poll_writeback, monitor_todo
 from loopx.control_plane.testing.canary_harness import run_json_cli, run_json_cli_result
+from loopx.control_plane.turn_driver.decision_capture import (
+    read_context_capture_file,
+    read_context_read_reference,
+)
+
+
+def _read_guard_capture(payload):
+    source = payload.get("captured_decision") or payload.get("recovery")
+    if source is None:
+        return payload
+    if "reference_file" in source:
+        _, source = read_context_read_reference(source["reference_file"])
+    _, _, decision = read_context_capture_file(source["path"], source["sha256"], private=True)
+    return decision
 
 
 def _monitor(**fields):
@@ -80,14 +94,15 @@ def test_real_large_authority_returns_complete_target(tmp_path, monkeypatch, pro
         before_missing = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL_ID)
         returncode, missing = run_json_cli_result(*guard, registry_path=registry, runtime_root=runtime)
         assert returncode != 0 and missing["ok"] is False
+        missing = _read_guard_capture(missing)
         assert missing["state"] == "blocked_health" and missing["should_run"] is False
         assert not state.exists()
         assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL_ID)["todos"] == before_missing["todos"]
 
         # Restore Goal intent only; the empty source cannot supply the 321 Todos.
         # Recovery must reuse the original Turn before the Monitor may execute.
         state.write_text(goal_source, encoding="utf-8")
-        admitted = run_json_cli(*guard, registry_path=registry, runtime_root=runtime)
+        admitted = _read_guard_capture(run_json_cli(*guard, registry_path=registry, runtime_root=runtime))
         assert admitted["should_run"] is True
         assert admitted["selected_todo"]["todo_id"] == target["todo_id"]
         poll = ["quota", "monitor-poll", *scope, "--todo-id", target["todo_id"],
@@ -97,7 +112,7 @@ def test_real_large_authority_returns_complete_target(tmp_path, monkeypatch, pro
         assert first["turn_continuation"]["current_turn_settled"] is True
         replay = run_json_cli(*poll, registry_path=registry, runtime_root=runtime)
         assert replay["replayed"] is True and replay["appended"] is False
-        closed = run_json_cli(*guard, registry_path=registry, runtime_root=runtime)
+        closed = _read_guard_capture(run_json_cli(*guard, registry_path=registry, runtime_root=runtime))
         assert closed["effective_action"] == "heartbeat_settled_skip"
         canonical = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL_ID)
         assert canonical["source_authority"] == ("sqlite_v0" if provider == "sqlite" else "file_v0")