diff --git a/loopx/control_plane/turn_driver/codex_context_history.py b/loopx/control_plane/turn_driver/codex_context_history.py index 6ac6e86bc6..385ac243e3 100644 --- a/loopx/control_plane/turn_driver/codex_context_history.py +++ b/loopx/control_plane/turn_driver/codex_context_history.py @@ -236,9 +236,23 @@ def observe_native_read(state: dict[str, Any], event: dict[str, Any], """Collect content-free IO facts alongside the existing pending byte ranges.""" pending = state.get("pending") call = event.get("tool_use_id") - if (not isinstance(pending, dict) or not isinstance(call, str) or not call + if (not isinstance(call, str) or not call or not pages or any(page.get("ok") is not True for page in pages)): return state + if pending is None: + recovery, candidate = state.get("recovery"), state.get("candidate") or {} + if (not isinstance(recovery, dict) or not isinstance(candidate, dict) + or any(page.get("sha256") != recovery.get("sha256") for page in pages) + or (candidate.get("capture_sha256") == recovery.get("sha256") + and candidate.get("capture_bytes_verified") is True)): + return state + # Invalidation clears pending reads. Record the first original-capture + # recovery read too, including a full capture delivered in one page. + # These are IO annotations only: the typed lifecycle still authenticates + # coverage and decides whether a complete baseline can be established. + pending = {} + if not isinstance(pending, dict): + return state read = {"call_id": call, "pages_sha256": _digest(pages)} reads = pending.get("native_reads", []) if not isinstance(reads, list): diff --git a/tests/control_plane/test_codex_context_capture.py b/tests/control_plane/test_codex_context_capture.py index af39e651f4..f6cacd2ae5 100644 --- a/tests/control_plane/test_codex_context_capture.py +++ b/tests/control_plane/test_codex_context_capture.py @@ -87,7 +87,7 @@ def _last_page(request): + [(host, surface, False, True, False) for host in ("benchmark", "code_mode") for surface in ("quota_packet", "turn_envelope")] + [("code_mode", surface, False, True, True) for surface in ("quota_packet", "turn_envelope")]) -def test_native_cli_capture_delta_and_compact_recovery(tmp_path, monkeypatch, host, surface, failed_primary, resumed, adaptive, hook_loss=None, shared_root=False, checkpoint_handoff=False, handoff_socket=None, native_pragma=False, native_read_batch=2, native_history_tokens=None, resume_loss=None, bounded_pages=False, ssh_goal=False, reader_output_bytes=None): +def test_native_cli_capture_delta_and_compact_recovery(tmp_path, monkeypatch, host, surface, failed_primary, resumed, adaptive, hook_loss=None, shared_root=False, checkpoint_handoff=False, handoff_socket=None, native_pragma=False, native_read_batch=2, native_history_tokens=None, resume_loss=None, bounded_pages=False, ssh_goal=False, reader_output_bytes=None, recovery_resume=False): # Real worker qualification follows returned cursors; output budgets can # shrink a page, so precomputed fixed offsets are not its production path. bounded_pages = bounded_pages or host == "benchmark" @@ -399,11 +399,27 @@ def response(request, number, message): return tool(page["next_command"], number) stages["phase"] = "recovered" return response(request, number, message) + elif phase == "recovery_resuming": + # The actual host has exited and reconstructed history again. Its + # model input must retain the recovered full body, and SessionStart + # must preserve that new baseline without another recovery read. + from loopx.control_plane.turn_driver.codex_context import _state + + assert delivered(request, "recovery") == Path(wires["warm"]["recovery"]["path"]).read_text().rstrip("\n") + journal = _state(lifecycle) + assert journal["candidate"]["target"]["context_epoch"] == stages["recovered_epoch"] + assert journal["candidate"]["capture_sha256"] == page_digests["recovery"] + stages["phase"] = "done" + return message else: assert phase == "recovered" recovered = delivered(request, "recovery") assert recovered == Path(wires["warm"]["recovery"]["path"]).read_text().rstrip("\n") - stages["phase"] = "done" + if recovery_resume: + from loopx.control_plane.turn_driver.codex_context import _state + + stages["recovered_epoch"] = _state(lifecycle)["context_epoch"] + stages["phase"] = "recovery_resuming" if recovery_resume else "done" return message return tool(cmd, number) @@ -587,6 +603,9 @@ def held(*args, **kwargs): if resumed: assert _app_server(binary, env, workspace, home, True, sandbox=sandbox, resume=session_id, turn_timeout_sec=60) == session_id + if recovery_resume: + assert _app_server(binary, env, workspace, home, True, sandbox=sandbox, + resume=session_id, turn_timeout_sec=60) == session_id assert not errors, errors assert stages["phase"] == "done" @@ -678,6 +697,12 @@ def held(*args, **kwargs): assert bounded_sizes and max(bounded_sizes) > 3072 +@pytest.mark.parametrize("surface", ["quota_packet", "turn_envelope"]) +def test_native_compact_recovery_survives_process_resume(tmp_path, monkeypatch, surface): + test_native_cli_capture_delta_and_compact_recovery(tmp_path, monkeypatch, + "code_mode", surface, False, True, True, bounded_pages=True, recovery_resume=True) + + @pytest.mark.parametrize("surface", ["quota_packet", "turn_envelope"]) @pytest.mark.parametrize("loss", ["replace", "unqualified_profile"]) def test_native_worker_resume_history_change_reads_current_full(tmp_path, monkeypatch, surface, loss): diff --git a/tests/control_plane/test_codex_context_resume.py b/tests/control_plane/test_codex_context_resume.py index 522dce22bb..f531e383a4 100644 --- a/tests/control_plane/test_codex_context_resume.py +++ b/tests/control_plane/test_codex_context_resume.py @@ -82,6 +82,63 @@ def test_completed_native_resume_keeps_delta_without_old_capture_prefetch(tmp_pa assert wire["changes"]["values"]["heartbeat_receipt"] == current["heartbeat_receipt"] +@pytest.mark.parametrize("single_page", [False, True]) +@pytest.mark.parametrize("lose_first", [False, True]) +@pytest.mark.parametrize("code_mode", [False, True]) +def test_compact_full_recovery_becomes_next_resume_baseline(tmp_path, monkeypatch, single_page, lose_first, code_mode): + from argparse import Namespace + from loopx.cli_commands.quota_capture import read_context_capture + + event, history, payload = observed_read(tmp_path, monkeypatch) + before = context._state(tmp_path) + context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PreCompact"}) + append(history, "compacted", {"message": "Summary without captured source bodies."}) + context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PostCompact"}) + recovery_instruction = context.codex_context_hook(str(tmp_path), {**event, "source": "compact"}) + invalid = context._state(tmp_path) + assert "candidate" not in invalid and "pending" not in invalid + assert invalid["context_epoch"] != before["context_epoch"] + assert invalid["recovery"] == before["recovery"] + assert invalid["recovery"]["command"] in str(recovery_instruction) + + reference = invalid["recovery"] + offset, index = 0, 0 + while True: + page = read_context_capture(Namespace(capture_file=reference["path"], capture_sha256=reference["sha256"], + offset=offset, page_bytes=64000 if single_page else 4096, + max_page_output_bytes=64000 if single_page else None, format="json")) + assert page["ok"] + call = f"recovery-{index}" + if code_mode: + # The nested hook fires before its parent emits the retained output. + append(history, "response_item", {"type": "custom_tool_call", "call_id": f"parent-{call}"}) + else: + append(history, "response_item", {"type": "function_call", "call_id": call}) + retained = "truncated" if lose_first and index == 0 else json.dumps(page) + if not code_mode: + append(history, "response_item", {"type": "function_call_output", "call_id": call, "output": retained}) + context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PostToolUse", + "tool_use_id": call, "tool_response": json.dumps(page)}) + if code_mode: + append(history, "response_item", {"type": "custom_tool_call_output", "call_id": f"parent-{call}", + "output": [{"type": "input_text", "text": retained}]}) + index += 1 + if page["eof"]: + break + offset = page["next_offset"] + assert (index == 1) == single_page + recovered = context._state(tmp_path) + assert recovered["candidate"]["capture_sha256"] == reference["sha256"] + assert recovered["candidate"]["target"]["context_epoch"] == invalid["context_epoch"] + append(history, "event_msg", {"type": "task_complete"}) + context.codex_context_hook(str(tmp_path), {**event, "source": "resume"}) + current = {**payload, "heartbeat_receipt": {"turn_instance_id": "after-compact"}} + _, wire = project(tmp_path, "after-compact", payload=current) + assert wire["mode"] == ("full" if lose_first else "delta") + if not lose_first: + assert wire["changes"]["values"]["heartbeat_receipt"] == current["heartbeat_receipt"] + + @pytest.mark.parametrize("change", ["compact", "rollback", "retained", "unknown", "truncate", "replace", "rewrite", "missing", "output", "duplicate", "model", "workspace", "session", "permission", "writable", "malformed"]) @pytest.mark.parametrize("version", ["0.162.1", "0.163.0-alpha.11.1"])