fix(context): publish current recovery before optional planning (#7628)

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>

Author: loopx-agent

docs/reference/required-work-context.md

@@ -519,6 +519,11 @@ as a substitute. Both paths establish a new chain only after a complete verified
 full read. Ordinary tools and turn endings do not invalidate it. An interrupted
 pending delivery requires full recovery. Repeated reset events are idempotent.
 Recovery reads the same immutable capture without re-admission or spend.
+The adapter commits the current complete capture reference and pending delivery
+under the existing session lock before optional material planning or encoding.
+A process exit during planning therefore recovers the current source rather
+than its predecessor. The planner reuses that archive; an unread current source
+still cannot establish a full native baseline.
 
 This deliberately replaces per-request complete-input proof with lifecycle
 assumptions. Downstream output loss or truncation not exposed to the hook, a
@@ -758,6 +763,9 @@ code-mode 父调用的保留输出须包含嵌套 reader hook 观察到的全部
 原捕获，不能用新 guard 替代。两条路径都只在完整且已校验的全量读取后建立新差分链。
 普通工具和正常轮次结束不无条件失效。尚未完成的交付被中断
 后须完整恢复。重复重置事件幂等；恢复读取同一不可变捕获，不重新准入或扣额。
+Adapter 在可选材料规划或编码前，先持既有 session 锁提交当前完整 capture 引用及
+pending delivery。规划过程中进程退出后，恢复指向当前来源而非上一份；planner
+复用同一 archive。尚未读取的当前来源仍不能建立完整 native baseline。
 
 这是以生命周期假设替换逐轮完整输入证明。Hook 未暴露的下游丢失／截断、漏掉失效事件，
 或在声明开关之外静默禁用 hooks，仍可能漏检；不把这些假设称为实测保留，也不新增审批门。

loopx/control_plane/turn_driver/codex_context.py

@@ -470,8 +470,9 @@ def register_native_context_view(directory: str | None, view: dict[str, Any]) ->
         _save(root, state)
 
 
-def _capture_required_context(payload: dict[str, Any], *, directory: Path,
-                              surface: str) -> tuple[dict[str, Any], dict[str, Any]]:
+def _capture_required_context(
+    payload: dict[str, Any], *, directory: Path, surface: str, archive: dict[str, Any] | None = None,
+) -> tuple[dict[str, Any], dict[str, Any]]:
     """Publish the archive first, then the model view selected by the typed owner.
 
     Only the required view may become a full/delta baseline. Detail pointers
@@ -480,8 +481,9 @@ def _capture_required_context(payload: dict[str, Any], *, directory: Path,
     from ..quota.turn_envelope import plan_progressive_context
     from ..effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError
 
-    archive = capture_context_projection(payload, directory=directory,
-        read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE)
+    if archive is None:
+        archive = capture_context_projection(payload, directory=directory,
+            read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE)
     try:
         plan = plan_progressive_context(payload, surface=surface)
     except (EffectRuntimeRemoteError, EffectRuntimeStartupError):
@@ -569,19 +571,29 @@ def project_codex_context(payload: dict[str, Any], *, directory: str | None, cap
                 if continuity_generation is not None:
                     scope["continuity_generation"] = continuity_generation
                 request = _native_context_request(state, scope)
-                required, recovery = _capture_required_context(payload, directory=capture_directory, surface=surface)
+                recovery = capture_context_projection(payload, directory=capture_directory,
+                    read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE)
                 page_bytes, output_bytes = native_context_page_budget()
-                recovery["command"] = context_capture_read_command(Path(recovery["path"]), recovery["sha256"],
-                    page_bytes=page_bytes, max_page_output_bytes=output_bytes)
-                state["recovery"] = {**{key: recovery[key] for key in ("path", "sha256", "command")},
-                    "goal_id": goal_id, "agent_id": agent_id,
-                    "total_bytes": Path(recovery["path"]).stat().st_size}
-                # Block reuse before optional wire IO, not after a first page
-                # or EOF. A codec/read interruption keeps this incomplete.
-                state["pending"] = {"path": recovery["path"], "sha256": recovery["sha256"],
-                    "total_bytes": state["recovery"]["total_bytes"], "ranges": {}}
-                _save(root, state)
+
+                def publish_recovery(reference: dict[str, Any]) -> None:
+                    reference["command"] = context_capture_read_command(Path(reference["path"]), reference["sha256"],
+                        page_bytes=page_bytes, max_page_output_bytes=output_bytes)
+                    state["recovery"] = {**{key: reference[key] for key in ("path", "sha256", "command")},
+                        "goal_id": goal_id, "agent_id": agent_id,
+                        "total_bytes": Path(reference["path"]).stat().st_size}
+                    state["pending"] = {"path": reference["path"], "sha256": reference["sha256"],
+                        "total_bytes": state["recovery"]["total_bytes"], "ranges": {}}
+                    _save(root, state)
+
+                # Publish the current complete source before optional planning
+                # or encoding can be interrupted. Never recover its predecessor.
+                publish_recovery(recovery)
                 recovery_committed = True
+                required, planned_recovery = _capture_required_context(payload, directory=capture_directory,
+                    surface=surface, archive=recovery)
+                if planned_recovery is not recovery:
+                    recovery = planned_recovery
+                    publish_recovery(recovery)
 
                 result = project_captured_context(required, directory=capture_directory,
                     request=request, captured_recovery=recovery,

tests/control_plane/test_saved_context_project.py

@@ -245,6 +245,64 @@ def fail_optional_wire(path, payload, **kwargs):
     assert json.loads(recovered) == original and path.read_bytes() == raw
 
 
+@pytest.mark.parametrize("source", ["compact", "resume"])
+def test_saved_projection_planner_process_exit_keeps_current_recovery(tmp_path, monkeypatch, source):
+    import subprocess
+    from loopx.control_plane.turn_driver import codex_context
+    from tests.control_plane.test_codex_context_lifecycle import setup, project, deliver, read_pages
+
+    native, captures = tmp_path / "native", tmp_path / "captures"
+    native.mkdir(mode=0o700)
+    captures.mkdir(mode=0o700)
+    event = setup(native, monkeypatch)
+    prior = {"ok": True, "goal_id": "fixture", "agent_identity": {"agent_id": "agent"},
+        "heartbeat_receipt": {"turn_instance_id": "prior-turn"}, "requirement": "Stop before publication."}
+    _, delivered = project(native, payload=prior)
+    deliver(native, event, read_pages(native, delivered))
+    assert codex_context._state(native)["candidate"]
+    prior_file = Path(delivered["recovery"]["path"])
+    prior_bytes = prior_file.read_bytes()
+
+    current = {**prior, "heartbeat_receipt": {"turn_instance_id": "current-turn"},
+        "requirement": "Retain this changed requirement and stop before deployment."}
+    original = tmp_path / "current.json"
+    raw = json.dumps(current).encode()
+    original.write_bytes(raw)
+    registry, runtime = tmp_path / "absent-registry.json", tmp_path / "absent-runtime"
+    script = (
+        "import os\n"
+        "from loopx.control_plane.quota import turn_envelope\n"
+        "def interrupt(*args, **kwargs): os._exit(37)\n"
+        "turn_envelope.plan_progressive_context = interrupt\n"
+        "from loopx.cli import main\n"
+        "raise SystemExit(main())\n"
+    )
+    arguments = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+        "quota", "context-project", "--capture-file", str(original),
+        "--capture-sha256", hashlib.sha256(raw).hexdigest(), "--goal-id", "fixture", "--agent-id", "agent",
+        "--turn-instance-id", "current-turn", "--context-delivery", "on", "--codex-app",
+        "--codex-context-dir", str(native), "--decision-output-root", str(captures)]
+    stopped = subprocess.run([sys.executable, "-c", script, *arguments], cwd=native,
+        env={**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[2])},
+        capture_output=True, text=True, timeout=30)
+    assert stopped.returncode == 37, stopped.stdout + stopped.stderr
+    current_files = list(captures.rglob("projection.json"))
+    assert len(current_files) == 1 and json.loads(current_files[0].read_bytes()) == current
+    journal = codex_context._state(native)
+    assert journal["recovery"]["path"] == str(current_files[0])
+    assert journal["pending"]["ranges"] == {}
+
+    resumed = codex_context.codex_context_hook(str(native), {**event, "source": source})
+    assert journal["recovery"]["sha256"] in resumed["hookSpecificOutput"]["additionalContext"]
+    assert "candidate" not in codex_context._state(native)
+    code, page = _run_cli(registry, runtime, "quota", "context-read", "--codex-context-dir", str(native),
+        "--goal-id", "fixture", "--agent-id", "agent", cwd=native)
+    assert code == 0 and page["eof"] and json.loads(page["content"]) == current
+    assert page["sha256"] == journal["recovery"]["sha256"]
+    assert prior_file.read_bytes() == prior_bytes and original.read_bytes() == raw
+    assert not registry.exists() and not runtime.exists()
+
+
 def test_saved_projection_recovery_failure_returns_typed_error_and_current_reference(
     tmp_path, monkeypatch, capsys,
 ):