fix(checkpoint): preserve complete large artifact witnesses

benchmark/edgebench/checkpoint_controller.py

@@ -227,12 +227,12 @@ def capture(controller_file: Path, *, token: str, capture_id: str, destination:
         components.append({"id": "host_phase_deadline", "source_path": str(phase_deadline_file)})
         metadata["host_phase_budget"] = _phase_budget(phase_raw, native)
     probe = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": components},
-                                  timeout=_remaining(deadline), retry_safe=False)
+                                  timeout=_remaining(deadline), retry_safe=False, large_local_snapshot=True)
     result = effect_runtime_result("runtime.checkpoint_artifacts", {
         "action": "capture", "components": components, "destination": str(destination),
         "expected_source_sha256": probe["source_sha256"],
         "point_metadata": metadata,
-    }, timeout=_remaining(deadline), retry_safe=False)
+    }, timeout=_remaining(deadline), retry_safe=False, large_local_snapshot=True)
     # Bind the actually copied bytes, not merely a prior source observation.
     _bound_bank(destination / "components/native_judge", reference)
     if phase_raw is not None and (destination / "components/host_phase_deadline").read_bytes() != phase_raw:
@@ -291,7 +291,7 @@ def restore(controller_file: Path, *, root: Path, expected_sha256: str, run_id:
     controller = _controller(controller_file)
     root = _physical(root)
     effect_runtime_result("runtime.checkpoint_artifacts", {
-        "action": "verify", "root": str(root), "expected_sha256": expected_sha256})
+        "action": "verify", "root": str(root), "expected_sha256": expected_sha256}, large_local_snapshot=True)
     manifest = json.loads((root / "checkpoint.json").read_bytes())
     expected_components = ["native_judge"]
     if phase_deadline_file is not None:
@@ -336,7 +336,7 @@ def restore(controller_file: Path, *, root: Path, expected_sha256: str, run_id:
             effect_runtime_result("runtime.checkpoint_artifacts", {
                 "action": "materialize", "root": str(root), "expected_sha256": expected_sha256,
                 "destination": str(stage / "unit"),
-            }, retry_safe=False)
+            }, retry_safe=False, large_local_snapshot=True)
             imported = stage / "unit/components/native_judge"
             _bound_bank(imported, reference)
             with _bank_import_lock(target, timeout):

benchmark/edgebench/checkpoint_worker.py

@@ -240,7 +240,7 @@ def capture_worker(controller_file: Path, *, token: str, capture_id: str,
                     if prepared is not None:
                         observed = effect_runtime_result("runtime.checkpoint_artifacts", {
                             "action": "probe", "components": components,
-                        }, timeout=_remaining(deadline), retry_safe=False)
+                        }, timeout=_remaining(deadline), retry_safe=False, large_local_snapshot=True)
                         if observed["source_sha256"] != prepared["source_sha256"]:
                             changed = []
                             for key, rows in observed["source_witness"].items():
@@ -298,11 +298,11 @@ def capture_worker(controller_file: Path, *, token: str, capture_id: str,
                     metadata["writer_barrier"] = "docker_pause"
                 probe = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": components,
                     "deadline_epoch_ms": sealing_epoch_ms},
-                                              timeout=_remaining(deadline), retry_safe=False)
+                                              timeout=_remaining(deadline), retry_safe=False, large_local_snapshot=True)
                 return effect_runtime_result("runtime.checkpoint_artifacts", {
                     "action": "capture", "components": components, "destination": str(destination),
                     "expected_source_sha256": probe["source_sha256"], "point_metadata": metadata,
                     "deadline_epoch_ms": sealing_epoch_ms,
-                }, timeout=_remaining(deadline), retry_safe=False)
+                }, timeout=_remaining(deadline), retry_safe=False, large_local_snapshot=True)
     finally:
         client.close()

benchmark/runtime/checkpoint.py

@@ -109,7 +109,7 @@ def _prior_replan_guard_seen(env: dict[str, str], current: Path, obligation_id:
         prior = effect_runtime_result("runtime.checkpoint_artifacts", {
             "action": "select_replan", "quota": guard, "goal_id": env["LOOPX_GOAL_ID"],
             "agent_id": env["LOOPX_AGENT_ID"], "turn_instance_id": path.parent.name,
-        })
+        }, large_local_snapshot=True)
         if prior["capture"] and prior["obligation_id"] == obligation_id:
             return True
     return False
