diff --git a/docs/adr/0042-k1-run-ledger-and-regression-telemetry.md b/docs/adr/0042-k1-run-ledger-and-regression-telemetry.md new file mode 100644 index 0000000..6a99243 --- /dev/null +++ b/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. diff --git a/docs/lab/010_K1_MISSION_CORE_INTERNAL_LIVE_BASELINE_20260822.redacted.md b/docs/lab/010_K1_MISSION_CORE_INTERNAL_LIVE_BASELINE_20260822.redacted.md new file mode 100644 index 0000000..258116c --- /dev/null +++ b/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. diff --git a/src/k1link/device_plugins/xgrids_k1/camera.py b/src/k1link/device_plugins/xgrids_k1/camera.py index 09ae30b..224bd89 100644 --- a/src/k1link/device_plugins/xgrids_k1/camera.py +++ b/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 - with self._lock: - self._require_open_locked() - if self._ffmpeg_path is None: - raise RuntimeError("локальный camera adapter FFmpeg не найден") - if self._producer is not None or self._recording_root is not None: - raise RuntimeError("camera acquisition producer уже активен") - if self._source_id is not None and ( - self._source_id != source_id or self._target_host != target - ): - raise RuntimeError("для camera gateway уже выбран другой source") - if self._source_id is None: - self._generation += 1 - self._source_id = source_id - self._target_host = target - self._recording_root = root - self._recording_media_segment_count = 0 - self._archive_summaries = [] - self._expected_source_end_generation = None - self._phase = "selected" - self._error = None - self._revision += 1 + reserved_generation: list[int] = [] + + def reserve() -> bool: + with self._lifecycle_lock: # noqa: SIM117 - lock order is intentional + with self._lock: + self._require_open_locked() + if self._ffmpeg_path is None: + raise RuntimeError("локальный camera adapter FFmpeg не найден") + if self._producer is not None or self._recording_root is not None: + raise RuntimeError("camera acquisition producer уже активен") + if self._source_id is not None and ( + self._source_id != source_id or self._target_host != target + ): + raise RuntimeError("для camera gateway уже выбран другой source") + if self._source_id is None: + self._generation += 1 + self._source_id = source_id + self._target_host = target + self._recording_root = root + self._recording_media_segment_count = 0 + self._archive_summaries = [] + 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: diff --git a/src/k1link/device_plugins/xgrids_k1/facade.py b/src/k1link/device_plugins/xgrids_k1/facade.py index 2399bb7..7fb9cda 100644 --- a/src/k1link/device_plugins/xgrids_k1/facade.py +++ b/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() - physical = self._physical_command_coordinator.snapshot() - control = self._application_control_session.snapshot() - ble_runtime = ble_runtime_snapshot() - semantic = self._semantic_topology_public_snapshot() - identity = self._device_identity_pin_public_snapshot() - idempotency = self._network_provisioning_idempotency_public_snapshot() - network_ledger = self._network_mutation_ledger.snapshot() + runtime = ( + runtime_snapshot if runtime_snapshot is not None else self.runtime.snapshot() + ) + physical = ( + physical_command_snapshot + if physical_command_snapshot is not None + else self._physical_command_coordinator.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() - control = self._application_control_session.snapshot() - native = ble_runtime_snapshot() + runtime = ( + runtime_snapshot if runtime_snapshot is not None else self.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() - control = self._application_control_session.snapshot() - native = ble_runtime_snapshot() - supervisor = self._connection_supervisor.snapshot() - network_ledger = self._network_mutation_ledger.snapshot() - idempotency = self._network_provisioning_idempotency_public_snapshot() - semantic_topology = self._semantic_topology_public_snapshot() - identity_pins = self._device_identity_pin_public_snapshot() + runtime = ( + runtime_snapshot if runtime_snapshot is not None else self.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() + ) + 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: ( - self._reserve_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, - 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, - ), + pre_retry_fence=reserve_activation, + commit_fence=commit_activation, 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( - 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, - ), + pre_prepare_fence=reserve_activation, + commit_fence=commit_activation, 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) diff --git a/src/k1link/device_plugins/xgrids_k1/physical_command_ledger.py b/src/k1link/device_plugins/xgrids_k1/physical_command_ledger.py index e8554ac..0ff6845 100644 --- a/src/k1link/device_plugins/xgrids_k1/physical_command_ledger.py +++ b/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, *, diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_session.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_session.py index 69c7b92..64b0fde 100644 --- a/src/k1link/device_plugins/xgrids_k1/protocol/application_session.py +++ b/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() - phase = self._phase - return { - "mode": "interactive-canonical", - "inspection_only": self._inspection_only, - "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": ( - self._physical_command_coordinator.snapshot() - if self._physical_command_coordinator is not None - else None - ), - } + physical_command = ( + self._physical_command_coordinator.snapshot() + if self._physical_command_coordinator is not None + else None + ) + return self._snapshot_locked(physical_command=physical_command) + + def snapshot_control_only(self) -> dict[str, object]: + """Return process-local control facts without reloading the durable ledger.""" + + with self._lock: + return self._snapshot_locked(physical_command=None) + + def snapshot_with_physical_command( + self, + physical_command: Mapping[str, object] | None, + ) -> dict[str, object]: + """Join one already-verified physical snapshot to current control facts.""" + + with self._lock: + return self._snapshot_locked(physical_command=physical_command) + + def _snapshot_locked( + self, + *, + physical_command: Mapping[str, object] | None, + ) -> dict[str, object]: + transport_snapshot = self._live_transport_snapshot_locked() + phase = self._phase + return { + "mode": "interactive-canonical", + "inspection_only": self._inspection_only, + "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() diff --git a/src/k1link/web/runtime_diagnostics.py b/src/k1link/web/runtime_diagnostics.py index 2640066..f02b9d2 100644 --- a/src/k1link/web/runtime_diagnostics.py +++ b/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", diff --git a/tests/test_viewer_diagnostics_api.py b/tests/test_viewer_diagnostics_api.py index c77cd2a..a73fd07 100644 --- a/tests/test_viewer_diagnostics_api.py +++ b/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 diff --git a/tests/test_xgrids_acquisition_lifecycle.py b/tests/test_xgrids_acquisition_lifecycle.py index 93cda05..644fb9b 100644 --- a/tests/test_xgrids_acquisition_lifecycle.py +++ b/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): diff --git a/tests/test_xgrids_application_session.py b/tests/test_xgrids_application_session.py index 8f296a4..64a9b1e 100644 --- a/tests/test_xgrids_application_session.py +++ b/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: diff --git a/tests/test_xgrids_camera_gateway.py b/tests/test_xgrids_camera_gateway.py index 53ffa13..7ed0055 100644 --- a/tests/test_xgrids_camera_gateway.py +++ b/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() diff --git a/tests/test_xgrids_physical_command_ledger.py b/tests/test_xgrids_physical_command_ledger.py index 4aec5b6..dbff63a 100644 --- a/tests/test_xgrids_physical_command_ledger.py +++ b/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,