fix(context): preserve complete recovery across native resume

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):

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):

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"])