fix(replan): carry complete lifecycle facts through private snapshots

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

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",

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",

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,

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,

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,

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,
         })

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,
+    )

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)

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

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) {