@@ -161,7 +161,7 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
         intent = effect_runtime_result("runtime.checkpoint_artifacts", {
             "action": "select_replan", "quota": guard, "goal_id": env["LOOPX_GOAL_ID"],
             "agent_id": env["LOOPX_AGENT_ID"], "turn_instance_id": receipt["turn_id"],
-        })
+        }, large_local_snapshot=True)
         from .checkpoint_continuation import restored_capture_gate
         restore_gate = restored_capture_gate(env, intent.get("obligation_id"),
             full_history=not request.get("native_session", False))
@@ -199,7 +199,7 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
                     raise ValueError("existing replan point has a different declaration")
                 verified = effect_runtime_result("runtime.checkpoint_artifacts", {
                     "action": "verify", "root": str(destination), "expected_sha256": manifest["sha256"],
-                })
+                }, large_local_snapshot=True)
                 receipt["checkpoint"].update(reason="replan_already_captured", destination=str(destination), **verified)
             else:
                 # A failed first capture is a missed point. Once its model is
@@ -360,7 +360,7 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
             probe = effect_runtime_result("runtime.checkpoint_artifacts", {
                 "action": "probe", "components": components,
                 "deadline_epoch_ms": deadline * 1000,
-            }, timeout=max(.1, deadline - time.time()), retry_safe=False)
+            }, timeout=max(.1, deadline - time.time()), retry_safe=False, large_local_snapshot=True)
             if request.get("native_session"):
                 # Native readback precedes the byte witness. A completed later Turn
                 # in that gap must not be sealed under the earlier cursor. Re-read
@@ -402,7 +402,7 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
                 "components": components, "point_metadata": metadata,
                 "expected_source_sha256": probe["source_sha256"],
                 "deadline_epoch_ms": deadline * 1000,
-            }, timeout=max(.1, deadline - time.time()), retry_safe=False)
+            }, timeout=max(.1, deadline - time.time()), retry_safe=False, large_local_snapshot=True)
             receipt["checkpoint"] = {"captured": True, "destination": str(destination), **captured}
     except Exception as error:
         # Capture failure is explicit but cannot rewrite an already settled work
@@ -480,7 +480,7 @@ def remaining():
                 "action": "select_replan", "quota": quota, "goal_id": goal_id,
                 "agent_id": agent_id, "turn_instance_id": (turn_instance_id if turn_instance_id is not None
                     else (quota.get("heartbeat_receipt") or {}).get("turn_instance_id")),
-            }, timeout=remaining())
+            }, timeout=remaining(), large_local_snapshot=True)
         if not failed_readback and native_process is not None and interrupt_when is not None and interrupt_when(intent):
             from .worker import interrupt_checkpoint_process
 

benchmark/runtime/checkpoint_continuation.py

@@ -64,7 +64,7 @@ def _restored_context(env, current_obligation_id=None, *, verify_native, include
     unit = Path(source)
     manifest = effect_runtime_result("runtime.checkpoint_artifacts", {
         "action": "read_basis", "root": str(unit), "expected_sha256": sha,
-    })
+    }, large_local_snapshot=True)
     # Admission already verified all artifact bytes. Here only the immutable
     # declaration, original pending packet and native history are inputs; do
     # not rehash large code/incumbent archives on every ordinary work wake.
@@ -86,13 +86,13 @@ def _restored_context(env, current_obligation_id=None, *, verify_native, include
             "action": "select_replan", "quota": packet,
             "goal_id": metadata["loopx_state"]["goal_id"], "agent_id": metadata["loopx_state"]["agent_id"],
             "turn_instance_id": packet["heartbeat_receipt"]["turn_instance_id"],
-        })
+        }, large_local_snapshot=True)
         if intent != pending or intent.get("capture") is not True:
             raise ValueError("restored original replan binding differs")
     result = effect_runtime_result("runtime.checkpoint_artifacts", {
         "action": "restore_capture", **lineage, "checkpoint_sha256": sha,
         "current_obligation_id": current_obligation_id, "review": None,
-    })
+    }, large_local_snapshot=True)
     # Existing saved preflights are discovery only. Each candidate is resolved
     # through the canonical current-Goal settlement owner, never copied logs.
     for guard_path in sorted(Path(env["LOOPX_WAKE_LOG_DIR"]).glob("*/checkpoint-start-guard.json"), reverse=True):
@@ -111,7 +111,7 @@ def _restored_context(env, current_obligation_id=None, *, verify_native, include
                 "semantic_replan_guard": observed.semantic_replan_guard,
             },
             **({"include_review_basis": True} if include_review_basis else {}),
-        })
+        }, large_local_snapshot=True)
         if result["capture_allowed"]:
             # Qualification happens before launching this wake's model. A
             # current receipt alone cannot stand in for native continuation.
