From 008fc4d59ec110f4d4ca881a6514bafde9543a49 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 2 Nov 2026 05:30:55 +0530 Subject: [PATCH 1/2] test(monitor): inspect original native guard captures Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_monitor_target_transport.py | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/tests/control_plane/test_monitor_target_transport.py b/tests/control_plane/test_monitor_target_transport.py index 70974708d4..5836b6857d 100644 --- a/tests/control_plane/test_monitor_target_transport.py +++ b/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,6 +94,7 @@ 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"] @@ -87,7 +102,7 @@ def test_real_large_authority_returns_complete_target(tmp_path, monkeypatch, pro # 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") From f214f512e38697f5c8eaecac695ceb8c0aebbf4e Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 2 Nov 2026 05:30:55 +0530 Subject: [PATCH 2/2] refactor(runtime): share local snapshot method eligibility Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/effect_runtime.py | 31 +++---------------- loopx/control_plane/effect_runtime_server.ts | 29 ++--------------- .../effect_runtime_snapshot_methods.json | 27 ++++++++++++++++ .../test_effect_runtime_integration.py | 28 +++++++++++++++++ 4 files changed, 61 insertions(+), 54 deletions(-) create mode 100644 loopx/control_plane/effect_runtime_snapshot_methods.json diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 6d4ab200b8..e68e66757f 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/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 diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts index cfb2465ff2..6c4d1d151c 100644 --- a/loopx/control_plane/effect_runtime_server.ts +++ b/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, diff --git a/loopx/control_plane/effect_runtime_snapshot_methods.json b/loopx/control_plane/effect_runtime_snapshot_methods.json new file mode 100644 index 0000000000..7bc340b13d --- /dev/null +++ b/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" +] diff --git a/tests/control_plane/test_effect_runtime_integration.py b/tests/control_plane/test_effect_runtime_integration.py index 5b8f42de61..d17cd35964 100644 --- a/tests/control_plane/test_effect_runtime_integration.py +++ b/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"}]