fix(k1): restore low-latency live recovery path

docs/adr/0042-k1-run-ledger-and-regression-telemetry.md

@@ -0,0 +1,150 @@
+# ADR 0042 — K1 run ledger and regression telemetry
+
+Status: proposed for acceptance
+Date: 2026-08-22
+
+Named physical references and redacted evidence hashes are retained in
+[`docs/lab/010_K1_MISSION_CORE_INTERNAL_LIVE_BASELINE_20260822.redacted.md`](../lab/010_K1_MISSION_CORE_INTERNAL_LIVE_BASELINE_20260822.redacted.md).
+That report deliberately distinguishes the fast Mission Core internal
+ideal-path reference from the accepted recovery-capable checkpoint and from
+manufacturer/LixelGO evidence.
+
+## Context
+
+The K1 integration has repeatedly returned to the same visible failures after
+otherwise successful recovery work: delayed first point cloud, a camera that
+repeatedly disappears and reopens, browser decoder failures, slow START/STOP,
+and connection-mode transitions that disturb an already accepted Bridge path.
+
+The current evidence is individually strong but operationally fragmented. Raw
+MQTT capture, camera fMP4 epochs, acquisition checkpoints, browser diagnostics,
+redacted manifests and operator timing notes use different identities and must
+be correlated manually. A successful run is described in Ops, but is not yet an
+executable baseline that can reject a later regression.
+
+The repository already owns an accepted telemetry plane under ADR 0031:
+Telegraf and a bounded native JSONL outbox publish through authenticated MQTT to
+the Mission Core normalizer and TimescaleDB. Introducing another broker or
+database solely for K1 would create a second lifecycle, credential boundary and
+recovery problem without improving the local K1-to-Mission-Core evidence path.
+
+## Decision
+
+Every physical or replayed K1 acquisition is one explicit **K1 run**. The run is
+identified by the acquisition id and its evidence-session lineage. All derived
+events carry the same redacted identity, code revision, UI build id, connection
+mode and monotonic clock origin.
+
+The authoritative write path is:
+
+```text
+K1 control, MQTT receiver, camera gateway and browser diagnostics
+    -> local append-only K1 run journal
+    -> bounded durable outbox
+    -> authenticated telemetry MQTT topic
+    -> existing Mission Core telemetry normalizer
+    -> existing TimescaleDB
+    -> run comparison and acceptance report
+```
+
+The local journal remains the source of truth during network loss. Database
+availability never gates START, STOP, raw evidence capture, camera archival or
+local recovery. Delivery resumes from the outbox when the telemetry contour is
+available.
+
+The initial event contract is `missioncore.k1-run-event/v1`. Each event contains:
+
+- `run_id`, `acquisition_id`, redacted `evidence_session_id` lineage;
+- UTC and monotonic timestamps plus a per-run sequence number;
+- Mission Core commit, UI build id and K1 compatibility-profile id;
+- connection mode, host-path epoch and producer generations without addresses,
+  credentials, project names or raw device identifiers;
+- event code, stage, outcome and bounded numeric facts;
+- source component and source schema version.
+
+The required timeline includes at least:
+
+- operator START intent, physical command publication and confirmed SCANNING;
+- calibration start/end when the device exposes those facts;
+- first authoritative PCL received and first PCL published to Rerun;
+- browser Rerun store admitted and first visible point-cloud frame;
+- camera producer init, first media, browser first frame and sustained playing;
+- camera decoder errors, preview-reader retirements, IDR admission, queue age,
+  bytes and fragment counts;
+- connection gap start, recovery attempts, recovered binding and gap end;
+- operator STOP intent, confirmed READY and archive finalization;
+- terminal run result and automatically evaluated acceptance checks.
+
+An acceptance report is generated from the journal after each run. It compares
+the run with a named accepted baseline and reports exact deltas. The following
+conditions are hard failures rather than performance warnings:
+
+- any browser `MEDIA_ERR_DECODE`;
+- any decode error in the archived camera stream;
+- a missing first point-cloud admission or missing terminal archive;
+- loss of acquisition/control lineage;
+- a device state asserted by the UI without authoritative device evidence.
+
+Latency and FPS thresholds are versioned baseline checks. They are never
+implemented by silently reducing point count, image quality, bitrate or source
+FPS. A threshold change requires its own reviewed acceptance run.
+
+The engineering work format changes accordingly:
+
+1. Name the accepted baseline and the single invariant targeted by the change.
+2. Make one bounded implementation increment.
+3. Run focused synthetic tests before a physical test.
+4. Perform one canonical UI run; do not substitute direct device commands.
+5. Generate and retain the run comparison automatically.
+6. Do not merge a change that improves its target while regressing an already
+   accepted invariant.
+
+## Rejected alternatives
+
+### Redis
+
+Redis is useful for disposable cache and coordination. It is not the evidence
+system of record for offline physical runs and does not replace a durable local
+journal or analytical store.
+
+### Kafka
+
+Kafka is not admitted for this local single-producer contour. The existing MQTT
+plane already supplies authenticated fan-out and reconnect delivery, while the
+local outbox supplies offline durability. Kafka would add a second broker and
+operational quorum without repairing the K1, RTSP, browser or local-network
+boundaries. It can be reconsidered only when independent high-volume consumers,
+cross-site retention or replay requirements exceed the accepted MQTT contour.
+
+### ClickHouse
+
+ClickHouse is not admitted as a second telemetry database. Existing TimescaleDB
+already stores run identity, timestamps and JSON payloads and is sufficient for
+the current event volume and comparisons. The event schema remains storage
+neutral, so a later measured analytical-volume limit can justify a reviewed
+ClickHouse migration without changing producers.
+
+## Implementation increments
+
+1. Generate a local K1 run report from the evidence and diagnostic artifacts
+   already produced today; no runtime transport change.
+2. Emit the versioned append-only event contract directly during a run and seal
+   it with the evidence session.
+3. Tail the bounded outbox through the accepted ADR 0031 telemetry plane and
+   extend the normalizer with K1 run series.
+4. Add baseline comparison and a CI/replay gate. Physical acceptance remains a
+   separate operator-controlled UI test.
+
+No product window, status family or K1 control command is introduced by this
+decision.
+
+## Consequences
+
+- A stable K1 run becomes executable regression evidence rather than a prose
+  memory.
+- Camera, point-cloud, control and recovery events can be compared on one clock
+  and one run identity.
+- Wi-Fi or telemetry-plane loss cannot erase the local run or block the device.
+- The existing telemetry stack is extended instead of duplicated.
+- The first implementation work is an evidence/reporting increment; it does not
+  authorize broad recomposition of the fragile K1 communication path.

docs/lab/010_K1_MISSION_CORE_INTERNAL_LIVE_BASELINE_20260822.redacted.md

