diff --git a/benchmark/edgebench/checkpoint_controller.py b/benchmark/edgebench/checkpoint_controller.py
index 8e42612281..d9aa99d1ea 100644
--- a/benchmark/edgebench/checkpoint_controller.py
+++ b/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):
diff --git a/benchmark/edgebench/checkpoint_worker.py b/benchmark/edgebench/checkpoint_worker.py
index 4061361a42..55d8e4be46 100644
--- a/benchmark/edgebench/checkpoint_worker.py
+++ b/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()
diff --git a/benchmark/runtime/checkpoint.py b/benchmark/runtime/checkpoint.py
index bbb66de210..9edaf81043 100644
--- a/benchmark/runtime/checkpoint.py
+++ b/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
 
diff --git a/benchmark/runtime/checkpoint_continuation.py b/benchmark/runtime/checkpoint_continuation.py
index 37df0873d5..52802fda60 100644
--- a/benchmark/runtime/checkpoint_continuation.py
+++ b/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}
diff --git a/benchmark/runtime/checkpoint_session_refresh.py b/benchmark/runtime/checkpoint_session_refresh.py
index 1a077ce377..c898238811 100644
--- a/benchmark/runtime/checkpoint_session_refresh.py
+++ b/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")
diff --git a/benchmark/runtime/checkpoint_wake.py b/benchmark/runtime/checkpoint_wake.py
index 299a2f0dbe..fec2a576e2 100644
--- a/benchmark/runtime/checkpoint_wake.py
+++ b/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
diff --git a/benchmark/runtime/sforge_warm_start.py b/benchmark/runtime/sforge_warm_start.py
index 86a15f4c9e..58e6471466 100644
--- a/benchmark/runtime/sforge_warm_start.py
+++ b/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)
diff --git a/benchmark/tests/qualify_checkpoint_joint.py b/benchmark/tests/qualify_checkpoint_joint.py
index 38c1f5a011..88bb21d635 100644
--- a/benchmark/tests/qualify_checkpoint_joint.py
+++ b/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()
diff --git a/benchmark/tests/test_checkpoint_handoff.py b/benchmark/tests/test_checkpoint_handoff.py
index b8214a5a75..7ca7c929b1 100644
--- a/benchmark/tests/test_checkpoint_handoff.py
+++ b/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}}
diff --git a/benchmark/tests/test_checkpoint_joint_receiver.py b/benchmark/tests/test_checkpoint_joint_receiver.py
index 1910877ec0..219d862018 100644
--- a/benchmark/tests/test_checkpoint_joint_receiver.py
+++ b/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(
diff --git a/docs/reference/research-checkpoint.md b/docs/reference/research-checkpoint.md
index 831c88ae86..ccb63dd3cd 100644
--- a/docs/reference/research-checkpoint.md
+++ b/docs/reference/research-checkpoint.md
@@ -44,6 +44,13 @@ history and qualified native stores. The native inventory excludes credential
 files; arbitrary caller-declared artifacts remain private and are not privacy
 certified. A new destination is required; capture never overwrites a point.
 
+Artifact operations use the existing private local snapshot transport for full
+inventories and retained metadata above the ordinary 2 MiB Effect wire limit.
+The existing 64 MiB local snapshot bound and original operation deadlines still
+apply. No witness is truncated, no capture is enabled implicitly, and a lost
+capture response must be reconciled under its original identity before another
+operation is considered.
+
 The inventory covers only declared file roots plus the automatically owned
 session and LoopX state participants. A conversation mentioning an external
 candidate or report does not capture that file. Declare every needed research
diff --git a/loopx/cli_commands/research_checkpoint.py b/loopx/cli_commands/research_checkpoint.py
index 3ae2993ecb..87c40d5dc7 100644
--- a/loopx/cli_commands/research_checkpoint.py
+++ b/loopx/cli_commands/research_checkpoint.py
@@ -53,7 +53,7 @@ def required(*names):
                      "operation_id", "expected_revision", "lease_key", "lease_version")
         if args.operation == "verify":
             result = effect_runtime_result("runtime.checkpoint_artifacts", {
-                "action": "verify", "root": str(args.root), "expected_sha256": args.expected_sha256})
+                "action": "verify", "root": str(args.root), "expected_sha256": args.expected_sha256}, large_local_snapshot=True)
         elif not args.execute:
             result = {"dry_run": True, "operation": args.operation, "execution_authority_granted": False,
                 "next_action": "Review the explicit source/destination and use --execute for this operation."}
diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py
index 823c224e4d..a98f7228dd 100644
--- a/loopx/control_plane/effect_runtime.py
+++ b/loopx/control_plane/effect_runtime.py
@@ -52,6 +52,7 @@
     "performance_diagnosis.inspect",
     "state.recovery.archive.verify",
     "state.recovery.manifest.verify",
+    "runtime.checkpoint_artifacts",
 })
 MAX_STARTUP_DIAGNOSTIC_BYTES = 8 * 1024
 STARTUP_LOCK_TIMEOUT_SECONDS = 15.0
diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts
index 7f024eed59..b6dc1eec95 100644
--- a/loopx/control_plane/effect_runtime_server.ts
+++ b/loopx/control_plane/effect_runtime_server.ts
@@ -38,6 +38,7 @@ const LOCAL_SNAPSHOT_METHODS = new Set([
   "performance_diagnosis.inspect",
   "state.recovery.archive.verify",
   "state.recovery.manifest.verify",
+  "runtime.checkpoint_artifacts",
 ]);
 const DEFAULT_IDLE_MS = 5 * 60 * 1_000;
 // Bounds for LOOPX_EFFECT_RUNTIME_IDLE_MS. The upper bound is the largest
diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py
index b5f609b4e6..2acab8852b 100644
--- a/loopx/hosts/codex_checkpoint.py
+++ b/loopx/hosts/codex_checkpoint.py
@@ -305,7 +305,7 @@ def _fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: st
     deadline = time.monotonic() + timeout
     effect_runtime_result("runtime.checkpoint_artifacts", {
         "action": "verify", "root": str(unit), "expected_sha256": expected_sha256,
-    }, timeout=timeout)
+    }, timeout=timeout, large_local_snapshot=True)
     manifest = json.loads((unit / "checkpoint.json").read_text(encoding="utf-8"))
     native = manifest["point_metadata"]["native_session"]
     if native.get("interrupted_guard") is not None:
@@ -319,7 +319,7 @@ def _fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: st
         intent = effect_runtime_result("runtime.checkpoint_artifacts", {
             "action": "select_replan", "quota": json.loads(raw), "goal_id": guard["goal_id"],
             "agent_id": guard["agent_id"], "turn_instance_id": guard["turn_instance_id"],
-        }, timeout=max(.1, deadline - time.monotonic()))
+        }, timeout=max(.1, deadline - time.monotonic()), large_local_snapshot=True)
         if not intent["capture"] or intent != manifest["point_metadata"].get("replan"):
             raise ValueError("interrupted checkpoint original replan binding differs")
     home = Path(env["CODEX_HOME"])
@@ -465,7 +465,7 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh
     deadline = time.monotonic() + timeout
     effect_runtime_result("runtime.checkpoint_artifacts", {
         "action": "verify", "root": str(unit), "expected_sha256": expected_sha256,
-    }, timeout=timeout)
+    }, timeout=timeout, large_local_snapshot=True)
     metadata = json.loads((unit / "checkpoint.json").read_bytes())["point_metadata"]
     source_profile = (metadata.get("session") or {}).get("session_profile_digest")
     from loopx.control_plane.content_digest import BARE_SHA256_PATTERN
diff --git a/loopx/hosts/research_checkpoint.py b/loopx/hosts/research_checkpoint.py
index ed5583c4f9..62e3332eea 100644
--- a/loopx/hosts/research_checkpoint.py
+++ b/loopx/hosts/research_checkpoint.py
@@ -185,7 +185,7 @@ def read_timeout():
 
     def verify_artifact():
         effect_runtime_result("runtime.checkpoint_artifacts", {
-            "action": "verify", "root": str(unit), "expected_sha256": digest}, timeout=read_timeout())
+            "action": "verify", "root": str(unit), "expected_sha256": digest}, timeout=read_timeout(), large_local_snapshot=True)
         read_timeout()  # A late successful read cannot authorize the next write.
 
     def read_session():