@@ -161,6 +161,6 @@ def restored_capture_history(env, preflight, *, session_id):
         raise ValueError("restored capture requires its admitted session and completed cursor")
     manifest = effect_runtime_result("runtime.checkpoint_artifacts", {
         "action": "read_basis", "root": binding["checkpoint_artifact_root"], "expected_sha256": sha,
-    })
+    }, large_local_snapshot=True)
     return {"retained_history": retained_native_history(binding, manifest["point_metadata"]),
             "required_completed_turn_id": cursor}

benchmark/runtime/checkpoint_session_refresh.py

@@ -49,7 +49,7 @@ def _prepare_fresh_replan(env, session, wake, declaration):
 
     unit = Path(declaration["root"])
     manifest = effect_runtime_result("runtime.checkpoint_artifacts", {
-        "action": "read_basis", "root": str(unit), "expected_sha256": declaration["sha256"]})
+        "action": "read_basis", "root": str(unit), "expected_sha256": declaration["sha256"]}, large_local_snapshot=True)
     metadata = manifest["point_metadata"]
     component = metadata.get("held_decision_component_id")
     if component not in {entry["id"] for entry in manifest["components"]}:
@@ -65,7 +65,7 @@ def _prepare_fresh_replan(env, session, wake, declaration):
         "action": "select_replan", "quota": packet,
         "goal_id": metadata["loopx_state"]["goal_id"],
         "agent_id": metadata["loopx_state"]["agent_id"],
-        "turn_instance_id": packet["heartbeat_receipt"]["turn_instance_id"]})
+        "turn_instance_id": packet["heartbeat_receipt"]["turn_instance_id"]}, large_local_snapshot=True)
     if selected.get("capture") is not True or selected != (
             metadata.get("replan") or metadata.get("operator_metadata", {}).get("replan")):
         raise ValueError("fresh replan requires the original pending decision boundary")

benchmark/runtime/checkpoint_wake.py

@@ -151,7 +151,7 @@ def _capture(self, held, deadline):
                 metadata["held_decision_component_id"] = decision_id
             probe = effect_runtime_result("runtime.checkpoint_artifacts", {
                 "action": "probe", "components": components, "deadline_epoch_ms": window * 1000,
-            }, timeout=max(.1, window - time.time()), retry_safe=False)
+            }, timeout=max(.1, window - time.time()), retry_safe=False, large_local_snapshot=True)
             if native is not None:
                 if restored_capture_history(self.env, self.receipt.get("checkpoint_restore"),
                         session_id=self.receipt["session"]["session_id"]) != native_options:
@@ -181,7 +181,7 @@ def _capture(self, held, deadline):
                 "action": "capture", "components": components, "destination": str(destination),
                 "expected_source_sha256": probe["source_sha256"], "point_metadata": metadata,
                 "deadline_epoch_ms": window * 1000,
-            }, timeout=max(.1, window - time.time()), retry_safe=False)
+            }, timeout=max(.1, window - time.time()), retry_safe=False, large_local_snapshot=True)
 
     def run(self, execution, body, wake, session_wake, stdout, stderr, timeout):
         from .worker import child_process, native_command

benchmark/runtime/sforge_warm_start.py

@@ -56,7 +56,7 @@ def verify(self, workspace: str) -> dict:
             raise ValueError("Original warm start deadline expired before checkpoint verification")
         effect_runtime_result("runtime.checkpoint_artifacts", {
             "action": "verify", "root": str(self.root), "expected_sha256": self.sha256},
-            timeout=min(_CHECKPOINT_VERIFY_TIMEOUT_SECONDS, remaining))
+            timeout=min(_CHECKPOINT_VERIFY_TIMEOUT_SECONDS, remaining), large_local_snapshot=True)
         metadata = json.loads((self.root / "checkpoint.json").read_bytes())["point_metadata"]
         native, source = metadata["native_session"], metadata["loopx_state"]
         if (native["guest_workspace"] != workspace or native["guest_home"] != _CODEX_HOME
@@ -192,7 +192,7 @@ def restore_workspace(declaration: dict, workspace: Path):
     unit = Path(declaration["root"])
     effect_runtime_result("runtime.checkpoint_artifacts", {
         "action": "verify", "root": str(unit), "expected_sha256": declaration["sha256"]},
-        timeout=_phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS))
+        timeout=_phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS), large_local_snapshot=True)
     _phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS)
     manifest = json.loads((unit / "checkpoint.json").read_bytes())
     metadata = manifest["point_metadata"]
@@ -227,7 +227,7 @@ def restore_workspace(declaration: dict, workspace: Path):
     if research:
         copied = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe",
             "components": [{"id": key, "source_path": str(target)} for key, _, target in research]},