@@ -0,0 +1,213 @@
+# K1 ↔ Mission Core internal live baseline — 2026-08-22
+
+Status: accepted internal engineering reference
+
+Scope: one owner-controlled XGRIDS/LixelKity K1, exact firmware 3.0.2
+
+Connection: Mission Core Bridge/direct-LAN path
+
+Operator path: canonical Mission Core UI only
+
+## Classification
+
+This is the fast **Mission Core ↔ K1 internal reference**, not a vendor/LixelGO
+reference, not a production SLA and not proof for arbitrary Wi-Fi conditions,
+another K1, another firmware, Quick Connect, Direct Connect, LTE, a tunnel or a
+vehicle installation.
+
+The reference exists because the same visible regressions repeatedly returned:
+slow START/STOP, delayed point-cloud admission, a camera that disappeared and
+reopened, browser decoder errors and a discrete trajectory. It records both the
+fastest clean internal run and the accepted recovery-capable checkpoint that
+followed it. Raw MQTT, video and device evidence remain outside Git; only
+redacted summaries, hashes and engineering conclusions are retained here.
+
+No point count, image quality, bitrate, source FPS or K1 protocol setting was
+reduced to obtain either result.
+
+## Baseline A — fast ideal-path reference
+
+- Code: `eaad9de` (`fix(k1): stabilize live recovery and media admission`).
+- Evidence session: `20260822T105904Z_viewer_live`.
+- Local redacted manifest SHA-256:
+  `990b9433d40965b2b712df155f5f96d38770d53fddde804852c11eca90d96b3e`.
+- Local MQTT summary SHA-256:
+  `9438cc8947c7a868ff8ffcfc163233dfea40a95b7aae34050806bb072b2a7750`.
+- Local camera summary SHA-256:
+  `bc333e1dd1e7800204b242eecd01a83d1c91c388ab2202a238e8d2fc4a43aa14`.
+- Final phase: `idle`; stop reason: `external_stop`; rejected MQTT messages: 0.
+
+Operator-observed UI/physical timing:
+
+| Boundary | Observed time |
+| --- | ---: |
+| START click → K1 calibration onset | < 5 s |
+| K1 calibration | 21 s |
+| calibration end → visible point cloud | 2 s |
+| calibration end → visible right camera | 4 s |
+| STOP click → physical stop onset | 1 s |
+
+Recorded Mission Core facts:
+
+| Metric | Value |
+| --- | ---: |
+| PCL frames / pose frames | 121 / 132 |
+| points published | 412,175 |
+| point decode errors | 0 |
+| preview dropped | 0 |
+| MQTT callback → publish p50 / p95 | 23.839 / 41.541 ms |
+| camera archive | complete, 70 segments, 5,645,411 bytes |
+| camera failure code | none |
+| offline H.264 decode errors | 0 |
+
+This run used a fresh, short physical-command history: the terminal ledger was
+2,989 bytes at revision 15 with no archive segments. That distinction is
+material. The timings are a low-latency reference, not evidence that a mature
+recovery ledger can be cryptographically reread several times per UI poll for
+free.
+
+## Regression that exposed the shared bottleneck
+
+The recovery-capable working line later produced session
+`20260822T124524Z_viewer_live`:
+
+- START click → calibration onset: 26 s;
+- calibration end → cloud: about 2 s;
+- calibration end → camera: 19 s with repeated disappearance and large lag;
+- STOP click → physical stop onset: 12 s;
+- preview dropped: 393;
+- callback → publish p50/p95: 110.331/309.434 ms;
+- the retained fMP4 archive contained real H.264 macroblock/bytestream decode
+  errors, while the ideal reference did not.
+
+The camera UI resets were therefore a symptom, not the primary cause. Increasing
+preview queues or reducing media/point quality would only hide overload.
+
+Profiling found repeated full parsing, validation and archive-hash traversal of
+the physical START/STOP ledger inside passive `state.read`, nested public state
+projections and the first-PCL camera path. The mature ledger was 18,378 bytes at
+revision 1,219 plus five immutable archive segments / 318,027 bytes. One passive
+state request took 240–680 ms while the UI requested state four times per second.
+That contention delayed MQTT publication and FFmpeg stdout consumption together.
+
+## Baseline B — accepted recovery-capable checkpoint
+
+Evidence session: `20260822T130323Z_viewer_live`.
+
+- Local redacted manifest SHA-256:
+  `f870bedd9036a3f03993344d405eab4ef668d1f0fdae55e16adf8ecdcb51f48a`.
+- Local MQTT summary SHA-256:
+  `2d42d31c14d98d77bcefc8a1b9983e102cace2375ada9e5f89f021a33920a55d`.
+- Local camera summary SHA-256:
+  `1386ea16615a9716b699d6ba35733ef26d367b6a5257d572cde11509e546cdda`.
+- Final phase: `idle`; stop reason: `external_stop`; rejected MQTT messages: 0.
+
+Operator-observed UI/physical timing:
+
+| Boundary | Observed time |
+| --- | ---: |
+| START click → K1 calibration onset | 14 s |
+| K1 calibration | 22 s |
+| calibration end → visible point cloud | about 1 s |
+| calibration end → visible right camera | 8 s |
+| STOP click → physical stop onset | 7 s |
+
+Recorded Mission Core facts:
+
+| Metric | Value |
+| --- | ---: |
+| PCL frames / pose frames | 369 / 400 |
+| points published | 1,318,018 |
+| point decode errors | 0 |
+| preview dropped | 70 |
+| callback → publish p50 / p95 | 83.934 / 223.589 ms |
+| first-PCL camera admission | 65 ms |
+| camera authority / FFmpeg / post-spawn commit | 5 / 16 / 2 ms |
+| total backend camera activation | 23 ms |
+| backend activation → browser playing | 5.383 s |
+| browser camera restarts / slow-consumer retirements | 0 / 0 |
+| camera archive | complete, 342 segments, 27,469,046 bytes |
+| camera failure code | none |
+| offline H.264 decode errors | 0 |
+
+The accepted STOP was emitted once. Mission Core received an exactly correlated
+successful K1 application response after about 2.35 s, later observed fresh
+`READY` and resolved the durable physical command as
+`stop-standby-observed`. No automatic physical retry was introduced.
+
+After the bounded read-path change, passive `state.read` measured 23–43 ms on
+the same mature ledger instead of 240–680 ms.
+
+## What restored the usable path
+
+1. A single verified physical-ledger snapshot is shared across one public state
+   projection instead of being recursively reloaded by nested projections.
+2. Process-local control facts can be read without implicitly traversing the
+   durable physical ledger again.
+3. Passive snapshots reuse an already cryptographically verified ledger only
+   while the ledger directory, main file, archive directory and every bounded
+   archive entry retain the exact device/inode/mode/owner/link-count/size/
+   `mtime_ns`/`ctime_ns` fingerprint. Every mutation and proof still performs a
+   complete reload. Cross-process atomic writes and same-size archive tampering
+   invalidate the cache and fail closed.
+4. Camera authority is fully proved before FFmpeg `Popen`; after spawn only a
+   bounded in-memory compare-and-swap may commit the producer. Reader threads
+   therefore start without a second durable-ledger pause that can back up the
+   fMP4 pipe.
+5. First-authoritative-PCL camera admission no longer recursively rereads the
+   same physical ledger on the 10 Hz ingress callback.
+6. Structured timings were added for first-PCL admission, camera activation and
+   STOP dispatch. These are diagnostics only; they do not add a K1 command,
+   status, retry or product window.
+
+## Fragile boundaries that remain open
+
+- Baseline B is usable but does not equal Baseline A: 70 preview drops and
+  83.934/223.589 ms p50/p95 remain above the clean reference of 0 drops and
+  23.839/41.541 ms.
+- START and physical STOP onset remain materially slower than the fast reference.
+  The current evidence separates local validation, correlated application
+  response and later `READY`, but does not yet expose a device-side calibration
+  onset or physical motor/LED timestamp on the same clock.
+- Camera presentation still depends on RTSP startup, the next usable H.264
+  parameter-set/IDR boundary and browser MSE admission. The accepted archive has
+  non-monotonic-DTS warnings but no H.264 decode corruption.
+- A successful application response is necessary but not sufficient to dismiss
+  the interface. Mission Core must keep the active scene until authoritative K1
+  state proves the requested physical transition.
+- Wi-Fi loss, controller-process loss, K1 power/battery loss and return must
+  retain the current no-automatic-START/STOP rule and exact lineage. They need
+  separate UI acceptance runs; this baseline must not be used to infer them.
+- Bridge remains the accepted product path. Quick Connect/Bridge switching must
+  not reset the whole reactive application or reuse the other mode's reconnect
+  affordance, and must not perturb the Bridge protocol to improve Quick Connect.
+- The original LixelGO packet captures remain manufacturer-app evidence. They
+  are not interchangeable with this Mission Core internal baseline.
+
+## Regression rule
+
+Future K1 work starts from one named baseline and changes one bounded invariant.
+The canonical acceptance path is the Mission Core UI. A run is rejected if it
+introduces an archived-camera H.264 decode error, browser `MEDIA_ERR_DECODE`,
+missing point-cloud admission, missing terminal archive, lost control/acquisition
+lineage, duplicate physical command or a UI state not backed by authoritative K1
+evidence. Performance must not be recovered by silently reducing source data.
+
+## Delivery validation
+
+- Canonical UI physical run: accepted as Baseline B above.
+- Focused backend regression suite for the touched viewer diagnostics,
+  acquisition lifecycle, application session, camera gateway and physical
+  ledger: 800/800 passed.
+- Ruff: passed for every changed Python source and test file.
+- mypy: passed for all five changed source modules.
+- `git diff --check`: passed.
+- Offline decode of the accepted 50-second camera window: no H.264 decode
+  corruption; only non-monotonic-DTS warnings were observed.
+- Canonical Mission Core service remained available on `127.0.0.1:8000` after
+  validation.
+
+The next performance increment should be driven by a generated run comparison
+under ADR 0042. Kafka, Redis or ClickHouse are not a substitute for fixing the
+local control/media hot path and are not admitted without a measured requirement
+beyond the existing MQTT/Timescale telemetry contour.

src/k1link/device_plugins/xgrids_k1/camera.py

@@ -661,11 +661,13 @@ class XgridsK1CameraGateway:
         target_host: str,
         session_dir: Path,
         *,
+        pre_prepare_fence: CameraProducerCommitFence,
         commit_fence: CameraProducerCommitFence,
         committed_before_start: CameraProducerCommittedObserver | None = None,
     ) -> dict[str, Any]:
-        """Prepare the first acquisition camera outside every shared gate."""
+        """Reserve authority before Popen, then commit without a dead-pipe gap."""
 
+        activation_started = time.monotonic()
         if source_id not in CAMERA_SOURCE_PATHS:
             raise ValueError("неизвестный camera source")
         target = validate_private_ipv4(target_host)
@@ -674,29 +676,41 @@ class XgridsK1CameraGateway:
             raise ValueError("observation session directory does not exist")
         if not root.is_relative_to(self._repository_root):
             raise ValueError("camera recording root must stay inside the repository")