@@ -532,7 +532,7 @@ def settled():
                 {**entry, "id": allocated[entry["id"]]} for entry in native["metadata"]["store_members"]]}
             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 native_session_checkpoint(env, session.session_id, timeout=max(.1, deadline-time.time()), transport=reader) != native:
                 raise ValueError("native checkpoint source moved before sealing")
             _, confirmed = native_binding_participant(env, components, metadata)
@@ -545,7 +545,7 @@ def settled():
             result = effect_runtime_result("runtime.checkpoint_artifacts", {
                 "action": "capture", "destination": str(destination), "components": components,
                 "expected_source_sha256": probe["source_sha256"], "point_metadata": metadata,
-                "deadline_epoch_ms": deadline*1000}, timeout=max(.1, deadline-time.time()), retry_safe=False)
+                "deadline_epoch_ms": deadline*1000}, timeout=max(.1, deadline-time.time()), retry_safe=False, large_local_snapshot=True)
     return result
 
 
diff --git a/tests/control_plane/test_checkpoint_artifact_transport.py b/tests/control_plane/test_checkpoint_artifact_transport.py
new file mode 100644
index 0000000000..10c214d714
--- /dev/null
+++ b/tests/control_plane/test_checkpoint_artifact_transport.py
@@ -0,0 +1,117 @@
+"""Complete checkpoint witnesses cross the existing private Effect transport."""
+import hashlib
+import json
+import time
+
+import pytest
+
+from loopx.cli import main
+from loopx.control_plane.effect_runtime import (
+    MAX_REQUEST_BYTES,
+    MAX_RESPONSE_BYTES,
+    EffectRuntimeRemoteError,
+    EffectRuntimeResponseAmbiguous,
+    effect_runtime_result,
+)
+
+
+# This is a complete-transport functional check, not a storage latency SLO.
+# A complete inventory performs durable per-file copy/flush work; keep its
+# finite RPC allowance separate from artifact deadlines
+# or the benchmark's separate live-capture/sealing clocks.
+LARGE_ARTIFACT_RPC_SECONDS = 240
+
+
+def test_large_member_transport_with_explicit_rpc_budget_preserves_inventory(tmp_path, capsys):
+    source = tmp_path / "research"
+    source.mkdir(mode=0o700)
+    content = b"retained negative evidence\n"
+    expected = [{"path": "", "kind": "directory", "mode": 0o700}]
+    for index in range(8192):
+        name = f"{index:05}-" + "e" * 150 + ".txt"
+        path = source / name
+        path.write_bytes(content)
+        path.chmod(0o600)
+        expected.append({"path": name, "kind": "file", "mode": 0o600,
+                         "size": len(content), "sha256": hashlib.sha256(content).hexdigest()})
+    witness = {"research": expected}
+    # The expected inventory is independent of the production walker. A count
+    # alone would miss dropped, reordered or substituted scientific members.
+    assert len(json.dumps(witness, separators=(",", ":")).encode()) > MAX_RESPONSE_BYTES
+    components = [{"id": "research", "source_path": str(source)}]
+    request = {"action": "probe", "components": components}
+    with pytest.raises(EffectRuntimeResponseAmbiguous):
+        effect_runtime_result("runtime.checkpoint_artifacts", request, timeout=LARGE_ARTIFACT_RPC_SECONDS)
+    observed = effect_runtime_result("runtime.checkpoint_artifacts", request,
+                                     timeout=LARGE_ARTIFACT_RPC_SECONDS, large_local_snapshot=True)
+    assert observed["source_witness"] == witness
+    unit = tmp_path / "point"
+    captured = effect_runtime_result("runtime.checkpoint_artifacts", {
+        "action": "capture", "components": components, "destination": str(unit),
+        "point_metadata": {}, "expected_source_sha256": observed["source_sha256"],
+    }, timeout=LARGE_ARTIFACT_RPC_SECONDS, retry_safe=False, large_local_snapshot=True)
+    manifest = json.loads((unit / "checkpoint.json").read_bytes())
+    assert manifest["source_witness"] == witness and manifest["artifact_witness"] == witness
+    assert captured["source_stability_verified"] and not captured["execution_authority_granted"]
+    assert main(["--format", "json", "research-checkpoint", "verify", "--root", str(unit),
+                 "--expected-sha256", captured["sha256"]]) == 0
+    verified = json.loads(capsys.readouterr().out)
+    assert verified["integrity_verified"] and not verified["research_point_qualified"]
+    destination = tmp_path / "receiver"
+    restored = effect_runtime_result("runtime.checkpoint_artifacts", {
+        "action": "materialize", "root": str(unit), "expected_sha256": captured["sha256"],
+        "destination": str(destination),
+    }, timeout=LARGE_ARTIFACT_RPC_SECONDS, retry_safe=False, large_local_snapshot=True)
+    assert restored["materialized"] and not restored["activation_performed"]
+    assert (destination / "checkpoint.json").read_bytes() == (unit / "checkpoint.json").read_bytes()
+    assert sorted(p.name for p in (destination / "components/research").iterdir()) == [
+        row["path"] for row in expected[1:]]
+    member = destination / "components/research" / expected[-1]["path"]
+    assert member.read_bytes() == content
+    member.unlink()
+    with pytest.raises(EffectRuntimeRemoteError):
+        effect_runtime_result("runtime.checkpoint_artifacts", {"action": "verify", "root": str(destination),
+            "expected_sha256": captured["sha256"]}, timeout=LARGE_ARTIFACT_RPC_SECONDS, large_local_snapshot=True)
+
+
+def test_large_checkpoint_metadata_uses_private_request_and_response(tmp_path):
+    source = tmp_path / "candidate"
+    source.write_bytes(b"candidate")
+    components = [{"id": "code", "source_path": str(source)}]
+    observed = effect_runtime_result("runtime.checkpoint_artifacts", {
+        "action": "probe", "components": components}, large_local_snapshot=True)
+    request = {"action": "capture", "components": components, "destination": str(tmp_path / "point"),
+        "point_metadata": {"retained_basis": "x" * (MAX_REQUEST_BYTES + 1)},
+        "expected_source_sha256": observed["source_sha256"]}
+    result = effect_runtime_result("runtime.checkpoint_artifacts", request,
+        timeout=30, retry_safe=False, large_local_snapshot=True)
+    basis = effect_runtime_result("runtime.checkpoint_artifacts", {
+        "action": "read_basis", "root": request["destination"], "expected_sha256": result["sha256"]},
+        timeout=30, large_local_snapshot=True)
+    assert basis["point_metadata"] == request["point_metadata"]
+    assert result["source_stability_verified"] and not result["execution_authority_granted"]
+
+
+def test_large_rpc_allowance_cannot_extend_artifact_capture_deadline(tmp_path):
+    source = tmp_path / "candidate"
+    source.write_bytes(b"retained evidence")
+    components = [{"id": "code", "source_path": str(source)}]
+    observed = effect_runtime_result("runtime.checkpoint_artifacts", {
+        "action": "probe", "components": components}, large_local_snapshot=True)
+    # The same valid request succeeds with a future artifact deadline. Do not
+    # infer the rejection cause from internal error prose hidden by the bridge.
+    live = effect_runtime_result("runtime.checkpoint_artifacts", {
+        "action": "capture", "components": components,
+        "destination": str(tmp_path / "live"), "point_metadata": {},
+        "expected_source_sha256": observed["source_sha256"],
+        "deadline_epoch_ms": (time.time() + LARGE_ARTIFACT_RPC_SECONDS) * 1000,
+    }, timeout=LARGE_ARTIFACT_RPC_SECONDS, retry_safe=False, large_local_snapshot=True)
+    assert live["source_stability_verified"]
+    unit = tmp_path / "expired"
+    with pytest.raises(EffectRuntimeRemoteError):
+        effect_runtime_result("runtime.checkpoint_artifacts", {
+            "action": "capture", "components": components,
+            "destination": str(unit), "point_metadata": {},
+            "expected_source_sha256": observed["source_sha256"], "deadline_epoch_ms": 0,
+        }, timeout=LARGE_ARTIFACT_RPC_SECONDS, retry_safe=False, large_local_snapshot=True)
+    assert not unit.exists()