From 9de6605602908d6258ceafaaf9a32ec15023505b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 18:14:34 +0800 Subject: [PATCH] Recover a retried native context offer against its original predecessor Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/quota/context_delivery.ts | 2 + .../turn_driver/codex_context.py | 13 +++++- .../test_context_delivery_cli.py | 20 +++++++++ .../test_native_context_delivery_cost.py | 4 +- .../test_saved_context_project.py | 45 +++++++++++++++++++ 5 files changed, 80 insertions(+), 4 deletions(-) diff --git a/loopx/control_plane/quota/context_delivery.ts b/loopx/control_plane/quota/context_delivery.ts index f0569628ba..dfec0e2f61 100644 --- a/loopx/control_plane/quota/context_delivery.ts +++ b/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); diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index 99a3f9c84b..6773998caa 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/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 {}) diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py index f783ad5e12..40c2242e40 100644 --- a/tests/control_plane/test_context_delivery_cli.py +++ b/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 diff --git a/tests/control_plane/test_native_context_delivery_cost.py b/tests/control_plane/test_native_context_delivery_cost.py index 996095af67..73ea1c45bf 100644 --- a/tests/control_plane/test_native_context_delivery_cost.py +++ b/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": diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index 0756f4a688..3d1cc25ed1 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/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()