-        with self._lifecycle_lock:  # noqa: SIM117 - lock order is intentional
+        reserved_generation: list[int] = []
-            with self._lock:
+
-                self._require_open_locked()
+        def reserve() -> bool:
-                if self._ffmpeg_path is None:
+            with self._lifecycle_lock:  # noqa: SIM117 - lock order is intentional
-                    raise RuntimeError("локальный camera adapter FFmpeg не найден")
+                with self._lock:
-                if self._producer is not None or self._recording_root is not None:
+                    self._require_open_locked()
-                    raise RuntimeError("camera acquisition producer уже активен")
+                    if self._ffmpeg_path is None:
-                if self._source_id is not None and (
+                        raise RuntimeError("локальный camera adapter FFmpeg не найден")
-                    self._source_id != source_id or self._target_host != target
+                    if self._producer is not None or self._recording_root is not None:
-                ):
+                        raise RuntimeError("camera acquisition producer уже активен")
-                    raise RuntimeError("для camera gateway уже выбран другой source")
+                    if self._source_id is not None and (
-                if self._source_id is None:
+                        self._source_id != source_id or self._target_host != target
-                    self._generation += 1
+                    ):
-                self._source_id = source_id
+                        raise RuntimeError("для camera gateway уже выбран другой source")
-                self._target_host = target
+                    if self._source_id is None:
-                self._recording_root = root
+                        self._generation += 1
-                self._recording_media_segment_count = 0
+                    self._source_id = source_id
-                self._archive_summaries = []
+                    self._target_host = target
-                self._expected_source_end_generation = None
+                    self._recording_root = root
-                self._phase = "selected"
+                    self._recording_media_segment_count = 0
-                self._error = None
+                    self._archive_summaries = []
-                self._revision += 1
+                    self._expected_source_end_generation = None
+                    self._phase = "selected"
+                    self._error = None
+                    self._revision += 1
+                    reserved_generation.append(self._generation)
+                    return True
+
+        if not pre_prepare_fence(reserve):
+            raise ValueError("camera activation authority устарела до запуска adapter")
+        if len(reserved_generation) != 1:
+            raise RuntimeError("camera activation fence did not reserve exactly once")
+        authority_reserved = time.monotonic()
         prepared = self._prepare_selected_producer()
+        ffmpeg_prepared = time.monotonic()
 
         def commit() -> bool:
             with self._lifecycle_lock:
@@ -719,7 +733,29 @@ class XgridsK1CameraGateway:
         if not committed:
             self._discard_prepared_producer(prepared, failure_code="stale-activation-commit")
             raise ValueError("camera activation lineage устарела")
-        return self.snapshot()
+        readers_started = time.monotonic()
+        snapshot = self.snapshot()
+        logger.info(
+            "K1 camera acquisition producer activation timing",
+            extra={
+                "event_code": "k1_camera_activation_timing",
+                "camera_generation": reserved_generation[0],
+                "camera_authority_wait_ms": int(round(
+                    (authority_reserved - activation_started) * 1_000,
+                )),
+                "camera_ffmpeg_prepare_ms": int(round(
+                    (ffmpeg_prepared - authority_reserved) * 1_000,
+                )),
+                "camera_post_spawn_commit_ms": int(round(
+                    (readers_started - ffmpeg_prepared) * 1_000,
+                )),
+                "camera_activation_total_ms": int(round(
+                    (readers_started - activation_started) * 1_000,
+                )),
+                "device_command_sent": False,
+            },
+        )
+        return snapshot
 
     def stop(self, generation: int) -> dict[str, Any]:
         with self._lifecycle_lock:

src/k1link/device_plugins/xgrids_k1/facade.py

@@ -599,6 +599,18 @@ class _ActiveStreamRecoveryLineage:
     target_port: int
 
 
+@dataclass(frozen=True, slots=True)
+class _CameraProducerActivationClaim:
+    """Pre-Popen authority reduced to a bounded post-Popen local CAS."""
+
+    acquisition_id: str
+    evidence_session_id: str
+    start_operation_id: str | None
+    stop_operation_id: str | None
+    runtime_producer_generation: int
+    recovery_generation: int
+
+
 @dataclass(frozen=True, slots=True)
 class _PostRecoveryCameraRestartKey:
     """Exact recovered-PCL and camera progress CAS claimed by one worker."""
@@ -3848,7 +3860,7 @@ class XgridsK1CompatibilityService:
     def _retire_terminal_prestart_control_failure(self) -> bool:
         """Automatically clear only proven local, pre-START terminal state."""
 
-        snapshot = self._application_control_session.snapshot()
+        snapshot = self._control_session_snapshot_without_physical_command()
         self._retire_prepared_acquisition_on_terminal_control_failure(snapshot)
         state = str(snapshot.get("state") or "unknown")
         failure = snapshot.get("failure")
@@ -3909,6 +3921,13 @@ class XgridsK1CompatibilityService:
                 or self.runtime.snapshot().get("source_mode") != "idle"
             ):
                 return False
+            control = self._control_session_snapshot_without_physical_command()
+            if control.get("state") not in {
+                "connection-ready",
+                "workspace-ready",
+                "project-ready",
+            }:
+                return False
             physical = self._physical_command_coordinator.snapshot()
             if (
                 physical.get("status") not in {"empty", "resolved"}
@@ -3916,14 +3935,6 @@ class XgridsK1CompatibilityService:
                 or _physical_command_reports_active(physical)
             ):
                 return False
-
-            control = dict(self._application_control_session.snapshot())
-            if control.get("state") not in {
-                "connection-ready",
-                "workspace-ready",
-                "project-ready",
-            }:
-                return False
             verified = control.get("verified_control")
             supervisor = self._connection_supervisor.snapshot()
             structurally_current = bool(
@@ -5618,6 +5629,14 @@ class XgridsK1CompatibilityService:
         intent: Literal["select-device", "change-network"],
         transition_gate_owned: bool,
         active_binding: Mapping[str, object] | None = None,
+        runtime_snapshot: Mapping[str, object] | None = None,
+        physical_command_snapshot: Mapping[str, object] | None = None,
+        control_session_snapshot: Mapping[str, object] | None = None,
+        native_runtime_snapshot: Mapping[str, object] | None = None,
+        semantic_topology_snapshot: Mapping[str, object] | None = None,
+        identity_pin_snapshot: Mapping[str, object] | None = None,
+        provisioning_idempotency_snapshot: Mapping[str, object] | None = None,
+        network_ledger_snapshot: NetworkMutationLedgerSnapshot | None = None,
     ) -> list[str]:
         """Return one factual Bridge-only pre-START handoff admission policy."""
 
@@ -5631,14 +5650,44 @@ class XgridsK1CompatibilityService:
             configured_mode = self._connection_mode
             pending_required_mode = self._connection_reconfiguration_required_connection_mode
             lifecycle_holders = set(self._application_control_process_lease_holders)
-        runtime = self.runtime.snapshot()
+        runtime = (
-        physical = self._physical_command_coordinator.snapshot()
+            runtime_snapshot if runtime_snapshot is not None else self.runtime.snapshot()
-        control = self._application_control_session.snapshot()
+        )
-        ble_runtime = ble_runtime_snapshot()
+        physical = (
-        semantic = self._semantic_topology_public_snapshot()
+            physical_command_snapshot
-        identity = self._device_identity_pin_public_snapshot()
+            if physical_command_snapshot is not None
-        idempotency = self._network_provisioning_idempotency_public_snapshot()
+            else self._physical_command_coordinator.snapshot()
-        network_ledger = self._network_mutation_ledger.snapshot()
+        )
+        control = (
+            control_session_snapshot
+            if control_session_snapshot is not None
+            else self._application_control_session.snapshot()
+        )
+        ble_runtime = (
+            native_runtime_snapshot
+            if native_runtime_snapshot is not None
+            else ble_runtime_snapshot()
+        )
+        semantic = (
+            semantic_topology_snapshot
+            if semantic_topology_snapshot is not None
+            else self._semantic_topology_public_snapshot()
+        )
+        identity = (
+            identity_pin_snapshot
+            if identity_pin_snapshot is not None
+            else self._device_identity_pin_public_snapshot()
+        )
+        idempotency = (
+            provisioning_idempotency_snapshot
+            if provisioning_idempotency_snapshot is not None
+            else self._network_provisioning_idempotency_public_snapshot()
+        )
+        network_ledger = (
+            network_ledger_snapshot
+            if network_ledger_snapshot is not None
+            else self._network_mutation_ledger.snapshot()
+        )
 
         reasons: list[str] = []
         if not transition_gate_owned:
@@ -5741,6 +5790,10 @@ class XgridsK1CompatibilityService:
         self,
         *,
         allow_owned_network_holder: bool = False,
+        physical_command_snapshot: Mapping[str, object] | None = None,
+        runtime_snapshot: Mapping[str, object] | None = None,
+        control_session_snapshot: Mapping[str, object] | None = None,
+        native_runtime_snapshot: Mapping[str, object] | None = None,
     ) -> dict[str, object]:
         """Project one exact local-only retirement checkpoint.
 
@@ -5750,7 +5803,11 @@ class XgridsK1CompatibilityService:
         ledger performs the final operation/revision/transport CAS.
         """
 
-        physical = self._physical_command_coordinator.snapshot()
+        physical = (
+            physical_command_snapshot
+            if physical_command_snapshot is not None
+            else self._physical_command_coordinator.snapshot()
+        )
         record = physical.get("record")
         with self._lock:
             provisioning_active = self._provisioning_active
@@ -5761,9 +5818,19 @@ class XgridsK1CompatibilityService:
             reconfiguration_active = self._connection_reconfiguration_intent is not None
             pending_local_control_retirement = self._pending_local_control_retirement
             lifecycle_holders = set(self._application_control_process_lease_holders)
