From 2ea61dbf37d08096cbc37d9b246b9cf00355f722 Mon Sep 17 00:00:00 2001 From: song Date: Sun, 25 Oct 2026 11:46:04 +0800 Subject: [PATCH] fix(context): keep fallback recovery bound to the current capture Persist current full recovery before optional projection and candidate updates. Invalidate stale native selectors on publication failure or same-session context loss, and retain the original receipt and capture in typed CLI errors. Cover oversized warm RPCs and IO/identity faults through compact and resume. Clarify the enabled recovery failure contract in the host documentation. Related to #7196. Signed-off-by: song --- benchmark/runtime/RUNTIME.md | 8 +- docs/heartbeat-automation-prompt.md | 19 +- loopx/cli_commands/quota.py | 16 +- loopx/cli_commands/quota_capture.py | 25 ++- .../turn_driver/codex_context.py | 127 ++++++++----- .../turn_driver/decision_capture.py | 29 ++- .../test_codex_context_lifecycle.py | 172 +++++++++++++++++- .../test_context_delivery_cli.py | 28 ++- .../test_quota_decision_capture.py | 59 ++++++ .../test_saved_context_project.py | 101 +++++++++- 10 files changed, 504 insertions(+), 80 deletions(-) diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 2680162dbc..fe37dac73c 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -114,8 +114,12 @@ native tool presents the saved decision using the same current-input observer; no guard, checkpoint, admission or debit is repeated. Cold delivery supplies the full body and its target; qualified warm delivery supplies the exact delta. The small response points to bounded reads and an original full-read fallback. -No native setup, invalid cache or uncertain retention keeps full input. A bad -source hash/scope is refused before publication. This does not renew decision +Uncertain retention keeps full input. If enabled native preparation cannot +establish safe recovery, the command returns a typed transport failure with the +original full-read reference and receipt; use that explicit reference without +rerunning admission. An optional cache write after recovery is established can +still fall back to the same full input. A bad source hash/scope is refused before +publication. This does not renew decision freshness or establish autonomous solver benefit. Default-off preloading is unchanged. Include the projection command, descriptor, recovery and page reads in subsequent total-cost comparisons. diff --git a/docs/heartbeat-automation-prompt.md b/docs/heartbeat-automation-prompt.md index 7788a2ee6b..b8e4cb7246 100644 --- a/docs/heartbeat-automation-prompt.md +++ b/docs/heartbeat-automation-prompt.md @@ -439,7 +439,15 @@ run-mode declarations are configuration errors: the CLI exits nonzero with a reason before admission or transport writes, including in native hooks. Correct the declaration or select `off`; there is no owner question or new approval. In a supported run mode, missing or invalid retained-input evidence still -recovers the same original full capture automatically. `off` creates no native journal, hook configuration or input +delivers the same original full capture. The current recovery reference is saved +before optional delta computation, so an oversized warm request or failed +candidate update cannot select an older capture after compaction or resume. +If the current recovery cannot be saved, the old session lookup is invalidated; +use the explicit original capture reference. If preparation fails before that +can be established, or invalidation also fails, the CLI reports +`quota_decision_transport_failed` with the saved decision and receipt. Do not +rerun admission to recover a presentation failure. +`off` creates no native journal, hook configuration or input observation; directories alone never turn it on. This declaration check is not host authentication, execution authority or proof of retained input. Every warm delta still needs independently verified complete current-input evidence. @@ -504,9 +512,12 @@ loopx --format json quota context-project --context-delivery on --codex-app \ The JSON response supplies `mode`, the digest-pinned read `command`, the target when available, and an original full `recovery` reference. Cold mode delivers complete context plus its target; a warm delta still requires actual current-input -proof. Follow every page and reconstruct before dependent work. Invalid optional -transport/cache falls back to that same full capture; invalid source hash or -Goal/Agent/Turn refuses publication. This is integrity-checked presentation of +proof. Follow every page and reconstruct before dependent work. Optional codec +or delivery-cache failure after recovery publication falls back to that same +full capture. Enabled native preparation or unsafe recovery publication fails +with `ok: false`, `quota_decision_transport_failed`, and the original `recovery` +reference and receipt; follow that explicit reference instead of a session lookup. +Invalid source hash or Goal/Agent/Turn refuses publication. This is integrity-checked presentation of the caller's snapshot, not fresh authority, registry lookup, installation, admission or accounting. The original file stays unchanged; generated caches remain private. A missing projection command can use `quota context-read` with diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index df42d85377..045c0beb8e 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -492,6 +492,7 @@ def _emit_quota_result( agent = payload.get("agent_identity") decision_identity = (str(payload.get("goal_id") or ""), str(agent.get("agent_id") or "") if isinstance(agent, Mapping) else str(args.agent_id or "")) + captured_payload = payload payload = _project_quota_cli_payload( payload, args, detail_sections, context.scheduler_context if context is not None else None, @@ -530,8 +531,19 @@ def _emit_quota_result( capture_directory=capture_directory, goal_id=decision_identity[0], agent_id=decision_identity[1], surface="turn_envelope" if getattr(args, "turn_envelope", False) else "quota_packet", capture_id=capture_id) - except (OSError, ValueError): - pass # Lost optional capture: keep the admitted full payload, never rerun its guard. + except (QuotaDecisionTransportError, OSError, ValueError) as exc: + if not isinstance(exc, QuotaDecisionTransportError): + exc = QuotaDecisionTransportError( + "Current context recovery could not be prepared. Read the saved " + "original decision and receipt; do not rerun admission." + ) + # The original receipt survives even when the selected presentation + # is a Turn envelope without the quota packet's receipt fields. + failure = quota_decision_transport_failure_payload( + args, captured_payload, exc, capture_directory=capture_directory, + ) + print_payload(failure, args.format, _quota_renderer(args)) + return 1 print_payload(payload, args.format, _quota_renderer(args)) return 0 if succeeded else 1 diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 42c6d99923..2cc83c7e63 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -117,8 +117,9 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]: """Present an immutable already-decided quota capture, without authority IO. Publication is delayed until this native tool runs, so the same actual-input - observer can qualify a resumed baseline. Any optional transport failure - returns the original full read reference, never a replacement guard. + observer can qualify a resumed baseline. Optional codec failure returns the + original full read reference. An unsafe recovery selector fails explicitly + with that same reference, never a replacement guard. Invalid run-mode opt-ins fail before that optional transport path. """ try: @@ -144,6 +145,7 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]: full_reference = dict(reference) mode = "full" target = None + native_prepared = False try: from ..control_plane.turn_driver.codex_context import codex_context_directory, project_codex_context from ..control_plane.turn_driver.decision_capture import write_decision_capture @@ -162,6 +164,7 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]: projected = project_codex_context(payload, directory=directory, capture_directory=cache, goal_id=args.goal_id, agent_id=args.agent_id, surface="quota_packet", capture_id=hashlib.sha256(raw).hexdigest()) + native_prepared = True wire = cache / "delivery.json" write_decision_capture(wire, projected) digest = hashlib.sha256(wire.read_bytes()).hexdigest() @@ -174,8 +177,22 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]: # Deliver that exact full body plus its target in this small # descriptor, rather than paging another wrapper around it. reference = {key: projected["recovery"][key] for key in ("path", "sha256", "command")} - except (OSError, ValueError, TypeError, RecursionError): - pass # Optional presentation unavailable: the original full reference is already verified. + except (QuotaDecisionTransportError, OSError, ValueError, TypeError, RecursionError) as exc: + if isinstance(exc, QuotaDecisionTransportError) or (activation["enabled"] and not native_prepared): + if not isinstance(exc, QuotaDecisionTransportError): + exc = QuotaDecisionTransportError( + "Current context recovery could not be prepared. Read the saved " + "original decision and receipt; do not rerun admission." + ) + return {"ok": False, "error_code": exc.error_code, + "reason": " ".join(str(exc).split())[:500], + "heartbeat_receipt": receipt, "capture_sha256": args.capture_sha256, + "recovery": full_reference, + "instruction": "Context presentation was not published. Read recovery.command for the " + "original captured decision and preserve its Turn receipt; do not rerun admission. " + "Authority and freshness are unchanged. " + CONTEXT_CAPTURE_READ_GUIDANCE} + # Off is inert; after native preparation an optional wire write can fall + # back to the original reference without leaving an older selector live. return {"ok": True, "mode": mode, **reference, **({"target": target} if target else {}), "recovery": full_reference, "capture_sha256": args.capture_sha256, diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index 092a9983b2..cca52b6944 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/loopx/control_plane/turn_driver/codex_context.py @@ -15,7 +15,9 @@ import stat import uuid -from .decision_capture import write_decision_capture, project_captured_context, CONTEXT_CAPTURE_READ_GUIDANCE +from .decision_capture import ( + write_decision_capture, capture_context_projection, project_captured_context, CONTEXT_CAPTURE_READ_GUIDANCE, +) # Native adapter activation reuses scheduler-owned host declarations. These # settings are presentation opt-ins, never host authentication or authority. @@ -302,56 +304,93 @@ def _trace_bundle(session: str, tool_id: str) -> str | None: return matches[0] if len(matches) == 1 else None +def _discard_recovery(root: Path) -> None: + """Invalidate only this adapter's journal while its private root is locked.""" + path = root / "session.json" + try: + info = path.lstat() + except FileNotFoundError: + return + if info.st_uid != os.getuid() or not (stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode)): + raise ValueError("context recovery journal is not an owned file") + # unlink never follows a rejected symlink or deletes its target. + path.unlink() + + +def _native_context_request(state: dict[str, Any], scope: dict[str, Any]) -> dict[str, Any]: + """A saved candidate is reusable only with independent current-input proof.""" + from .codex_input_trace import codex_context_in_active_input + + request: dict[str, Any] = {"scope": scope} + candidate = state.get("candidate") + if isinstance(candidate, dict) and isinstance(state.get("active_call"), str): + try: + baseline = _read(Path(candidate["path"])) + trace = _trace_bundle(scope["session_id"], state["active_call"]) + if trace: + observation = dict(trace_bundle=trace, tool_call_id=state["active_call"], + capture_file=candidate["capture_file"], capture_sha256=candidate["capture_sha256"]) + if codex_context_in_active_input(observation, baseline=baseline, session_id=scope["session_id"]): + request.update(baseline=baseline, retained={"complete": True, "target": baseline["target"]}, + native_input=observation) + except (OSError, ValueError, KeyError, TypeError, RecursionError): + pass # Missing candidate affects reuse, never the current recovery reference. + return request + + def project_codex_context(payload: dict[str, Any], *, directory: str | None, capture_directory: Path, goal_id: str, agent_id: str, surface: str, capture_id: str) -> dict[str, Any]: - """Existing CLI consumer, sharing the rendezvous projector and recovery IO. + """Commit current recovery before optional projection or candidate updates. - Serialization prevents an invalidation from moving the epoch during - preparation. The current capture reference is durable before publication, - so a later compact event recovers those exact bytes. Full candidates are - never advanced from file existence, hook success or a saved delivery flag. + Serialization prevents invalidation from moving the epoch during preparation. + Full fallback must recover this capture or refuse lookup, never an older one. + Candidates still need independent current-input proof before reuse. """ - from .codex_input_trace import codex_context_in_active_input + from ..quota.error_codes import QuotaDecisionTransportError if directory is None: return payload try: with _locked(directory) as root: - state = _state(root) - session, epoch = state.get("session_id"), state.get("context_epoch") - if (not isinstance(session, str) or not isinstance(epoch, str) - or not session or not epoch or session != os.environ.get("CODEX_THREAD_ID") - or Path(state.get("cwd", "")).resolve() != Path.cwd().resolve()): - return payload - scope = dict(goal_id=goal_id, agent_id=agent_id, session_id=session, context_epoch=epoch) - request: dict[str, Any] = {"scope": scope} - candidate = state.get("candidate") - if isinstance(candidate, dict) and isinstance(state.get("active_call"), str): - try: - baseline = _read(Path(candidate["path"])) - trace = _trace_bundle(session, state["active_call"]) - if trace: - observation = dict(trace_bundle=trace, tool_call_id=state["active_call"], - capture_file=candidate["capture_file"], capture_sha256=candidate["capture_sha256"]) - if codex_context_in_active_input(observation, baseline=baseline, session_id=session): - request.update(baseline=baseline, retained={"complete": True, "target": baseline["target"]}, - native_input=observation) - except (OSError, ValueError, KeyError, TypeError, RecursionError): - pass # Missing candidate: publish a new full capture, never an old recovery ref. - result = project_captured_context(payload, directory=capture_directory, request=request, - goal_id=goal_id, agent_id=agent_id, surface=surface, capture_id=capture_id) - if result.get("schema_version") != "context_delivery_v1": + recovery_committed = False + try: + state = _state(root) + session, epoch = state.get("session_id"), state.get("context_epoch") + native_session = os.environ.get("CODEX_THREAD_ID") + if (not isinstance(session, str) or not session + or (native_session and session != native_session)): + return payload # Do not replace a different native session's recovery. + if (not native_session or not isinstance(state.get("cwd"), str) + or Path(state["cwd"]).resolve() != Path.cwd().resolve()): + # A later tool can restore cwd/identity and make an old + # selector eligible again, so invalidate before full fallback. + raise ValueError("context recovery requires the same native session and workspace") + if not isinstance(epoch, str) or not epoch: + raise ValueError("context recovery requires a native context epoch") + recovery = capture_context_projection(payload, directory=capture_directory) + state["recovery"] = {**{key: recovery[key] for key in ("path", "sha256", "command")}, + "goal_id": goal_id, "agent_id": agent_id} + _save(root, state) + recovery_committed = True + + scope = dict(goal_id=goal_id, agent_id=agent_id, session_id=session, context_epoch=epoch) + result = project_captured_context(payload, directory=capture_directory, + request=_native_context_request(state, scope), captured_recovery=recovery, + goal_id=goal_id, agent_id=agent_id, surface=surface, capture_id=capture_id) + if result.get("schema_version") == "context_delivery_v1" and result["mode"] == "full": + path = capture_directory / "presentation.json" + write_decision_capture(path, result) + state["candidate"] = dict(path=str(path), capture_file=recovery["path"], + capture_sha256=recovery["sha256"]) + _save(root, state) return result - recovery = result["recovery"] - state["recovery"] = {**{key: recovery[key] for key in ("path", "sha256", "command")}, - "goal_id": goal_id, "agent_id": agent_id} - if result["mode"] == "full": - path = capture_directory / "presentation.json" - write_decision_capture(path, result) - state["candidate"] = dict(path=str(path), capture_file=recovery["path"], - capture_sha256=recovery["sha256"]) - _save(root, state) - return result - except (OSError, ValueError, KeyError, TypeError, RecursionError): - # A broken optional adapter preserves the already admitted full result. - return payload + except (OSError, ValueError, KeyError, TypeError, RecursionError): + if not recovery_committed: + _discard_recovery(root) + return payload + except (OSError, ValueError, KeyError, TypeError, RecursionError) as exc: + # No safe fallback if the old selector cannot be invalidated under lock. + raise QuotaDecisionTransportError( + "Current context recovery could not be published or safely invalidated. " + "Read the saved original decision and receipt; do not rerun admission." + ) from exc diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index 79d078107d..b9bdc73c9a 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/loopx/control_plane/turn_driver/decision_capture.py @@ -115,9 +115,24 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0, pa "--offset", str(offset), "--page-bytes", str(page_bytes)]) +def capture_context_projection(payload: dict[str, Any], *, directory: Path) -> dict[str, Any]: + """Save the full projection independently of the optional presentation codec.""" + projection = directory / "projection.json" + write_decision_capture(projection, payload) + digest = hashlib.sha256(projection.read_bytes()).hexdigest() + return { + "path": str(projection), "sha256": digest, + "instruction": "Complete JSON for this original capture. If the baseline is missing, compacted, " + "truncated or uncertain, use this full projection before dependent work. " + + CONTEXT_CAPTURE_READ_GUIDANCE, + "command": context_capture_read_command(projection, digest), + } + + def project_captured_context( payload: dict[str, Any], *, directory: Path | None, request: dict[str, Any] | None, goal_id: str, agent_id: str, surface: str, capture_id: str, + captured_recovery: dict[str, Any] | None = None, ) -> dict[str, Any]: """Adapt an admitted capture to the shared TS presentation owner. @@ -126,6 +141,8 @@ def project_captured_context( delivery. Unsupported ordinary CLI clients never supply this request. Every opt-in attempt saves the complete projected JSON for same-capture recovery; neither malformed requests nor optimization failures rerun the guard. + A native caller may pass its already-published recovery reference, never a + reference supplied in the external request. """ if request is None or directory is None: return payload @@ -133,8 +150,8 @@ def project_captured_context( from ..quota.turn_envelope import prepare_context_delivery try: - projection = directory / "projection.json" - write_decision_capture(projection, payload) + recovery = captured_recovery if captured_recovery is not None else capture_context_projection( + payload, directory=directory) scope = request.get("scope") if not isinstance(scope, dict) or scope.get("goal_id") != goal_id or scope.get("agent_id") != agent_id: return payload @@ -155,13 +172,7 @@ def project_captured_context( delivery = prepare_context_delivery(payload, scope=scope, capture_id=capture_id, surface=surface, baseline=request.get("baseline"), retained=retained) - delivery["recovery"] = { - "path": str(projection), "sha256": hashlib.sha256(projection.read_bytes()).hexdigest(), - "instruction": "Complete JSON for this original capture. If the baseline is missing, compacted, " - "truncated or uncertain, use this full projection before dependent work. " - + CONTEXT_CAPTURE_READ_GUIDANCE, - } - delivery["recovery"]["command"] = context_capture_read_command(projection, delivery["recovery"]["sha256"]) + delivery["recovery"] = recovery return delivery except (OSError, ValueError, TypeError, EffectRuntimeRemoteError, EffectRuntimeStartupError): # The original full payload is already available. An unavailable optional diff --git a/tests/control_plane/test_codex_context_lifecycle.py b/tests/control_plane/test_codex_context_lifecycle.py index c0f9bfc610..bb84d98551 100644 --- a/tests/control_plane/test_codex_context_lifecycle.py +++ b/tests/control_plane/test_codex_context_lifecycle.py @@ -186,16 +186,160 @@ def unavailable(*args, **kwargs): result = context.codex_context_hook(str(tmp_path), {**event, "source": "compact"}) assert wire["recovery"]["sha256"] in result["hookSpecificOutput"]["additionalContext"] assert (tmp_path / "session.json").read_bytes() == before - # A primary store failure during publication returns the whole admitted - # payload. It cannot leak an untracked candidate delta. - payload, full = project(tmp_path, "two") - assert full == payload -@pytest.mark.parametrize("damage", ["missing", "corrupt", "symlink", "public", "other_workspace"]) -def test_invalid_journal_falls_back_to_original_full(tmp_path, monkeypatch, damage): +def _assert_recovery_after_fallback(root, event, source, capture): + import argparse + import hashlib + from loopx.cli_commands.quota_capture import read_context_capture + + hook = context.codex_context_hook(str(root), {**event, "source": source}) + args = argparse.Namespace(codex_context_dir=str(root), goal_id="fixture", agent_id="agent", + offset=0, page_bytes=8192) + page = read_context_capture(args) + if capture is None: + assert hook == {} and page["ok"] is False and "content" not in page + return + raw = capture.read_bytes() + digest = hashlib.sha256(raw).hexdigest() + assert page["ok"] is True and page["sha256"] == digest + assert page["content"].encode() == raw[:len(page["content"].encode())] + assert digest in hook["hookSpecificOutput"]["additionalContext"] + + +@pytest.mark.parametrize("source", ["compact", "resume"]) +def test_oversized_warm_request_keeps_current_full_recovery(tmp_path, monkeypatch, source): + from test_codex_input_retention import fixture + from loopx.control_plane.effect_runtime import EffectRuntimeRejected, MAX_REQUEST_BYTES + from loopx.control_plane.quota.turn_envelope import prepare_context_delivery + from loopx.control_plane.turn_driver.codex_input_trace import codex_context_in_active_input + + event = setup(tmp_path, monkeypatch) + before = {"ok": True, "goal_id": "fixture", "stop": "Stop before publication.", + "plan_summary": {"requirements": "x" * (MAX_REQUEST_BYTES // 2 + 8192)}} + after = {**before, "stop": "Stop before publication or deployment."} + _, first = project(tmp_path, payload=before) + original = Path(first["recovery"]["path"]).read_bytes() + assert len(original) < MAX_REQUEST_BYTES + traces = tmp_path / "traces" + traces.mkdir() + observation, _, _, _ = fixture(traces) + bundle = Path(observation["trace_bundle"]) + # Actual production verifier consumes these synthetic native trace files; + # neither the projector nor the RPC size limit is mocked. + request = bundle / "payloads/input.json" + request.write_text(json.dumps({"input": [{"type": "function_call_output", "output": json.dumps(first)}]})) + assert request.stat().st_size < MAX_REQUEST_BYTES + current_bundle = traces / "trace-live" + bundle.rename(current_bundle) + observation.update(trace_bundle=str(current_bundle), capture_file=first["recovery"]["path"], + capture_sha256=first["recovery"]["sha256"]) + assert codex_context_in_active_input(observation, baseline=first, session_id="session") + monkeypatch.setenv("CODEX_ROLLOUT_TRACE_ROOT", str(traces)) + context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PreToolUse", "tool_use_id": "guard"}) + scope = {key: first["target"][key] for key in ("goal_id", "agent_id", "session_id", "context_epoch")} + with pytest.raises(EffectRuntimeRejected) as rejected: + prepare_context_delivery(after, scope=scope, capture_id="two", surface="quota_packet", + baseline=first, retained={"complete": True, "target": first["target"]}) + assert rejected.value.diagnostic_code == "request_too_large" + _, delivered = project(tmp_path, "two", payload=after) + assert delivered.get("context", delivered) == after + current_capture = tmp_path / "two/projection.json" + assert current_capture.stat().st_size < MAX_REQUEST_BYTES + assert json.loads(current_capture.read_bytes()) == after + _assert_recovery_after_fallback(tmp_path, event, source, current_capture) + assert Path(first["recovery"]["path"]).read_bytes() == original + + +@pytest.mark.parametrize("source", ["compact", "resume"]) +@pytest.mark.parametrize("failed_write", ["recovery", "candidate"]) +def test_journal_failure_never_recovers_previous_capture(tmp_path, monkeypatch, source, failed_write): + event = setup(tmp_path, monkeypatch) + _, first = project(tmp_path) + original = Path(first["recovery"]["path"]).read_bytes() + current_capture = tmp_path / "two/projection.json" + original_save, failures = context._save, [] + + def fail_current_write(root, state): + current_recovery = state.get("recovery", {}).get("sha256") not in (None, first["recovery"]["sha256"]) + current_candidate = state.get("candidate", {}).get("capture_file") == str(current_capture) + if current_recovery and (failed_write == "recovery" or current_candidate): + failures.append(failed_write) + raise OSError("synthetic current publication unavailable") + original_save(root, state) + + with monkeypatch.context() as publication: + publication.setattr(context, "_save", fail_current_write) + payload, delivered = project(tmp_path, "two", payload={"ok": True, "goal_id": "fixture", + "requirements": "Preserve the changed source and stop before deployment."}) + assert failures and delivered.get("context", delivered) == payload + assert json.loads(current_capture.read_bytes()) == payload + _assert_recovery_after_fallback(tmp_path, event, source, + current_capture if failed_write == "candidate" else None) + assert Path(first["recovery"]["path"]).read_bytes() == original + + +@pytest.mark.parametrize("source", ["compact", "resume"]) +@pytest.mark.parametrize("caller", ["subdirectory", "missing_session", "foreign_session"]) +def test_fallback_identity_restoration_keeps_recovery_scoped(tmp_path, monkeypatch, source, caller): + event = setup(tmp_path, monkeypatch) + _, first = project(tmp_path) + original_capture = Path(first["recovery"]["path"]) + original = original_capture.read_bytes() + context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PreToolUse", "tool_use_id": "guard"}) + with monkeypatch.context() as native_tool: + if caller == "subdirectory": + subdirectory = tmp_path / "subdirectory" + subdirectory.mkdir() + native_tool.chdir(subdirectory) + elif caller == "missing_session": + native_tool.delenv("CODEX_THREAD_ID") + else: + native_tool.setenv("CODEX_THREAD_ID", "foreign-session") + payload, delivered = project(tmp_path, "two", payload={"ok": True, "goal_id": "fixture", + "requirements": "Changed source; stop before deployment."}) + assert delivered == payload + _assert_recovery_after_fallback(tmp_path, event, source, + original_capture if caller == "foreign_session" else None) + assert original_capture.read_bytes() == original + + +@pytest.mark.parametrize("failure", ["invalidation", "lock"]) +def test_unavailable_recovery_boundary_refuses_current_publication(tmp_path, monkeypatch, failure): + from loopx.control_plane.quota.error_codes import QuotaDecisionTransportError + setup(tmp_path, monkeypatch) + _, first = project(tmp_path) + journal = tmp_path / "session.json" + original = journal.read_bytes() + unlink = Path.unlink + + def unavailable(*args, **kwargs): + raise OSError("synthetic journal publication unavailable") + + def cannot_invalidate(path, *args, **kwargs): + if path == journal: + raise OSError("synthetic journal invalidation unavailable") + return unlink(path, *args, **kwargs) + + with monkeypatch.context() as publication: + publication.setattr(context, "_save", unavailable) + publication.setattr(Path, "unlink", cannot_invalidate) + if failure == "lock": + publication.setattr(context, "_locked", unavailable) + with pytest.raises(QuotaDecisionTransportError): + project(tmp_path, "two", payload={"ok": True, "goal_id": "fixture", "requirements": "Current source."}) + assert journal.read_bytes() == original + assert json.loads(Path(first["recovery"]["path"]).read_bytes())["requirements"] == "Preserve final stop condition." + + +@pytest.mark.parametrize("damage", ["missing", "corrupt", "symlink", "public", "other_workspace", + "missing_epoch", "invalid_epoch", "read_failure", "projection_failure"]) +def test_invalid_journal_falls_back_to_original_full(tmp_path, monkeypatch, damage): + event = setup(tmp_path, monkeypatch) + project(tmp_path) path = tmp_path / "session.json" + before = path.read_bytes() if damage == "missing": path.unlink() elif damage == "corrupt": @@ -210,8 +354,22 @@ def test_invalid_journal_falls_back_to_original_full(tmp_path, monkeypatch, dama state = context._state(tmp_path) state["cwd"] = str(tmp_path.parent) context._save(tmp_path, state) - payload, full = project(tmp_path) + elif damage in {"missing_epoch", "invalid_epoch"}: + state = context._state(tmp_path) + state.pop("context_epoch") + if damage == "invalid_epoch": + state["context_epoch"] = 3 + context._save(tmp_path, state) + with monkeypatch.context() as failed: + if damage in {"read_failure", "projection_failure"}: + def unavailable(*args, **kwargs): + raise OSError("synthetic pre-publication IO unavailable") + failed.setattr(context, "_state" if damage == "read_failure" else "capture_context_projection", unavailable) + payload, full = project(tmp_path, "two") assert full == payload + if damage == "symlink": + assert other.read_bytes() == before + _assert_recovery_after_fallback(tmp_path, event, "compact", None) def test_real_hook_cli_is_transport_only_without_registry_or_installation(tmp_path): diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py index 9965ced4d8..908d6bb4f9 100644 --- a/tests/control_plane/test_context_delivery_cli.py +++ b/tests/control_plane/test_context_delivery_cli.py @@ -116,6 +116,8 @@ def test_real_cli_host_rendezvous_sends_delta_and_keeps_same_capture_full(tmp_pa before_source = state.read_bytes() root = tmp_path / "captures" root.mkdir() + native_directory = tmp_path / "unobserved-native-state" + native_directory.mkdir(mode=0o700) scope = dict(goal_id=GOAL_ID, agent_id=AGENT_ID, session_id="host-session", context_epoch="epoch-1") saved = [] @@ -151,7 +153,7 @@ def release(): "--turn-instance-id", "host-context-delivery", "--context-delivery", setting, "--decision-output-root", str(root), "--decision-handoff-socket", str(endpoint), - *(["--codex-context-dir", str(tmp_path / "unavailable-native-state")] if native_context else []), + *(["--codex-context-dir", str(native_directory)] if native_context else []), *(["--turn-envelope"] if surface == "turn_envelope" else []), cwd=project) path, raw = future.result(timeout=30) assert code == 0, wire @@ -251,7 +253,7 @@ def test_native_context_cli_without_retention_stays_full_and_preserves_receipt(t assert _spend_run_count(runtime) == 0 and state.read_bytes() == original -@pytest.mark.parametrize("native_failure", [False, True]) +@pytest.mark.parametrize("native_failure", [None, "projection", "directory"]) def test_native_projection_never_layers_host_projection(tmp_path, monkeypatch, native_failure): from loopx.cli import build_parser from loopx.cli_commands import quota @@ -259,7 +261,8 @@ def test_native_projection_never_layers_host_projection(tmp_path, monkeypatch, n args = build_parser().parse_args(["--format", "json", "quota", "should-run", "--goal-id", "fixture", "--agent-id", "worker", "--context-delivery", "on", "--codex-app", "--codex-context-dir", str(tmp_path)]) - original = {"ok": True, "goal_id": "fixture", "agent_identity": {"agent_id": "worker"}} + original = {"ok": True, "goal_id": "fixture", "agent_identity": {"agent_id": "worker"}, + "heartbeat_receipt": {"turn_instance_id": "original-turn", "status": "committed"}} monkeypatch.setattr(quota, "handoff_decision", lambda *args: ({"scope": "host-request"}, "digest")) monkeypatch.setattr(quota, "_project_quota_cli_payload", lambda payload, *args, **kwargs: payload) @@ -268,17 +271,30 @@ def competing(*args, **kwargs): def native(payload, **kwargs): assert payload == original - if native_failure: + if native_failure == "directory": + pytest.fail("failed native directory preparation must not reach projection") + if native_failure == "projection": raise OSError("native observation unavailable") return {"native_projection": payload} + def unavailable_directory(**kwargs): + raise OSError("native directory unavailable") + monkeypatch.setattr(quota, "project_captured_context", competing) monkeypatch.setattr(codex_context, "project_codex_context", native) + if native_failure == "directory": + monkeypatch.setattr(codex_context, "codex_context_directory", unavailable_directory) printed = [] assert quota._emit_quota_result(original, args, context=None, registry_path=tmp_path / "absent.json", heartbeat_turn_id=None, usage_quota_started=0, capture_directory=tmp_path, - detail_sections=frozenset(), print_payload=lambda value, *args: printed.append(value)) == 0 - assert printed == [original if native_failure else {"native_projection": original}] + detail_sections=frozenset(), print_payload=lambda value, *args: printed.append(value)) == (1 if native_failure else 0) + if native_failure: + assert len(printed) == 1 and printed[0]["ok"] is False + assert printed[0]["error_code"] == "quota_decision_transport_failed" + assert printed[0]["heartbeat_receipt"] == original["heartbeat_receipt"] + assert printed[0]["captured_decision"] == {"path": str(tmp_path / "decision.json")} + else: + assert printed == [{"native_projection": original}] assert json.loads((tmp_path / "decision.json").read_text()) == original diff --git a/tests/control_plane/test_quota_decision_capture.py b/tests/control_plane/test_quota_decision_capture.py index 993a345a72..58dafe2c15 100644 --- a/tests/control_plane/test_quota_decision_capture.py +++ b/tests/control_plane/test_quota_decision_capture.py @@ -571,6 +571,65 @@ def test_refused_selection_handoff_is_typed_and_keeps_the_committed_capture(tmp_ assert _heartbeat_receipt_count(runtime, TURN_ID) >= 1 +@pytest.mark.parametrize('provider', ['file', 'sqlite']) +@pytest.mark.parametrize('turn_envelope', [False, True]) +def test_native_recovery_failure_keeps_original_receipt_and_capture( + tmp_path, monkeypatch, capsys, provider, turn_envelope, +): + from loopx.cli import main + from loopx.control_plane.quota.error_codes import QuotaDecisionTransportError + from loopx.control_plane.turn_driver import codex_context + from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime + from tests.control_plane.test_quota_settlement_cli import _spend_run_count + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path) + state = _configure_read_only_todo(project) + initialize_canonical_source_fixture(runtime, GOAL_ID, state_path=state, provider=provider) + original_state = state.read_bytes() + context, captures = tmp_path / 'native', tmp_path / 'captures' + context.mkdir(mode=0o700) + captures.mkdir(mode=0o700) + attempts = [] + + def refuse_recovery(payload, **kwargs): + saved = kwargs['capture_directory'] / 'decision.json' + attempts.append(saved.read_bytes()) + if turn_envelope: + assert payload['schema_version'] == 'loopx_turn_envelope_v0' + assert 'heartbeat_receipt' not in payload + raise QuotaDecisionTransportError( + 'Current context recovery could not be published or safely invalidated. ' + 'Read the saved original decision/receipt; do not rerun admission.' + ) + + monkeypatch.setattr(codex_context, 'project_codex_context', refuse_recovery) + args = ['--registry', str(registry), '--runtime-root', str(runtime), '--format', 'json', + 'quota', 'should-run', '--goal-id', GOAL_ID, '--agent-id', AGENT_ID, + '--todo-id', TODO_ID, '--turn-instance-id', TURN_ID, '--codex-app', + '--available-capability', 'shell', '--available-capability', 'filesystem_read', + '--context-delivery', 'on', '--codex-context-dir', str(context), + '--decision-output-root', str(captures)] + if turn_envelope: + args.append('--turn-envelope') + code = main(args) + failure = json.loads(capsys.readouterr().out) + assert code == 1 and failure['ok'] is False + assert failure['error_code'] == 'quota_decision_transport_failed' + assert failure['status'] == 'decision_transport_failed' + saved = list(captures.glob('*/decision.json')) + assert len(saved) == len(attempts) == 1 + assert saved[0].read_bytes() == attempts[0] + original = json.loads(attempts[0]) + assert original['ok'] is True + assert failure['heartbeat_receipt'] == original['heartbeat_receipt'] + assert failure['rollout_event'] == original['rollout_event'] + assert failure['captured_decision'] == {'path': str(saved[0])} + assert _heartbeat_receipt_count(runtime, TURN_ID) == 1 + assert _spend_run_count(runtime) == 0 + assert state.read_bytes() == original_state + + def test_ended_wake_handoff_endpoint_is_rejected_with_current_wake_remedy(tmp_path): from loopx.control_plane.turn_driver.decision_capture import validate_handoff_socket diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index c06c60f60f..3b4a66a6c7 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/tests/control_plane/test_saved_context_project.py @@ -45,7 +45,7 @@ def test_saved_projection_checks_original_before_any_publication(tmp_path, failu @pytest.mark.parametrize("root_kind", ["unsafe", "symlink", "missing"]) -def test_optional_cache_failure_returns_original_full_reference(tmp_path, root_kind): +def test_enabled_cache_preparation_failure_returns_error_with_original_reference(tmp_path, monkeypatch, root_kind): from loopx.cli import build_parser from loopx.cli_commands.quota_capture import project_saved_context_capture @@ -54,6 +54,13 @@ def test_optional_cache_failure_returns_original_full_reference(tmp_path, root_k "heartbeat_receipt": {"turn_instance_id": "saved-turn"}}).encode() path.write_bytes(raw) context.mkdir(mode=0o700) + stale = context / "session.json" + stale.write_text(json.dumps({"session_id": "native-session", "cwd": str(tmp_path), + "recovery": {"goal_id": "saved-goal", "agent_id": "saved-agent", + "path": str(tmp_path / "prior.json"), "sha256": "a" * 64}})) + stale_bytes = stale.read_bytes() + monkeypatch.setenv("CODEX_THREAD_ID", "native-session") + monkeypatch.chdir(tmp_path) if root_kind == "unsafe": root.mkdir(mode=0o755) root.chmod(0o755) @@ -65,9 +72,99 @@ def test_optional_cache_failure_returns_original_full_reference(tmp_path, root_k "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn", "--decision-output-root", str(root), "--context-delivery", "on", "--codex-app", "--codex-context-dir", str(context)]) result = project_saved_context_capture(args) + assert result["ok"] is False and result["error_code"] == "quota_decision_transport_failed" + assert result["recovery"]["path"] == str(path) and result["recovery"]["sha256"] == digest + assert "prior.json" not in json.dumps(result) + assert path.read_bytes() == raw and stale.read_bytes() == stale_bytes + + +def test_saved_projection_wire_failure_keeps_committed_current_recovery(tmp_path, monkeypatch): + from loopx.cli import build_parser + from loopx.cli_commands.quota_capture import project_saved_context_capture + from loopx.control_plane.turn_driver import codex_context, decision_capture + + path, root, context = tmp_path / "original.json", tmp_path / "cache", tmp_path / "native" + original = {"ok": True, "goal_id": "saved-goal", "agent_identity": {"agent_id": "saved-agent"}, + "heartbeat_receipt": {"turn_instance_id": "saved-turn"}, "requirement": "Current complete requirements."} + raw = json.dumps(original).encode() + path.write_bytes(raw) + root.mkdir(mode=0o700) + context.mkdir(mode=0o700) + monkeypatch.setenv("CODEX_THREAD_ID", "native-session") + monkeypatch.chdir(tmp_path) + codex_context.codex_context_hook(str(context), {"hook_event_name": "SessionStart", "source": "startup", + "session_id": "native-session", "cwd": str(tmp_path)}) + write_capture = decision_capture.write_decision_capture + + def fail_optional_wire(path, payload, **kwargs): + if path.name == "delivery.json": + raise OSError("optional wire unavailable") + return write_capture(path, payload, **kwargs) + + monkeypatch.setattr(decision_capture, "write_decision_capture", fail_optional_wire) + digest = hashlib.sha256(raw).hexdigest() + args = build_parser().parse_args(["--format", "json", "quota", "context-project", + "--capture-file", str(path), "--capture-sha256", digest, "--goal-id", "saved-goal", + "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn", "--context-delivery", "on", + "--codex-app", "--codex-context-dir", str(context), "--decision-output-root", str(root)]) + result = project_saved_context_capture(args) assert result["ok"] and result["mode"] == "full" assert result["path"] == str(path) and result["sha256"] == digest - assert path.read_bytes() == raw and list(context.iterdir()) == [] + recovered_path, recovered_digest = codex_context.read_codex_context_reference(str(context), + goal_id="saved-goal", agent_id="saved-agent") + recovered = Path(recovered_path).read_bytes() + assert hashlib.sha256(recovered).hexdigest() == recovered_digest + assert json.loads(recovered) == original and path.read_bytes() == raw + + +def test_saved_projection_recovery_failure_returns_typed_error_and_current_reference( + tmp_path, monkeypatch, capsys, +): + from loopx.cli import main + from loopx.control_plane.quota.error_codes import QuotaDecisionTransportError + from loopx.control_plane.turn_driver import codex_context + + original = {"ok": True, "goal_id": "saved-goal", "agent_identity": {"agent_id": "saved-agent"}, + "heartbeat_receipt": {"turn_instance_id": "saved-turn", "status": "committed"}, + "requirement": "Use the current captured decision, never the prior session reference."} + raw = json.dumps(original).encode() + path, root, context = tmp_path / "original.json", tmp_path / "cache", tmp_path / "native" + path.write_bytes(raw) + root.mkdir(mode=0o700) + context.mkdir(mode=0o700) + stale = context / "session.json" + stale.write_text(json.dumps({"recovery": {"path": "prior-capture.json"}})) + stale_bytes = stale.read_bytes() + digest = hashlib.sha256(raw).hexdigest() + + def refuse_recovery(payload, **kwargs): + assert payload == original + raise QuotaDecisionTransportError( + 'Current context recovery could not be published or safely invalidated. ' + 'Read the saved original decision/receipt; do not rerun admission.' + ) + + monkeypatch.setattr(codex_context, "project_codex_context", refuse_recovery) + registry, runtime = tmp_path / "absent-registry.json", tmp_path / "absent-runtime" + code = main(["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "quota", "context-project", "--capture-file", str(path), "--capture-sha256", digest, + "--goal-id", "saved-goal", "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn", + "--decision-output-root", str(root), "--context-delivery", "on", "--codex-app", + "--codex-context-dir", str(context)]) + result = json.loads(capsys.readouterr().out) + assert code == 1 and result["ok"] is False + assert result["error_code"] == "quota_decision_transport_failed" + assert result["heartbeat_receipt"] == original["heartbeat_receipt"] + assert result["capture_sha256"] == digest + assert result["recovery"]["path"] == str(path) and result["recovery"]["sha256"] == digest + assert "prior-capture" not in json.dumps(result) + assert not list(root.glob("*/delivery.json")) + assert path.read_bytes() == raw and stale.read_bytes() == stale_bytes + assert not registry.exists() and not runtime.exists() + code, recovered = _run_cli(registry, runtime, "quota", "context-read", + "--capture-file", result["recovery"]["path"], "--capture-sha256", result["recovery"]["sha256"]) + assert code == 0 and recovered["eof"] and recovered["content"].encode() == raw + assert not registry.exists() and not runtime.exists() def test_preflight_instruction_pins_both_projection_and_full_fallback(tmp_path):