fix(context): keep fallback recovery bound to the current capture

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.

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

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
 

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,

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