-        runtime = self.runtime.snapshot()
+        runtime = (
-        control = self._application_control_session.snapshot()
+            runtime_snapshot if runtime_snapshot is not None else self.runtime.snapshot()
-        native = ble_runtime_snapshot()
+        )
+        control = (
+            control_session_snapshot
+            if control_session_snapshot is not None
+            else self._application_control_session.snapshot()
+        )
+        native = (
+            native_runtime_snapshot
+            if native_runtime_snapshot is not None
+            else ble_runtime_snapshot()
+        )
 
         reasons: list[str] = []
         retireable_ambiguous = bool(
@@ -5892,10 +5959,23 @@ class XgridsK1CompatibilityService:
         self,
         *,
         allow_owned_network_holder: bool = False,
+        physical_command_snapshot: Mapping[str, object] | None = None,
+        runtime_snapshot: Mapping[str, object] | None = None,
+        control_session_snapshot: Mapping[str, object] | None = None,
+        native_runtime_snapshot: Mapping[str, object] | None = None,
+        supervisor_snapshot: ConnectionSupervisorSnapshot | None = None,
+        network_ledger_snapshot: NetworkMutationLedgerSnapshot | None = None,
+        provisioning_idempotency_snapshot: Mapping[str, object] | None = None,
+        semantic_topology_snapshot: Mapping[str, object] | None = None,
+        identity_pin_snapshot: Mapping[str, object] | None = None,
     ) -> dict[str, object]:
         """Project one exact fresh UUID checkpoint for local retirement reopen."""
 
-        physical = self._physical_command_coordinator.snapshot()
+        physical = (
+            physical_command_snapshot
+            if physical_command_snapshot is not None
+            else self._physical_command_coordinator.snapshot()
+        )
         record = physical.get("record")
         with self._lock:
             provisioning_active = self._provisioning_active
@@ -5910,14 +5990,44 @@ class XgridsK1CompatibilityService:
             fresh_devices = self._fresh_ble_devices_locked()
             discovery_generation = self._ble_discovery_generation
             generation_floors = dict(self._physical_retirement_reopen_generation_floors)
-        runtime = self.runtime.snapshot()
+        runtime = (
-        control = self._application_control_session.snapshot()
+            runtime_snapshot if runtime_snapshot is not None else self.runtime.snapshot()
-        native = ble_runtime_snapshot()
+        )
-        supervisor = self._connection_supervisor.snapshot()
+        control = (
-        network_ledger = self._network_mutation_ledger.snapshot()
+            control_session_snapshot
-        idempotency = self._network_provisioning_idempotency_public_snapshot()
+            if control_session_snapshot is not None
-        semantic_topology = self._semantic_topology_public_snapshot()
+            else self._application_control_session.snapshot()
-        identity_pins = self._device_identity_pin_public_snapshot()
+        )
+        native = (
+            native_runtime_snapshot
+            if native_runtime_snapshot is not None
+            else ble_runtime_snapshot()
+        )
+        supervisor = (
+            supervisor_snapshot
+            if supervisor_snapshot is not None
+            else self._connection_supervisor.snapshot()
+        )
+        network_ledger = (
+            network_ledger_snapshot
+            if network_ledger_snapshot is not None
+            else self._network_mutation_ledger.snapshot()
+        )
+        idempotency = (
+            provisioning_idempotency_snapshot
+            if provisioning_idempotency_snapshot is not None
+            else self._network_provisioning_idempotency_public_snapshot()
+        )
+        semantic_topology = (
+            semantic_topology_snapshot
+            if semantic_topology_snapshot is not None
+            else self._semantic_topology_public_snapshot()
+        )
+        identity_pins = (
+            identity_pin_snapshot
+            if identity_pin_snapshot is not None
+            else self._device_identity_pin_public_snapshot()
+        )
 
         active_retirements = (
             self._active_physical_retirement_documents(record)
@@ -6925,6 +7035,45 @@ class XgridsK1CompatibilityService:
             binding_key,
         )
 
