From 11e26ea72fecb0cf1cabb197bf32a3e397dff929 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 14:18:24 +0800 Subject: [PATCH 1/2] fix(replan): route complete semantic facts through local snapshots Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/effect_runtime.py | 1 + loopx/control_plane/effect_runtime_server.ts | 1 + .../goals/goal_frontier/__init__.py | 4 +- .../control_plane/quota/should_run_prepare.py | 5 +- .../work_items/lifecycle_recovery.py | 6 +- .../work_items/progress_observation.py | 9 +-- .../work_items/replan_semantics_codec.py | 15 +++++ .../test_lifecycle_guard_reentry.py | 8 ++- .../test_replan_semantics_transport.py | 55 +++++++++++++++++++ .../effect_runtime_snapshot.test.ts | 3 + 10 files changed, 95 insertions(+), 12 deletions(-) create mode 100644 loopx/control_plane/work_items/replan_semantics_codec.py create mode 100644 tests/control_plane/test_replan_semantics_transport.py diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 16f8651c68..823c224e4d 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -45,6 +45,7 @@ "coordination.local_authority.todo_source", "todo.context.page", "work_item.context.project", + "work_item.replan_semantics.project", "goal.checkpoint_read_context.resolve", "goal.checkpoint_read_context.commit", "goal.checkpoint_read_context.inspect_replay", diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts index 60aeb88f96..7f024eed59 100644 --- a/loopx/control_plane/effect_runtime_server.ts +++ b/loopx/control_plane/effect_runtime_server.ts @@ -31,6 +31,7 @@ const LOCAL_SNAPSHOT_METHODS = new Set([ "coordination.local_authority.todo_source", "todo.context.page", "work_item.context.project", + "work_item.replan_semantics.project", "goal.checkpoint_read_context.resolve", "goal.checkpoint_read_context.commit", "goal.checkpoint_read_context.inspect_replay", diff --git a/loopx/control_plane/goals/goal_frontier/__init__.py b/loopx/control_plane/goals/goal_frontier/__init__.py index 257b7ceac7..27e49c5374 100644 --- a/loopx/control_plane/goals/goal_frontier/__init__.py +++ b/loopx/control_plane/goals/goal_frontier/__init__.py @@ -61,7 +61,7 @@ GOAL_ACCEPTANCE_HOLD_TRIGGERS, acceptance_gaps_from_held_goal_binding, ) -from ...effect_runtime import effect_runtime_result +from ...work_items.replan_semantics_codec import project_replan_semantics from .ack_policy import ( attach_rejected_replan_feedback, autonomous_replan_ack_satisfies_obligation, @@ -924,7 +924,7 @@ def _vision_gap_acknowledged( ) ): return False - return effect_runtime_result("work_item.replan_semantics.project", { + return project_replan_semantics({ "operation": "vision_ack", "acceptance_gaps": acceptance_gaps, "ack": latest_replan_ack, diff --git a/loopx/control_plane/quota/should_run_prepare.py b/loopx/control_plane/quota/should_run_prepare.py index 89e43c0a20..2e88f2b803 100644 --- a/loopx/control_plane/quota/should_run_prepare.py +++ b/loopx/control_plane/quota/should_run_prepare.py @@ -37,7 +37,8 @@ ) from ..quota.blocked_transition_notice import build_blocked_transition_notice from ..quota.error_codes import HeartbeatReceiptIdentityConflictError -from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result +from ..effect_runtime import EffectRuntimeRejected +from ..work_items.replan_semantics_codec import project_replan_semantics from ..agents.capability_memory import resolve_agent_capabilities from ..quota.goal_boundary import ( goal_boundary as _goal_boundary, @@ -200,7 +201,7 @@ def _preserve_receipt_bound_replan_obligation( if not preserved_replan_id: return dict(replan_obligation) if replan_obligation is not None else None try: - result = effect_runtime_result("work_item.replan_semantics.project", { + result = project_replan_semantics({ "operation": "receipt_bound_obligation", "current_obligation": dict(replan_obligation) if replan_obligation is not None else None, "selected_obligation_id": preserved_replan_id, "guard_scoped": guard_scoped, diff --git a/loopx/control_plane/work_items/lifecycle_recovery.py b/loopx/control_plane/work_items/lifecycle_recovery.py index 7f9707521f..df03fff539 100644 --- a/loopx/control_plane/work_items/lifecycle_recovery.py +++ b/loopx/control_plane/work_items/lifecycle_recovery.py @@ -4,7 +4,7 @@ from collections.abc import Callable from typing import Any -from ..effect_runtime import effect_runtime_result +from .replan_semantics_codec import project_replan_semantics def receipt_bound_lifecycle_candidate( @@ -15,7 +15,7 @@ def receipt_bound_lifecycle_candidate( items: list[dict[str, Any]], derive: Callable[[dict[str, Any]], dict[str, Any] | None], ) -> dict[str, Any] | None: - facts = effect_runtime_result("work_item.replan_semantics.project", { + facts = project_replan_semantics({ "operation": "lifecycle_source", "guard_recorded_at": guard_recorded_at, "agent_id": agent_id, "items": items, }) @@ -26,7 +26,7 @@ def receipt_bound_lifecycle_candidate( "completed_without_successor_count": len(gaps)}) if not source or source.get("obligation_id") != selected_obligation_id: return None - result = effect_runtime_result("work_item.replan_semantics.project", { + result = project_replan_semantics({ "operation": "lifecycle_obligation", "obligation": source, "selected_obligation_id": selected_obligation_id, "guard_recorded_at": guard_recorded_at, "agent_id": agent_id, "items": items, diff --git a/loopx/control_plane/work_items/progress_observation.py b/loopx/control_plane/work_items/progress_observation.py index bb83a47f2a..9a5ea58582 100644 --- a/loopx/control_plane/work_items/progress_observation.py +++ b/loopx/control_plane/work_items/progress_observation.py @@ -8,6 +8,7 @@ from ...turn_identity import normalize_turn_instance_id from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result +from .replan_semantics_codec import project_replan_semantics from ..goals.goal_vision_state import normalize_goal_vision_state from ..todos.contract import ( normalize_todo_task_domain, @@ -331,7 +332,7 @@ def replan_writeback_requirements( ) -> dict[str, Any]: """Adapt the shared typed discharge policy for CLI and host projection.""" try: - result = effect_runtime_result("work_item.replan_semantics.project", { + result = project_replan_semantics({ "operation": "requirements", "obligation": dict(obligation), }) except EffectRuntimeRejected as exc: @@ -350,7 +351,7 @@ def replan_successor_transition_allowed( ) -> bool: """Ask the typed owner whether a Todo transition can settle this review.""" try: - result = effect_runtime_result("work_item.replan_semantics.project", { + result = project_replan_semantics({ "operation": "transition_eligibility", "obligation": dict(obligation), "stride_strategy": stride_strategy, }) @@ -367,7 +368,7 @@ def guarded_replan_transition_delta( ) -> dict[str, Any] | None: """Adapt revalidated canonical receipts to the TS-owned exact Turn gate.""" try: - result = effect_runtime_result("work_item.replan_semantics.project", { + result = project_replan_semantics({ "operation": "turn_transition", "guard_scoped": guard_scoped, "selected_obligation_id": selected_obligation_id, "transition_candidates": transition_candidates, @@ -482,7 +483,7 @@ def semantic_delta_from_writeback( if vision: vision["state"] = normalize_goal_vision_state(vision.get("state")) try: - result = effect_runtime_result("work_item.replan_semantics.project", { + result = project_replan_semantics({ "operation": "qualify", "obligation": dict(obligation), "observation_delta": observation_delta, "agent_vision": vision, }) diff --git a/loopx/control_plane/work_items/replan_semantics_codec.py b/loopx/control_plane/work_items/replan_semantics_codec.py new file mode 100644 index 0000000000..5db59f2e16 --- /dev/null +++ b/loopx/control_plane/work_items/replan_semantics_codec.py @@ -0,0 +1,15 @@ +"""Local snapshot transport for the existing TypeScript replan decision owner.""" +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from ..effect_runtime import effect_runtime_result + + +def project_replan_semantics(params: Mapping[str, Any]) -> dict[str, Any]: + # Lifecycle recovery carries canonical rows, including retained evidence; + # both its input and result can exceed the ordinary RPC wire budget. + return effect_runtime_result( + "work_item.replan_semantics.project", params, large_local_snapshot=True, + ) diff --git a/tests/control_plane/test_lifecycle_guard_reentry.py b/tests/control_plane/test_lifecycle_guard_reentry.py index ffae88862c..fed7184930 100644 --- a/tests/control_plane/test_lifecycle_guard_reentry.py +++ b/tests/control_plane/test_lifecycle_guard_reentry.py @@ -12,8 +12,10 @@ @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("later_user_gate", [False, True]) +@pytest.mark.parametrize("large_history", [False, True]) def test_hidden_completed_source_recovers_its_original_guard_after_new_work( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, later_user_gate: bool, + large_history: bool, ) -> None: original_id = "todo_original_lifecycle" completion_key = "local_completion_" + sha256( @@ -32,6 +34,7 @@ def test_hidden_completed_source_recovers_its_original_guard_after_new_work( # Enough newer terminal rows to move the original outside display budgets. recent = [{**completed, "todo_id": f"todo_later_{i}", "index": i + 2, "text": f"Accepted independent source {i}.", "no_followup": True, + "note": "Synthetic retained evidence. " + ("x" * 96_000 if large_history else ""), "completion_continuation": "no_followup", "completion_turn_key": f"later-{i}", "completed_at": "2026-08-02T00:00:00Z", "updated_at": "2026-08-02T00:00:00Z"} for i in range(24)] @@ -54,8 +57,9 @@ def test_hidden_completed_source_recovers_its_original_guard_after_new_work( closed = call("todo", "complete", "--goal-id", GOAL, "--todo-id", original_id, "--agent-id", AGENT, "--completion-identity-key", completion_key, - "--no-follow-up", "--note", "The finite maintenance outcome was independently accepted.") + "--no-follow-up", "--reason", "The finite maintenance outcome was independently accepted.") assert closed["completion_recovery"] == "lifecycle_reentry_terminal_closeout" + assert call("todo", "list", "--goal-id", GOAL, "--todo-id", original_id)["todo"]["note"] == completed["note"] # A later human request owns a different live frontier, not this old review. later = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT, "--text", "Validate the newly requested independent source.", @@ -73,6 +77,8 @@ def test_hidden_completed_source_recovers_its_original_guard_after_new_work( assert recovered["autonomous_replan_obligation"]["resolution_mode"] == "receipt_bound_replan_settlement" assert recovered["replan_action_packet"]["decision"] == "vision_outcome_checkpoint_required" assert recovered["replan_action_packet"]["semantic_delta"] is None + if large_history: + assert len(json.dumps(recent).encode()) > 2 * 1024 * 1024 denied = call("quota", "spend-slot", "--goal-id", GOAL, "--agent-id", AGENT, "--codex-app", "--turn-instance-id", turn, "--replan-obligation-id", obligation["obligation_id"], "--source", "heartbeat", "--execute", expected_code=1) diff --git a/tests/control_plane/test_replan_semantics_transport.py b/tests/control_plane/test_replan_semantics_transport.py new file mode 100644 index 0000000000..8089b057ca --- /dev/null +++ b/tests/control_plane/test_replan_semantics_transport.py @@ -0,0 +1,55 @@ +"""Complete replan facts use the private transport, never a display window.""" +from __future__ import annotations + +import tempfile +from pathlib import Path + +import pytest + +from loopx.control_plane import effect_runtime +from loopx.control_plane.work_items.replan_semantics_codec import project_replan_semantics + + +@pytest.fixture +def isolated_runtime(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + for name in ("TMPDIR", "TEMP", "TMP"): + monkeypatch.setenv(name, str(tmp_path)) + monkeypatch.setattr(tempfile, "tempdir", str(tmp_path)) + yield tmp_path + effect_runtime.effect_runtime_result("runtime.shutdown", {}, retry_safe=False) + + +@pytest.mark.parametrize("large", [False, True]) +def test_lifecycle_source_preserves_complete_input_and_output(isolated_runtime: Path, large: bool): + item = { + "todo_id": "todo_accepted_source", "index": 1, + "status": "done", "done": True, "task_class": "advancement_task", + "claimed_by": "fixture-agent", "last_actor_agent_id": "fixture-agent", + "completed_at": "2026-08-01T00:00:00Z", "updated_at": "2026-08-01T00:00:00Z", + "completion_turn_key": "original-completion", + "no_followup": True, "completion_continuation": "no_followup", + "completion_recovery": "lifecycle_reentry_terminal_closeout", + "note": "x" * (effect_runtime.MAX_REQUEST_BYTES + 128) if large else "Accepted source.", + } + params = {"operation": "lifecycle_source", "agent_id": "fixture-agent", + "guard_recorded_at": "2026-08-01T00:00:00.500Z", "items": [item]} + if large: + # Registration is not a global bypass: ordinary requests still fail. + with pytest.raises(effect_runtime.EffectRuntimeRejected, match="request is oversized"): + effect_runtime.effect_runtime_result("work_item.replan_semantics.project", params) + else: + assert effect_runtime.effect_runtime_result("work_item.replan_semantics.project", params) == { + "items": [item], "settled": True, + } + assert project_replan_semantics(params) == {"items": [item], "settled": True} + assert list(isolated_runtime.glob("loopx-effect-*/params.json")) == [] + assert list(isolated_runtime.glob("loopx-effect-*/response.json")) == [] + + +def test_rejected_large_projection_removes_private_files(isolated_runtime: Path): + with pytest.raises(effect_runtime.EffectRuntimeRejected): + project_replan_semantics({"operation": "unsupported", + "items": [{"note": "x" * (effect_runtime.MAX_REQUEST_BYTES + 128)}]}) + assert list(isolated_runtime.glob("loopx-effect-*/params.json")) == [] + assert list(isolated_runtime.glob("loopx-effect-*/response.json")) == [] + assert effect_runtime.effect_runtime_result("runtime.ping", {})["ready"] is True diff --git a/tests/control_plane_ts/effect_runtime_snapshot.test.ts b/tests/control_plane_ts/effect_runtime_snapshot.test.ts index 079fe4c99f..75aeb75fcf 100644 --- a/tests/control_plane_ts/effect_runtime_snapshot.test.ts +++ b/tests/control_plane_ts/effect_runtime_snapshot.test.ts @@ -6,6 +6,7 @@ import {join} from "node:path"; import {test} from "node:test"; import { openPrivateResponseSink, readPrivateJsonSnapshot, writePrivateResponse, + MAX_LOCAL_SNAPSHOT_BYTES, } from "../../loopx/control_plane/effect_runtime_snapshot.ts"; test("local snapshot accepts exact private bytes and rejects altered or unsafe references", async () => { @@ -21,6 +22,8 @@ test("local snapshot accepts exact private bytes and rejects altered or unsafe r /snapshot unavailable or invalid/); await assert.rejects(readPrivateJsonSnapshot({...ref, byte_count: bytes.length - 1}), /snapshot unavailable or invalid/); + await assert.rejects(readPrivateJsonSnapshot({...ref, byte_count: MAX_LOCAL_SNAPSHOT_BYTES + 1}), + /snapshot unavailable or invalid/); // Windows has no POSIX mode denial: the private-mode rejection is only // observable where process.getuid exists. if (process.getuid) { From e1332fad0c2f47bda163cadf8bdeb4fd9d42252e Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 14:18:24 +0800 Subject: [PATCH 2/2] docs: qualify complete replan fact transport boundary Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/typescript-control-plane-migration-v0.md | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 81b52f9dba..a08faf325e 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1613,10 +1613,18 @@ travel through a private, digest-bound local file reference. The same reducer checks all rows, lane-scoped ACKs and retry identities; neither the RPC budget nor a display window permits history truncation. Missing, altered or malformed snapshots fail before a decision, and the adapter removes temporary files on -return or rejection. This repairs a T3/S2 settlement blocker and supplies S7/R7 +return or rejection. The replan semantic projection also uses the existing +same-UID, digest-bound request/response snapshot channel for complete lifecycle +and receipt-recovery facts. Its callers share one transport adapter; lifecycle, +stride and receipt decisions remain in the same TypeScript owner. Ordinary RPCs +retain their 2 MiB limit and local snapshots retain their 64 MiB bound. File/SQLite +CLI recovery covers full facts above the wire budget, later independent work and +user gates, durable refresh and replay, and exactly one quota debit. This repairs +a T3/S2 settlement blocker and supplies S7/R7 wire-growth evidence; it does not make full-history parsing constant-memory or -qualify distributed execution. Cursor/checkpoint reduction remains a measured -follow-up with complete-source parity, not a second Python policy. See the +qualify installed-host adoption or distributed execution. Cursor/checkpoint +reduction remains a measured follow-up with complete-source parity, not a second +Python policy. See the [history decision evidence](ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.md). **Checkpoint read-context transport.** A long-lived Goal can exceed the Effect