Preserve the original predecessor when a saved native capture is presented twice

loopx/control_plane/quota/context_delivery.ts

@@ -613,6 +613,7 @@ export function encodeContextDelivery(request: JsonObject, legacyQuotaFields: re
   const fileBacked = !retained && continuity?.basis === "captured_file"
     && continuity.capture_bytes_verified === true && equal(continuity.target, binding);
   if (!binding || !prior || (!confirmed && !assumed && !fileBacked) || !compatible(binding, target)
+      || (fileBacked && binding.capture_id === target.capture_id)
       || binding.context_digest !== digest(prior)) return fullResult("baseline_unavailable");
   const depth = baseline?.ancestry;
   if (!Number.isSafeInteger(depth) || Number(depth) < 0 || Number(depth) >= MAX_ANCESTRY) return fullResult("full_refresh_due");
@@ -677,6 +678,7 @@ export function restoreContextDelivery(request: JsonObject, legacyQuotaFields: r
     const fileBacked = delivery.continuity_basis === "captured_file";
     if (delivery.schema_version !== VERSION || delivery.mode !== "delta" || !equal(delivery.target, target)
         || !equal(delivery.base, base) || !compatible(base, target) || base.context_digest !== digest(context)
+        || (fileBacked && base.capture_id === target.capture_id)
         || !Number.isSafeInteger(baseline.ancestry) || Number(baseline.ancestry) < 0
         || delivery.ancestry !== Number(baseline.ancestry) + 1 || Number(delivery.ancestry) > MAX_ANCESTRY) return fallback();
     const before = freshPaths(context, surface, legacyQuotaFields, base), after = freshPaths(currentContext, surface, legacyQuotaFields, target);

loopx/control_plane/turn_driver/codex_context.py

@@ -554,7 +554,7 @@ def _discard_recovery(root: Path) -> None:
     path.unlink()
 
 
-def _native_context_request(state: dict[str, Any], scope: dict[str, Any]) -> dict[str, Any]:
+def _native_context_request(state: dict[str, Any], scope: dict[str, Any], capture_id: str) -> dict[str, Any]:
     """Compare verified durable versions independently of model retention.
 
     The same session's last full capture is a file baseline after partial reads
@@ -564,6 +564,11 @@ def _native_context_request(state: dict[str, Any], scope: dict[str, Any]) -> dic
     """
     request: dict[str, Any] = {"scope": scope}
     previous = state.get("recovery")
+    target = previous.get("target") if isinstance(previous, dict) else None
+    if isinstance(target, dict) and target.get("capture_id") == capture_id:
+        # Publishing an offer does not mean a reader saw it. A retry may use
+        # the original predecessor, but never this offer as its own baseline.
+        previous = previous.get("comparison_base")
     if isinstance(previous, dict) and isinstance(previous.get("target"), dict):
         try:
             path, raw, payload = read_context_capture_file(previous["path"], previous["sha256"], private=True)
@@ -638,7 +643,7 @@ def project_codex_context(payload: dict[str, Any], *, directory: str | None, cap
                     for key in ("consumed_pointers", "confirmed_read_sections", "section_reads"):
                         state.pop(key, None)
                 state["read_scope"] = scope
-                request = _native_context_request(state, scope)
+                request = _native_context_request(state, scope, capture_id)
                 recovery = capture_context_projection(payload, directory=capture_directory,
                     read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE)
                 state["source_recovery"] = {key: recovery[key] for key in ("path", "sha256")}
@@ -685,6 +690,10 @@ def publish_recovery(reference: dict[str, Any]) -> None:
                 if result.get("schema_version") == "context_delivery_v1":
                     state["recovery"]["target"] = result["target"]
                     state["recovery"]["continuity_basis"] = "captured_file"
+                    if result.get("continuity_basis") == "captured_file" and result.get("mode") == "delta":
+                        base = request["baseline_reference"]
+                        state["recovery"]["comparison_base"] = {
+                            "path": base["path"], "sha256": base["sha256"], "target": result["base"]}
                     state["pending"].update(target=result["target"], ancestry=result["ancestry"],
                         full_read_sections=transaction["read_sections"],
                         read_sections=transaction["read_sections"] if result["mode"] == "full" else {})

tests/control_plane/test_context_delivery_cli.py

@@ -986,3 +986,23 @@ def measured_rpc(method, params, **kwargs):
     # Allow payload growth plus 25% structural headroom over the matched
     # safe-integer workload; count complete requests and results, not deltas alone.
     assert totals[1] / totals[0] <= 1.25 * source_sizes[1] / source_sizes[0]
+
+
+def test_file_delta_requires_a_different_source_even_for_identical_content():
+    from loopx.control_plane.quota.turn_envelope import prepare_context_delivery, restore_context_delivery
+
+    scope = dict(goal_id="g", agent_id="a", session_id="s", context_epoch="e")
+    body = {"requirement": "Identical content can belong to two independent captures. " * 1200}
+    def capture(identity, base=None):
+        return prepare_context_delivery(body, scope=scope, capture_id=identity, baseline=base,
+            continuity={"basis": "captured_file", "target": base["target"], "capture_bytes_verified": True}
+            if base else None)
+    old, current = capture("a"), capture("b")
+    warm = capture("b", old)
+    assert warm["mode"] == "delta"
+    assert capture("b", current)["mode"] == "full"
+    # Old producers could emit the same valid empty change tree referencing
+    # the target itself. The receiver has an independent, pinned current file.
+    legacy = {**warm, "base": current["target"]}
+    recovered = restore_context_delivery(legacy, baseline=current, current=current)
+    assert recovered["mode"] == "full" and recovered["context"] == body

tests/control_plane/test_native_context_delivery_cost.py

@@ -107,9 +107,9 @@ def test_small_codec_delta_uses_cheaper_complete_native_delivery(tmp_path, monke
     index_requests = []
     request = turn_envelope._turn_envelope_result
 
-    def observe_registration(directory, view):
+    def observe_registration(directory, view, *, hook_session=None):
         registrations.append((directory, copy.deepcopy(view)))
-        return register(directory, view)
+        return register(directory, view, hook_session=hook_session)
 
     def observe_index_request(operation, **kwargs):
         if operation == "context_select":

tests/control_plane/test_saved_context_project.py

@@ -938,3 +938,48 @@ def unexpected(*args, **kwargs):
     result = project_saved_context_capture(args)
     assert result["ok"] and result["mode"] == "full" and result["path"] == str(path)
     assert not root.exists() and path.read_bytes() == raw
+
+
+@pytest.mark.skipif(os.name != "posix", reason="Native session continuity uses a private POSIX journal")
+def test_retry_lost_navigation_uses_original_predecessor(tmp_path, monkeypatch):
+    from loopx.control_plane.turn_driver.codex_context import codex_context_hook
+
+    journal = tmp_path / "native"
+    journal.mkdir(mode=0o700)
+    monkeypatch.setenv("CODEX_THREAD_ID", "retry-session")
+    event = {"hook_event_name": "SessionStart", "source": "startup",
+             "session_id": "retry-session", "cwd": str(tmp_path)}
+    codex_context_hook(str(journal), event)
+    registry, runtime = tmp_path / "absent-registry", tmp_path / "absent-runtime"
+
+    def present(turn):
+        body = {"ok": True, "goal_id": "g", "agent_identity": {"agent_id": "a"},
+                "heartbeat_receipt": {"turn_instance_id": turn},
+                "requirement": "Preserve this independently recoverable requirement. " * 1900}
+        path = tmp_path / (turn + ".json")
+        if not path.exists():
+            path.write_text(json.dumps(body))
+        rc, view = _run_cli(registry, runtime, "quota", "context-project", "--capture-file", str(path),
+            "--capture-sha256", hashlib.sha256(path.read_bytes()).hexdigest(), "--goal-id", "g",
+            "--agent-id", "a", "--turn-instance-id", turn, "--context-delivery", "on",
+            "--codex-app", "--codex-context-dir", str(journal), cwd=tmp_path)
+        assert rc == 0, view
+        return machine_reference_oracle(view)
+
+    cold = present("cold")
+    assert cold["mode"] == present("cold")["mode"] == "full"
+    # No model read or PostToolUse receipt: the caller may have lost the notification.
+    warm = present("warm")
+    retry = present("warm")
+    for view in (warm, retry):
+        assert view["mode"] == "delta"
+        wire = json.loads(Path(view["path"]).read_bytes())
+        assert wire["base"] == cold["target"] and wire["target"] != wire["base"]
+        assert json.loads(Path(view["recovery"]["path"]).read_bytes())["heartbeat_receipt"] == {"turn_instance_id": "warm"}
+    # Loss of the actual predecessor or a new epoch must recover this source independently.
+    Path(warm["baseline_reference"]["path"]).unlink()
+    assert present("warm")["mode"] == "full"
+    codex_context_hook(str(journal), {**event, "hook_event_name": "PreCompact"})
+    codex_context_hook(str(journal), {**event, "source": "compact"})
+    assert present("warm")["mode"] == "full"
+    assert not registry.exists() and not runtime.exists()