+    def _control_session_snapshot_without_physical_command(self) -> dict[str, object]:
+        """Read volatile control facts without recursively reloading the ledger."""
+
+        snapshotter = getattr(
+            self._application_control_session,
+            "snapshot_control_only",
+            None,
+        )
+        snapshot = (
+            snapshotter()
+            if callable(snapshotter)
+            else self._application_control_session.snapshot()
+        )
+        if not isinstance(snapshot, Mapping):
+            raise TypeError("application control session snapshot must be a mapping")
+        return dict(snapshot)
+
+    def _control_session_snapshot_with_physical_command(
+        self,
+        physical_command: Mapping[str, object],
+        *,
+        fallback_snapshot: Mapping[str, object],
+    ) -> dict[str, object]:
+        """Publish one control/physical join without a second ledger reload."""
+
+        snapshotter = getattr(
+            self._application_control_session,
+            "snapshot_with_physical_command",
+            None,
+        )
+        if callable(snapshotter):
+            snapshot = snapshotter(physical_command)
+            if not isinstance(snapshot, Mapping):
+                raise TypeError("application control session snapshot must be a mapping")
+            return dict(snapshot)
+        snapshot = dict(fallback_snapshot)
+        snapshot["physical_command"] = dict(physical_command)
+        return snapshot
+
     def require_snapshot_runtime_id(self, expected_snapshot_runtime_id: str) -> None:
         """Reject an action issued by a browser bound to an older service.
 
@@ -6943,7 +7092,7 @@ class XgridsK1CompatibilityService:
         self._retry_pending_local_control_retirement()
         application_control = self._application_control.snapshot().as_dict()
         application_control_session = _application_control_session_public_snapshot(
-            self._application_control_session.snapshot()
+            self._control_session_snapshot_without_physical_command()
         )
         runtime = self.runtime.snapshot()
         camera_preview = self.camera_preview.snapshot()
@@ -6952,7 +7101,7 @@ class XgridsK1CompatibilityService:
         self._reconcile_connection_supervisor(application_control_session, runtime)
         if self._retire_orphaned_prestart_control_owner():
             application_control_session = _application_control_session_public_snapshot(
-                self._application_control_session.snapshot()
+                self._control_session_snapshot_without_physical_command()
             )
             self._reconcile_application_control_process_lease(application_control_session)
         # Preserve the terminal pre-START proof long enough to retire a local
@@ -6998,7 +7147,7 @@ class XgridsK1CompatibilityService:
         # owner to idle. Continue the same atomic snapshot from that factual
         # state instead of returning the stale pre-retirement failure row.
         application_control_session = _application_control_session_public_snapshot(
-            self._application_control_session.snapshot()
+            self._control_session_snapshot_without_physical_command()
         )
         self._reconcile_acquisition(
             runtime,
@@ -7013,12 +7162,12 @@ class XgridsK1CompatibilityService:
         # control owner. Publish that factual post-reduction state in the same
         # snapshot so passive BLE/read-only recovery is available immediately.
         application_control_session = _application_control_session_public_snapshot(
-            self._application_control_session.snapshot()
+            self._control_session_snapshot_without_physical_command()
         )
         self._reconcile_application_control_process_lease(application_control_session)
         if self._retire_terminal_prestart_control_failure():
             application_control_session = _application_control_session_public_snapshot(
-                self._application_control_session.snapshot()
+                self._control_session_snapshot_without_physical_command()
             )
             self._reconcile_application_control_process_lease(application_control_session)
         runtime = self.runtime.snapshot()
@@ -7038,14 +7187,35 @@ class XgridsK1CompatibilityService:
         active_acquisition_checkpoint = (
             self._active_acquisition_checkpoint_public_snapshot()
         )
-        operator_retirement = self._physical_operator_retirement_projection()
+        ble_runtime = ble_runtime_snapshot()
+        operator_retirement = self._physical_operator_retirement_projection(
+            physical_command_snapshot=physical_command,
+            runtime_snapshot=runtime,
+            control_session_snapshot=application_control_session,
+            native_runtime_snapshot=ble_runtime,
+        )
         physical_command["operator_retirement"] = operator_retirement
         physical_command["operator_retirement_allowed"] = operator_retirement["allowed"]
         physical_command["operator_retirement_reason_codes"] = operator_retirement["reason_codes"]
         physical_command["operator_reconciliation_reopen"] = (
-            self._physical_operator_reconciliation_reopen_projection()
+            self._physical_operator_reconciliation_reopen_projection(
+                physical_command_snapshot=physical_command,
+                runtime_snapshot=runtime,
+                control_session_snapshot=application_control_session,
+                native_runtime_snapshot=ble_runtime,
+                supervisor_snapshot=supervisor_snapshot,
+                network_ledger_snapshot=ledger_snapshot,
+                provisioning_idempotency_snapshot=idempotency_snapshot,
+                semantic_topology_snapshot=semantic_topology_store,
+                identity_pin_snapshot=device_identity_pin_store,
+            )
+        )
+        application_control_session = _application_control_session_public_snapshot(
+            self._control_session_snapshot_with_physical_command(
+                physical_command,
+                fallback_snapshot=application_control_session,
+            )
         )
-        ble_runtime = ble_runtime_snapshot()
         with self._lock:
             operation_phase = self._operation_phase
             operation_message = self._operation_message
@@ -7318,11 +7488,27 @@ class XgridsK1CompatibilityService:
             intent="select-device",
             transition_gate_owned=False,
             active_binding=active_binding,
+            runtime_snapshot=runtime,
+            physical_command_snapshot=physical_command,
+            control_session_snapshot=application_control_session,
+            native_runtime_snapshot=ble_runtime,
+            semantic_topology_snapshot=semantic_topology_store,
+            identity_pin_snapshot=device_identity_pin_store,
+            provisioning_idempotency_snapshot=idempotency_snapshot,
+            network_ledger_snapshot=ledger_snapshot,
         )
         change_network_reasons = self._connection_reconfiguration_safety_reasons(
             intent="change-network",
             transition_gate_owned=False,
             active_binding=active_binding,
+            runtime_snapshot=runtime,
+            physical_command_snapshot=physical_command,
+            control_session_snapshot=application_control_session,
+            native_runtime_snapshot=ble_runtime,
+            semantic_topology_snapshot=semantic_topology_store,
+            identity_pin_snapshot=device_identity_pin_store,
+            provisioning_idempotency_snapshot=idempotency_snapshot,
+            network_ledger_snapshot=ledger_snapshot,
         )
 
         def reconfiguration_decision(
@@ -14622,8 +14808,13 @@ class XgridsK1CompatibilityService:
                 # deliberately bypasses this ordinary idempotence fast path.
                 return
 
+        camera_admission_started = time.monotonic()
         physical = self._physical_command_coordinator.snapshot()
-        control = self._application_control_session.snapshot()
+        # ``physical`` above is the authoritative durable read for this
+        # admission attempt.  Control facts are process-local; asking the
+        # public control snapshot for them would recursively reload the same
+        # physical ledger while this 10 Hz publisher callback is blocked.
+        control = self._control_session_snapshot_without_physical_command()
 
         if control_mode == "plugin-commanded":
             start_operation_id = (
@@ -14753,6 +14944,17 @@ class XgridsK1CompatibilityService:
                         time.monotonic() + CAMERA_POST_PCL_ACTIVATION_RETRY_SECONDS
                     )
             raise
+        logger.info(
+            "K1 first-PCL camera admission timing",
+            extra={
+                "event_code": "k1_camera_pcl_admission_timing",
+                "evidence_session_id": out_dir.name,
+                "camera_pcl_admission_ms": int(
+                    round((time.monotonic() - camera_admission_started) * 1_000)
+                ),
+                "device_command_sent": False,
+            },
+        )
 
     def _reset_live_data_plane_observation(self) -> None:
         """Make a stopped/replaced producer incapable of authorizing new data."""
@@ -22237,7 +22439,10 @@ class XgridsK1CompatibilityService:
                     local_start_operation_id=start_operation_id,
                 )
             )
-            control = self._application_control_session.snapshot()
+            # Physical authority was loaded explicitly just above.  Do not
+            # recursively reload it through the control projection while the
+            # first-PCL camera worker is trying to leave the ingress path.
+            control = self._control_session_snapshot_without_physical_command()
             verified_control_value = control.get("verified_control")
             verified_control = (
                 cast(Mapping[str, Any], verified_control_value)
@@ -22299,28 +22504,37 @@ class XgridsK1CompatibilityService:
                 camera,
                 evidence_session_id=out_dir.name,
             )
+            activation_claims: list[_CameraProducerActivationClaim] = []
+
+            def reserve_activation(reserve: Callable[[], bool]) -> bool:
+                claim = self._reserve_camera_producer_activation_if_still_active(
+                    acquisition_id=acquisition.acquisition_id,
+                    evidence_session_id=out_dir.name,
+                    start_operation_id=start_operation_id,
+                    runtime_producer_generation=expected_runtime_generation,
+                    reserve=reserve,
+                )
+                if claim is None:
+                    return False
+                activation_claims.append(claim)
+                return True
+
+            def commit_activation(commit: Callable[[], bool]) -> bool:
+                if len(activation_claims) != 1:
+                    return False
+                return self._commit_reserved_camera_producer_activation(
+                    activation_claims[0],
+                    commit,
+                )
+
             if partial_same_session:
                 camera = self.camera_preview.retry_recording_producer(
                     DEFAULT_ACQUISITION_CAMERA_SOURCE,
                     target,
                     expected_generation=cast(int, camera["generation"]),
                     expected_recording_session=out_dir.name,
-                    pre_retry_fence=lambda reserve: (
+                    pre_retry_fence=reserve_activation,
-                        self._reserve_camera_restart_if_still_active(
+                    commit_fence=commit_activation,
-                            acquisition_id=acquisition.acquisition_id,
-                            evidence_session_id=out_dir.name,
-                            start_operation_id=start_operation_id,
-                            runtime_producer_generation=expected_runtime_generation,
-                            reserve=reserve,
-                        )
-                    ),
-                    commit_fence=lambda commit: self._commit_camera_restart_if_still_active(
-                        acquisition_id=acquisition.acquisition_id,
-                        evidence_session_id=out_dir.name,
-                        start_operation_id=start_operation_id,
-                        runtime_producer_generation=expected_runtime_generation,
-                        commit=commit,
-                    ),
                     committed_before_start=lambda committed: (
                         self._bind_live_perception_camera(out_dir.name, committed)
                     ),
@@ -22330,13 +22544,8 @@ class XgridsK1CompatibilityService:
                     DEFAULT_ACQUISITION_CAMERA_SOURCE,
                     target,
                     out_dir,
-                    commit_fence=lambda commit: self._commit_camera_restart_if_still_active(
+                    pre_prepare_fence=reserve_activation,
-                        acquisition_id=acquisition.acquisition_id,
+                    commit_fence=commit_activation,
-                        evidence_session_id=out_dir.name,
-                        start_operation_id=start_operation_id,
-                        runtime_producer_generation=expected_runtime_generation,
-                        commit=commit,
-                    ),
                     committed_before_start=lambda committed: (
                         self._bind_live_perception_camera(out_dir.name, committed)
                     ),
@@ -24378,6 +24587,104 @@ class XgridsK1CompatibilityService:
             action=reserve,
         )
 
+    def _reserve_camera_producer_activation_if_still_active(
+        self,
+        *,
+        acquisition_id: str,
+        evidence_session_id: str,
+        start_operation_id: str | None,
+        runtime_producer_generation: int,
+        reserve: Callable[[], bool],
+    ) -> _CameraProducerActivationClaim | None:
+        """Validate durable authority before Popen and freeze its local CAS."""
+
+        claims: list[_CameraProducerActivationClaim] = []
+
+        def reserve_and_capture() -> bool:
+            if not reserve():
+                return False
+            with self._lock:
+                claims.append(
+                    _CameraProducerActivationClaim(
+                        acquisition_id=acquisition_id,
+                        evidence_session_id=evidence_session_id,
+                        start_operation_id=start_operation_id,
+                        stop_operation_id=self._acquisition_stop_operation_id,
+                        runtime_producer_generation=runtime_producer_generation,
+                        recovery_generation=self._active_stream_recovery_generation,
+                    )
+                )
+            return True
+
+        accepted = self._camera_restart_action_if_still_active(
+            acquisition_id=acquisition_id,
+            evidence_session_id=evidence_session_id,
+            start_operation_id=start_operation_id,
+            runtime_producer_generation=runtime_producer_generation,
+            action=reserve_and_capture,
+        )
+        if not accepted:
+            return None
+        if len(claims) != 1:
+            raise RuntimeError("camera activation authority was not reserved exactly once")
+        return claims[0]
+
+    def _commit_reserved_camera_producer_activation(
+        self,
+        claim: _CameraProducerActivationClaim,
+        commit: Callable[[], bool],
+    ) -> bool:
+        """Commit a prepared FFmpeg candidate without durable reads or device I/O.
+
+        Durable physical/control/checkpoint authority was proved before Popen.
+        After Popen, a new STOP, acquisition replacement, runtime replacement or
+        recovery generation invalidates that frozen proof.  The remaining gate
+        is deliberately in-memory so stdout/stderr readers start immediately.
+        """
+
+        with self._camera_restart_commit_gate:
+            if self._camera_stop_priority_counts.get(claim.acquisition_id, 0) > 0:
+                return False
+            with self._acquisition_lifecycle_access():
+                runtime = self.runtime.snapshot()
+                with self._lock:
+                    acquisition = self._acquisition
+                    out_dir = self._acquisition_out_dir
+                    current_start_operation_id = self._acquisition_start_operation_id
+                    current_stop_operation_id = self._acquisition_stop_operation_id
+                    recovery_generation = self._active_stream_recovery_generation
+                    recovery_state = self._active_stream_recovery_state
+                local_current = bool(
+                    acquisition is not None
+                    and acquisition.acquisition_id == claim.acquisition_id
+                    and acquisition.state
+                    in {"starting", "awaiting_external_start", "acquiring"}
+                    and out_dir is not None
+                    and out_dir.name == claim.evidence_session_id
+                    and current_start_operation_id
+                    in {None, claim.start_operation_id}
+                    and current_stop_operation_id == claim.stop_operation_id
+                    and recovery_generation == claim.recovery_generation
+                    and recovery_state
+                    not in {
+                        "reconnecting",
+                        "blocked",
+                        "fault",
+                        "force-finishing",
+                        "force-finished",
+                    }
+                )
+                runtime_current = bool(
+                    runtime.get("phase") == "live"
+                    and runtime.get("source_mode") == "live"
+                    and runtime.get("source_ready") is True
+                    and runtime.get("producer_generation")
+                    == claim.runtime_producer_generation
+                )
+                if not local_current or not runtime_current:
+                    return False
+                return commit()
+
     def _commit_camera_restart_if_still_active(
         self,
         *,
@@ -24421,7 +24728,10 @@ class XgridsK1CompatibilityService:
         # only bounded in-process lineage checks and the camera-local commit
         # remain.
         physical = self._physical_command_coordinator.snapshot()
-        control = self._application_control_session.snapshot()
+        # The exact physical proof is already frozen for this pre-Popen
+        # fence.  Only volatile MQTT dialogue facts are needed here; a nested
+        # physical reload adds no authority and delays camera reader startup.
+        control = self._control_session_snapshot_without_physical_command()
         verified_control_value = control.get("verified_control")
         verified_control = (
             cast(Mapping[str, Any], verified_control_value)

src/k1link/device_plugins/xgrids_k1/physical_command_ledger.py

@@ -960,6 +960,27 @@ class PhysicalCommandLedgerSnapshot:
         return False
 
 
+_PhysicalCommandStatFingerprint = tuple[int, int, int, int, int, int, int, int, int]
+
+
+@dataclass(frozen=True, slots=True)
+class _PhysicalCommandLedgerFilesystemFingerprint:
+    """Cheap identity fence for one already-verified read-only snapshot.
+
+    The physical ledger remains fully reloaded for every mutation and proof.
+    Passive snapshots may reuse the in-memory, cryptographically verified
+    document only while the ledger directory, main file, archive directory and
+    every bounded archive entry retain their exact filesystem identities.
+    ``ctime_ns`` deliberately prevents a same-size rewrite with restored mtime
+    from bypassing the fence.
+    """
+
+    ledger_directory: _PhysicalCommandStatFingerprint | None
+    ledger_file: _PhysicalCommandStatFingerprint | None
+    archive_directory: _PhysicalCommandStatFingerprint | None
+    archive_entries: tuple[tuple[str, _PhysicalCommandStatFingerprint], ...]
+
+
 def active_operator_retirements(
     record: PhysicalCommandRecord,
 ) -> tuple[PhysicalCommandOperatorRetirement, ...]:
@@ -1255,12 +1276,24 @@ class PhysicalCommandLedger:
         self._record: PhysicalCommandRecord | None = None
         self._archive_history = _PhysicalCommandArchiveHistory()
         self._corrupt = False
+        self._snapshot_filesystem_fingerprint: (
+            _PhysicalCommandLedgerFilesystemFingerprint | None
+        ) = None
         with self._lock, self._process_lock_locked():
             self._reload_locked()
 
     def snapshot(self) -> PhysicalCommandLedgerSnapshot:
         with self._lock, self._process_lock_locked():
-            self._reload_locked()
+            current_fingerprint = self._snapshot_fingerprint_locked()
+            if (
+                current_fingerprint is None
+                or current_fingerprint != self._snapshot_filesystem_fingerprint
+            ):
+                self._reload_locked()
+                if not self._corrupt:
+                    self._snapshot_filesystem_fingerprint = (
+                        self._snapshot_fingerprint_locked()
+                    )
             if self._corrupt:
                 return PhysicalCommandLedgerSnapshot(
                     status="corrupt",
@@ -3072,6 +3105,7 @@ class PhysicalCommandLedger:
         self._record = plan.record
         self._archive_history = plan.history
         self._corrupt = False
+        self._snapshot_filesystem_fingerprint = None
         return plan.record
 
     def _publish_archive_plan_locked(
@@ -3157,6 +3191,10 @@ class PhysicalCommandLedger:
         )
 
     def _reload_locked(self) -> None:
+        # Every command/proof path calls the full loader.  It must invalidate a
+        # passive snapshot fingerprint even when the subsequent transition is
+        # rejected, so the next read establishes a fresh verified generation.
+        self._snapshot_filesystem_fingerprint = None
         try:
             parent = self.path.parent.lstat()
             _require_private_directory_metadata(parent, label="ledger directory")
@@ -3223,6 +3261,86 @@ class PhysicalCommandLedger:
         self._archive_history = archive_history
         self._corrupt = False
 
+    def _snapshot_fingerprint_locked(
+        self,
+    ) -> _PhysicalCommandLedgerFilesystemFingerprint | None:
+        """Capture a bounded metadata generation or decline cache reuse.
+
+        Failure never authorizes stale state: callers fall back to the complete
+        loader, whose established behavior is to mark unsafe evidence corrupt.
+        The process-wide flock is already held by every caller, serializing all
+        cooperating writers across Mission Core processes.
+        """
+
+        try:
+            try:
+                ledger_directory_metadata = self.path.parent.lstat()
+            except FileNotFoundError:
+                ledger_directory_metadata = None
+            if ledger_directory_metadata is not None:
+                _require_private_directory_metadata(
+                    ledger_directory_metadata,
+                    label="ledger directory",
+                )
+
+            try:
+                ledger_metadata = self.path.lstat()
+            except FileNotFoundError:
+                ledger_metadata = None
+            if ledger_metadata is not None:
+                _require_private_regular_file(
+                    ledger_metadata,
+                    label="ledger",
+                    empty=False,
+                )
+                if ledger_metadata.st_size > PHYSICAL_COMMAND_LEDGER_MAX_BYTES:
+                    return None
+
+            try:
+                archive_directory_metadata = self._archive_dir.lstat()
+            except FileNotFoundError:
+                archive_directory_metadata = None
+            archive_entries: list[
+                tuple[str, _PhysicalCommandStatFingerprint]
+            ] = []
+            if archive_directory_metadata is not None:
+                _require_private_directory_metadata(
+                    archive_directory_metadata,
+                    label="archive directory",
+                )
+                with os.scandir(self._archive_dir) as entries:
+                    for entry in entries:
+                        if len(archive_entries) >= PHYSICAL_COMMAND_ARCHIVE_MAX_SEGMENTS + 2:
+                            return None
+                        metadata = entry.stat(follow_symlinks=False)
+                        if not stat.S_ISREG(metadata.st_mode):
+                            return None
+                        archive_entries.append(
+                            (entry.name, _physical_command_stat_fingerprint(metadata))
+                        )
+                archive_entries.sort(key=lambda item: item[0])
+
+            return _PhysicalCommandLedgerFilesystemFingerprint(
+                ledger_directory=(
+                    _physical_command_stat_fingerprint(ledger_directory_metadata)
+                    if ledger_directory_metadata is not None
+                    else None
+                ),
+                ledger_file=(
+                    _physical_command_stat_fingerprint(ledger_metadata)
+                    if ledger_metadata is not None
+                    else None
+                ),
+                archive_directory=(
+                    _physical_command_stat_fingerprint(archive_directory_metadata)
+                    if archive_directory_metadata is not None
+                    else None
+                ),
+                archive_entries=tuple(archive_entries),
+            )
+        except (OSError, ValueError):
+            return None
+
     def _load_archive_history_locked(
         self,
         record: PhysicalCommandRecord,
@@ -3233,6 +3351,22 @@ class PhysicalCommandLedger:
         )
 
 
+def _physical_command_stat_fingerprint(
+    value: os.stat_result,
+) -> _PhysicalCommandStatFingerprint:
+    return (
+        value.st_dev,
+        value.st_ino,
+        value.st_mode,
+        value.st_uid,
+        value.st_gid,
+        value.st_nlink,
+        value.st_size,
+        value.st_mtime_ns,
+        value.st_ctime_ns,
+    )
+
+
 def _require_safe_edge_successor(
     previous: PhysicalCommandRecord | None,
     *,

src/k1link/device_plugins/xgrids_k1/protocol/application_session.py

@@ -661,6 +661,14 @@ class InteractiveApplicationControlSession:
                     )
                     self._set_phase_locked("stop-requested")
                     self._stop_requested.set()
+                    logger.info(
+                        "K1 STOP timing checkpoint",
+                        extra={
+                            "event_code": "k1_stop_dispatch_timing",
+                            "operation_stage": "prepared-worker-released",
+                            "device_command_sent": False,
+                        },
+                    )
             except BaseException as preparation_error:
                 if coordinator is not None:
                     try:
@@ -781,41 +789,66 @@ class InteractiveApplicationControlSession:
 
     def snapshot(self) -> dict[str, object]:
         with self._lock:
-            transport_snapshot = self._live_transport_snapshot_locked()
+            physical_command = (
-            phase = self._phase
+                self._physical_command_coordinator.snapshot()
-            return {
+                if self._physical_command_coordinator is not None
-                "mode": "interactive-canonical",
+                else None
-                "inspection_only": self._inspection_only,
+            )
-                "inspection_promotion_allowed": self._inspection_promotion_allowed,
+            return self._snapshot_locked(physical_command=physical_command)
-                "state": phase,
+
-                "session_generation": self._run_generation,
+    def snapshot_control_only(self) -> dict[str, object]:
-                "state_revision": self._state_revision,
+        """Return process-local control facts without reloading the durable ledger."""
-                "control_socket_open": phase not in {"idle", "completed", "closed", "failed"},
+
-                "can_open": self._can_open_locked(),
+        with self._lock:
-                "can_enter_workspace": phase == "connection-ready",
+            return self._snapshot_locked(physical_command=None)
-                "can_prepare_project": phase == "workspace-ready",
+
-                "can_start": phase == "project-ready",
+    def snapshot_with_physical_command(
-                "can_stop": phase == "scanning",
+        self,
-                # Retained for wire compatibility with the v1alpha2 snapshot.
+        physical_command: Mapping[str, object] | None,
-                # Standby is now concluded solely from live K1 protocol state.
+    ) -> dict[str, object]:
-                "can_confirm_standby": False,
+        """Join one already-verified physical snapshot to current control facts."""
-                "pending_operator_action": self._pending_operator_action_locked(),
+
-                "scripted_transitions": False,
+        with self._lock:
-                "automatic_retry": False,
+            return self._snapshot_locked(physical_command=physical_command)
-                "outcome_unknown": self._outcome_unknown,
+
-                "scanning_observer_errors": self._scanning_observer_errors,
+    def _snapshot_locked(
-                "failure": dict(self._failure) if self._failure is not None else None,
+        self,
-                "dialogue": (
+        *,
-                    dict(self._dialogue_snapshot) if self._dialogue_snapshot is not None else None
+        physical_command: Mapping[str, object] | None,
-                ),
+    ) -> dict[str, object]:
-                "transport": transport_snapshot,
+        transport_snapshot = self._live_transport_snapshot_locked()
-                "verified_control": (self._verified_control_snapshot_locked(transport_snapshot)),
+        phase = self._phase
-                "physical_command": (
+        return {
-                    self._physical_command_coordinator.snapshot()
+            "mode": "interactive-canonical",
-                    if self._physical_command_coordinator is not None
+            "inspection_only": self._inspection_only,
-                    else None
+            "inspection_promotion_allowed": self._inspection_promotion_allowed,
-                ),
+            "state": phase,
-            }
+            "session_generation": self._run_generation,
+            "state_revision": self._state_revision,
+            "control_socket_open": phase not in {"idle", "completed", "closed", "failed"},
+            "can_open": self._can_open_locked(),
+            "can_enter_workspace": phase == "connection-ready",
+            "can_prepare_project": phase == "workspace-ready",
+            "can_start": phase == "project-ready",
+            "can_stop": phase == "scanning",
+            # Retained for wire compatibility with the v1alpha2 snapshot.
+            # Standby is now concluded solely from live K1 protocol state.
+            "can_confirm_standby": False,
+            "pending_operator_action": self._pending_operator_action_locked(),
+            "scripted_transitions": False,
+            "automatic_retry": False,
+            "outcome_unknown": self._outcome_unknown,
+            "scanning_observer_errors": self._scanning_observer_errors,
+            "failure": dict(self._failure) if self._failure is not None else None,
+            "dialogue": (
+                dict(self._dialogue_snapshot) if self._dialogue_snapshot is not None else None
+            ),
+            "transport": transport_snapshot,
+            "verified_control": self._verified_control_snapshot_locked(transport_snapshot),
+            "physical_command": (
+                dict(physical_command) if physical_command is not None else None
+            ),
+        }
 
     def _run(self, generation: int) -> None:
         executor: PhysicalAcceptanceDialogueExecutor | None = None
@@ -1028,6 +1061,14 @@ class InteractiveApplicationControlSession:
                 stop_dispatch_admission_deadline_reached
             )
             self._set_phase("stopping")
+            logger.info(
+                "K1 STOP timing checkpoint",
+                extra={
+                    "event_code": "k1_stop_dispatch_timing",
+                    "operation_stage": "pre-dispatch-validation-complete",
+                    "device_command_sent": False,
+                },
+            )
             executor.execute_canonical_stop(
                 stop_command,
                 stop_permit,
@@ -1038,6 +1079,14 @@ class InteractiveApplicationControlSession:
                     stop_dispatch_admission_deadline_reached
                 ),
             )
+            logger.info(
+                "K1 STOP timing checkpoint",
+                extra={
+                    "event_code": "k1_stop_dispatch_timing",
+                    "operation_stage": "correlated-application-response",
+                    "device_command_sent": True,
+                },
+            )
             self._validate_connection_binding_snapshot("stop-post-response")
             self._set_phase("awaiting-standby-confirmation")
             executor.maintain_post_stop_until_standby()

src/k1link/web/runtime_diagnostics.py

@@ -70,6 +70,12 @@ _EXTRA_FIELDS: Final = (
     "camera_queue_rejection_reason",
     "camera_retry_count",
     "camera_generation",
+    "camera_pcl_admission_ms",
+    "camera_authority_wait_ms",
+    "camera_ffmpeg_prepare_ms",
+    "camera_post_spawn_commit_ms",
+    "camera_activation_total_ms",
+    "device_command_sent",
     "websocket_close_code",
     "transport_epoch",
     "camera_append_error_name",

tests/test_viewer_diagnostics_api.py

@@ -87,6 +87,12 @@ def test_private_scanner_diagnostics_are_durable_structured_and_bounded(
             "camera_queue_bytes": 12_000_000,
             "camera_queue_segments": 96,
             "camera_retry_count": 3,
+            "camera_pcl_admission_ms": 184,
+            "camera_authority_wait_ms": 91,
+            "camera_ffmpeg_prepare_ms": 42,
+            "camera_post_spawn_commit_ms": 7,
+            "camera_activation_total_ms": 140,
+            "device_command_sent": False,
             "websocket_close_code": 4_008,
             "transport_epoch": 7,
             "unapproved_secret_field": "must-not-be-written",
@@ -108,6 +114,12 @@ def test_private_scanner_diagnostics_are_durable_structured_and_bounded(
     assert document["mqtt_loop_result_code"] == 7
     assert document["mqtt_loop_phase"] == "post-publish-drain"
     assert document["automatic_retry"] is False
+    assert document["camera_pcl_admission_ms"] == 184
+    assert document["camera_authority_wait_ms"] == 91
+    assert document["camera_ffmpeg_prepare_ms"] == 42
+    assert document["camera_post_spawn_commit_ms"] == 7
+    assert document["camera_activation_total_ms"] == 140
+    assert document["device_command_sent"] is False
     assert document["side_effect_status"] == "unknown"
     assert document["network_change_attempted"] is True
     assert document["device_write_attempted"] is True

tests/test_xgrids_acquisition_lifecycle.py

@@ -1047,6 +1047,33 @@ def service_with_fake_runtime(
     return service, runtime
 
 
+def test_passive_state_reuses_one_final_physical_snapshot(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    service, _runtime = service_with_fake_runtime(tmp_path)
+    coordinator = service._physical_command_coordinator  # noqa: SLF001
+    original_snapshot = coordinator.snapshot
+    snapshot_calls = 0
+
+    def counted_snapshot() -> dict[str, object]:
+        nonlocal snapshot_calls
+        snapshot_calls += 1
+        return original_snapshot()
+
+    monkeypatch.setattr(coordinator, "snapshot", counted_snapshot)
+
+    state = service.state()
+
+    # Two earlier reads are deliberate pre/post settlement proofs. Presentation
+    # policy and the nested control-session projection must share the one final
+    # verified durable snapshot instead of recursively reloading the ledger.
+    assert snapshot_calls == 3
+    assert state["application_control_session"]["physical_command"] == state[
+        "physical_command"
+    ]
+
+
 def test_service_owns_one_wifi_association_probe_across_monitor_recreation(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
@@ -10283,9 +10310,11 @@ def test_classified_scanning_cancels_stop_eof_and_preserves_or_rearms_camera_epo
         target: str,
         session_dir: Path,
         *,
+        pre_prepare_fence: Callable[[Callable[[], bool]], bool],
         commit_fence: Callable[[Callable[[], bool]], bool],
         committed_before_start: Callable[[dict[str, object]], None] | None = None,
     ) -> dict[str, object]:
+        assert pre_prepare_fence(lambda: True) is True
         select_camera(source_id, target)
         start_recording(session_dir)
         assert committed_before_start is not None
@@ -13763,9 +13792,11 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
         target: str,
         session_dir: Path,
         *,
+        pre_prepare_fence: Callable[[Callable[[], bool]], bool],
         commit_fence: Callable[[Callable[[], bool]], bool],
         committed_before_start: Callable[[dict[str, object]], None] | None = None,
     ) -> dict[str, object]:
+        assert pre_prepare_fence(lambda: True) is True
         events.append(("select", (source_id, target)))
         camera_state["active_source_id"] = source_id
         events.append(("record", session_dir))
@@ -13849,6 +13880,27 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
         "snapshot",
         snapshot_physical,
     )
+    full_control_snapshot_calls = 0
+    control_only_snapshot_calls = 0
+    real_control_snapshot = control.snapshot
+
+    def snapshot_control() -> dict[str, object]:
+        nonlocal full_control_snapshot_calls
+        full_control_snapshot_calls += 1
+        return real_control_snapshot()
+
+    def snapshot_control_only() -> dict[str, object]:
+        nonlocal control_only_snapshot_calls
+        control_only_snapshot_calls += 1
+        return real_control_snapshot()
+
+    monkeypatch.setattr(control, "snapshot", snapshot_control)
+    monkeypatch.setattr(
+        control,
+        "snapshot_control_only",
+        snapshot_control_only,
+        raising=False,
+    )
     runtime.pcl_frames = 1
     frame = DecodedPointCloudView(
         context=ConsumerFrameContext(
@@ -13863,6 +13915,8 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
         positions_xyz=((0.0, 0.0, 0.0),),
     )
 
+    full_snapshots_before_pcl = full_control_snapshot_calls
+    control_only_snapshots_before_pcl = control_only_snapshot_calls
     service._observe_published_runtime_envelope(  # noqa: SLF001
         frame,
         runtime.producer_generation,
@@ -13871,6 +13925,8 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
     # Public manual controls stay closed until that worker proves an exact
     # recording epoch, preventing a left/right selection race.
     assert activation_entered.wait(timeout=2.0)
+    assert full_control_snapshot_calls == full_snapshots_before_pcl
+    assert control_only_snapshot_calls == control_only_snapshots_before_pcl + 1
     activating_state = service.state()
     activating_camera_streams = [
         stream
@@ -13890,6 +13946,8 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
             )
         )
     assert events == []
+    full_snapshots_before_camera_worker = full_control_snapshot_calls
+    control_only_snapshots_before_camera_worker = control_only_snapshot_calls
     release_activation.set()
     deadline = time.monotonic() + 2.0
     while len(events) < 2 and time.monotonic() < deadline:
@@ -13899,6 +13957,8 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera(
         ("select", ("sensor.camera.right", "192.168.1.20")),
         ("record", out_dir),
     ]
+    assert full_control_snapshot_calls == full_snapshots_before_camera_worker
+    assert control_only_snapshot_calls == control_only_snapshots_before_camera_worker + 2
     assert camera_state["active_source_id"] == "sensor.camera.right"
     assert camera_state["recording"] == {
         "active": True,
@@ -14068,9 +14128,11 @@ def test_failed_post_pcl_camera_activation_retries_on_later_authoritative_frame(
         target: str,
         session_dir: Path,
         *,
+        pre_prepare_fence: Callable[[Callable[[], bool]], bool],
         commit_fence: Callable[[Callable[[], bool]], bool],
         committed_before_start: Callable[[dict[str, object]], None] | None = None,
     ) -> dict[str, object]:
+        assert pre_prepare_fence(lambda: True) is True
         del commit_fence, committed_before_start
         select_calls.append((source_id, target))
         recording_calls.append(session_dir)
@@ -14827,10 +14889,12 @@ def test_stop_priority_overtakes_blocked_initial_camera_popen(
         _target: str,
         _session_dir: Path,
         *,
+        pre_prepare_fence: Callable[[Callable[[], bool]], bool],
         commit_fence: Callable[[Callable[[], bool]], bool],
         committed_before_start: Callable[[dict[str, object]], None] | None = None,
     ) -> dict[str, object]:
         del committed_before_start
+        assert pre_prepare_fence(lambda: True) is True
         candidate_blocked.set()
         assert release_candidate.wait(3.0)
         if not commit_fence(lambda: candidate_committed.append(True) or True):

tests/test_xgrids_application_session.py

@@ -193,6 +193,7 @@ class FakePhysicalCommandCoordinator:
         self.resolutions: list[tuple[str, str | None]] = []
         self.phase_probe: Callable[[], str] | None = None
         self.snapshot_override: dict[str, object] | None = None
+        self.snapshot_calls = 0
 
     def bind_control_session(self, binding: PhysicalCommandRuntimeBinding) -> None:
         self.bindings.append(binding)
@@ -217,6 +218,7 @@ class FakePhysicalCommandCoordinator:
         self.resolutions.append((f"{action}-not-dispatched", phase))
 
     def snapshot(self) -> dict[str, object]:
+        self.snapshot_calls += 1
         if self.snapshot_override is not None:
             return self.snapshot_override
         return {
@@ -433,6 +435,36 @@ def _successful_start_checkpoint_observer(
     return None
 
 
+def test_control_only_snapshot_reuses_preverified_physical_command() -> None:
+    coordinator = FakePhysicalCommandCoordinator()
+    session = InteractiveApplicationControlSession(
+        FakeAuthorityLoader(),
+        transport_factory=FakeTransport,  # type: ignore[arg-type]
+        physical_command_coordinator=coordinator,  # type: ignore[arg-type]
+    )
+
+    control_only = session.snapshot_control_only()
+    assert control_only["state"] == "idle"
+    assert control_only["physical_command"] is None
+    assert coordinator.snapshot_calls == 0
+
+    verified_physical = {
+        "status": "resolved",
+        "requires_reconciliation": False,
+    }
+    joined = session.snapshot_with_physical_command(verified_physical)
+    assert joined["physical_command"] == verified_physical
+    assert coordinator.snapshot_calls == 0
+
+    regular = session.snapshot()
+    assert regular["physical_command"] == {
+        "status": "test",
+        "requires_reconciliation": False,
+        "automatic_replay_allowed": False,
+    }
+    assert coordinator.snapshot_calls == 1
+
+
 def test_canonical_stages_require_operator_events_but_device_standby_does_not(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

tests/test_xgrids_camera_gateway.py

@@ -6,6 +6,7 @@ import json
 import sys
 import threading
 import time
+from collections.abc import Callable
 from io import BytesIO
 from pathlib import Path
 
@@ -1416,16 +1417,60 @@ def test_source_end_none_epoch_cas_waits_for_canonical_archive_seal(
         gateway.close()
 
 
+def test_acquisition_candidate_never_spawns_before_authority_reservation(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_silent_ffmpeg(tmp_path)))
+    session = tmp_path / "sessions" / "camera-pre-popen-denied"
+    session.mkdir(parents=True)
+    gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID)
+    popen_called = False
+
+    def forbidden_popen(*_: object, **__: object) -> object:
+        nonlocal popen_called
+        popen_called = True
+        raise AssertionError("FFmpeg must not start before authority reservation")
+
+    monkeypatch.setattr(camera_module.subprocess, "Popen", forbidden_popen)
+    try:
+        with pytest.raises(ValueError, match="authority"):
+            gateway.activate_recording_producer(
+                "sensor.camera.right",
+                "192.168.1.20",
+                session,
+                pre_prepare_fence=lambda _reserve: False,
+                commit_fence=lambda commit: commit(),
+            )
+
+        snapshot = gateway.snapshot()
+        assert popen_called is False
+        assert snapshot["active_source_id"] is None
+        assert snapshot["recording"]["active"] is False
+    finally:
+        gateway.close()
+
+
 def test_acquisition_candidate_binds_epoch_before_any_reader_thread_starts(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_silent_ffmpeg(tmp_path)))
     session = tmp_path / "sessions" / "camera-bind-before-reader"
     session.mkdir(parents=True)
     gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID)
+    caplog.set_level("INFO", logger="k1link.device_plugins.xgrids_k1.camera")
     bound_generation: list[int] = []
     reader_start_generation: list[int] = []
+    authority_reserved = False
+
+    def reserve_authority(reserve: Callable[[], bool]) -> bool:
+        nonlocal authority_reserved
+        assert authority_reserved is False
+        assert reserve() is True
+        authority_reserved = True
+        return True
 
     def bind(committed: dict[str, object]) -> None:
         recording = committed["recording"]
@@ -1436,6 +1481,7 @@ def test_acquisition_candidate_binds_epoch_before_any_reader_thread_starts(
 
     def start_threads(producer: object) -> None:
         generation = producer.generation  # type: ignore[attr-defined]
+        assert authority_reserved is True
         assert bound_generation == [generation]
         reader_start_generation.append(generation)
 
@@ -1445,6 +1491,7 @@ def test_acquisition_candidate_binds_epoch_before_any_reader_thread_starts(
             "sensor.camera.right",
             "192.168.1.20",
             session,
+            pre_prepare_fence=reserve_authority,
             commit_fence=lambda commit: commit(),
             committed_before_start=bind,
         )
@@ -1452,6 +1499,16 @@ def test_acquisition_candidate_binds_epoch_before_any_reader_thread_starts(
         assert bound_generation == [activated["recording"]["active_epoch"]]
         assert reader_start_generation == bound_generation
         assert activated["recording"]["media_ready"] is False
+        timing = [
+            record
+            for record in caplog.records
+            if getattr(record, "event_code", None) == "k1_camera_activation_timing"
+        ]
+        assert len(timing) == 1
+        assert timing[0].camera_generation == bound_generation[0]
+        assert timing[0].camera_authority_wait_ms >= 0
+        assert timing[0].camera_ffmpeg_prepare_ms >= 0
+        assert timing[0].camera_post_spawn_commit_ms >= 0
     finally:
         gateway.close()
 

tests/test_xgrids_physical_command_ledger.py

@@ -771,6 +771,71 @@ def test_immediate_archived_reconciliation_successor_fails_closed_on_archive_fau
         )
 
 
+def test_passive_snapshot_reuses_only_unchanged_verified_filesystem_generation(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    ledger = _ledger(tmp_path, monkeypatch)
+    _prepare_start(ledger)
+    real_reload = ledger._reload_locked
+    reload_count = 0
+
+    def counted_reload() -> None:
+        nonlocal reload_count
+        reload_count += 1
+        real_reload()
+
+    monkeypatch.setattr(ledger, "_reload_locked", counted_reload)
+
+    first = ledger.snapshot()
+    second = ledger.snapshot()
+    third = ledger.snapshot()
+
+    assert first == second == third
+    assert reload_count == 1
+
+
+def test_passive_snapshot_observes_atomic_write_from_second_ledger_instance(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    first = _ledger(tmp_path, monkeypatch)
+    _prepare_start(first)
+    assert first.snapshot().record is not None
+    second = PhysicalCommandLedger(tmp_path / "repository", clock=_clock)
+
+    second.mark_dispatching(START_OPERATION)
+
+    observed = first.snapshot()
+    assert observed.record is not None
+    assert observed.record.stage == "dispatching"
+    assert observed.record.revision == 2
+
+
+def test_passive_snapshot_ctime_fence_detects_same_size_archive_tamper(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    ledger = _ledger(tmp_path, monkeypatch)
+    _predecessor, _current, archive_path = _install_resolved_start_rebind_rollover(
+        ledger,
+        monkeypatch,
+    )
+    assert ledger.snapshot().status == "resolved"
+    original_metadata = archive_path.stat()
+    tampered = bytearray(archive_path.read_bytes())
+    tampered[-1] = ord("{") if tampered[-1] != ord("{") else ord("}")
+    archive_path.write_bytes(tampered)
+    os.utime(
+        archive_path,
+        ns=(original_metadata.st_atime_ns, original_metadata.st_mtime_ns),
+    )
+
+    assert archive_path.stat().st_size == original_metadata.st_size
+    assert archive_path.stat().st_mtime_ns == original_metadata.st_mtime_ns
+    assert ledger.snapshot().status == "corrupt"
+
+
 def test_classified_stop_ancestry_rejects_missing_confirmation_cycle_and_conflict(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,