-            timeout=_phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS))
+            timeout=_phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS), large_local_snapshot=True)
         if any(copied["source_witness"][key] != manifest["artifact_witness"][key] for key, _, _ in research):
             raise ValueError("Restored research files differ from the sealed artifacts; preserve this failed receiver")
     _phase_read_timeout(declaration, _CHECKPOINT_VERIFY_TIMEOUT_SECONDS)

benchmark/tests/qualify_checkpoint_joint.py

@@ -139,7 +139,7 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
                 raise ValueError("Reused point/configuration must be physical paths")
             packet = json.loads((point / "checkpoint.json").read_bytes())
             verified = effect_runtime_result("runtime.checkpoint_artifacts", {
-                "action": "verify", "root": str(point), "expected_sha256": packet["sha256"]})
+                "action": "verify", "root": str(point), "expected_sha256": packet["sha256"]}, large_local_snapshot=True)
             assert verified["integrity_verified"]
             metadata = packet["point_metadata"]
             assert metadata["held_decision_component_id"] and metadata["worker_file_snapshot"]["prepared_source_verified"]
@@ -231,7 +231,7 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
                 recaptured = newer[0]
                 next_packet = json.loads((recaptured / "checkpoint.json").read_bytes())
                 check = effect_runtime_result("runtime.checkpoint_artifacts", {
-                    "action": "verify", "root": str(recaptured), "expected_sha256": next_packet["sha256"]})
+                    "action": "verify", "root": str(recaptured), "expected_sha256": next_packet["sha256"]}, large_local_snapshot=True)
                 assert check["integrity_verified"]
                 third = output / "grandreceiver"
                 third.mkdir()

benchmark/tests/test_checkpoint_handoff.py

@@ -441,7 +441,8 @@ def test_restored_capture_cannot_drop_its_preflight_basis(tmp_path, monkeypatch,
         gate["capture_allowed"] = False
     reads = []
     monkeypatch.setattr(continuation, "load_codex_cli_session", lambda *a, **k: binding)
-    def basis(method, request):
+    def basis(method, request, *, large_local_snapshot):
+        assert large_local_snapshot is True
         assert request == {"action": "read_basis", "root": str(tmp_path / "point"), "expected_sha256": "a" * 64}
         reads.append(True)
         return {"point_metadata": {"native_session": native}}

benchmark/tests/test_checkpoint_joint_receiver.py

@@ -76,7 +76,7 @@ def test_captured_point_continues_with_fresh_authority(
     verified = effect_runtime_result("runtime.checkpoint_artifacts", {
         "action": "materialize", "root": str(unit), "expected_sha256": digest,
         "destination": str(materialized),
-    })
+    }, large_local_snapshot=True)
     assert verified["materialized"] and not verified["activation_performed"]
     source_workspace = materialized / "components" / roots[0]["id"]
     assert source_workspace.is_dir()
@@ -1003,7 +1003,7 @@ def run_native_work(current_guard, logs, purpose):
                 assert opportunity_after["remaining_cooldown_seconds"] <= opportunity["remaining_cooldown_seconds"]
                 assert actual_phase.read_bytes() == original_phase
                 check = effect_runtime_result("runtime.checkpoint_artifacts", {
-                    "action": "verify", "root": str(recaptured), "expected_sha256": captured["sha256"]})
+                    "action": "verify", "root": str(recaptured), "expected_sha256": captured["sha256"]}, large_local_snapshot=True)
                 assert check["integrity_verified"] and not check["research_point_qualified"]
                 (tmp_path / "full-recapture.json").write_text(json.dumps(boundary["checkpoint"]))
             elif boundary["checkpoint"].get("reason") == "no_fresh_replan":
@@ -1033,13 +1033,13 @@ def run_native_work(current_guard, logs, purpose):
                 assert captured["point_metadata"]["native_session"]["turn_ids"] == grown["turn_ids"]
                 assert (manual / "components/code").read_text() == "10"
                 verified_point = effect_runtime_result("runtime.checkpoint_artifacts", {
-                    "action": "verify", "root": str(manual), "expected_sha256": captured["sha256"]})
+                    "action": "verify", "root": str(manual), "expected_sha256": captured["sha256"]}, large_local_snapshot=True)
                 assert verified_point["integrity_verified"] and not verified_point["research_point_qualified"]
                 capture_after_wake(capture_env, completed)
                 assert completed["checkpoint"]["reason"] == "destination_already_present_no_replay"
         effect_runtime_result(
             "runtime.checkpoint_artifacts",
-            {"action": "verify", "root": str(unit), "expected_sha256": digest},
+            {"action": "verify", "root": str(unit), "expected_sha256": digest}, large_local_snapshot=True,
         )
         assert (
             hashlib.sha256(