diff --git a/README.md b/README.md index 4478685..053cc7e 100644 --- a/README.md +++ b/README.md @@ -143,20 +143,21 @@ The first passive replay derivative is now implemented as `missioncore.k1-local-surface/v1`. It binds the immutable `RAVNOVES00` pack, estimates a rolling local surface from map points plus compatible pose, and publishes height, slope, roughness, confidence and conservative observed -surface/occupied/unknown evidence. All `526/526` available samples produced a -diagnostic result; no free-space, command, navigation or safety authority is -inferred. Leave-current-frame-out qualification produced `525` next-frame -samples with `0.038543 m` p50 median residual and `12` strict temporal jumps; -the p95 residual tail remains too large for planner use. Read-only replay -triage narrows this to `37` attention frames in `21` episodes and four -high-priority frames. The selected scene, clickable complete-recording timeline -and source-frame review queue are visible in **Парк → Диагностика LiDAR**. -The selected frame can now switch to a prior-plane residual view: it preserves -the previous-TTL plane and overlays the current lower-cell evidence as -in-band, above-plane and below-plane observations. On source frame `1254`, -`36/98` evaluated cells are above the `0.16 m` band and none are below it, -localizing the heavy tail without naming an object or granting safety -authority. +surface/occupied/unknown evidence. The complete pack contains `4,489` timeline +frames, `3,928` available/valid LiDAR frames and `9,207,270` points; there were +no stale-pose, insufficient-surface or fit-failed frames. Prior-only +next-frame qualification produced `3,927` samples with `0.04065 m` p50 median +residual and `205` strict temporal jumps. The p95 residual tail remains too +large for planner use. The current complete derivative is the first item in +**Парк → Диагностика LiDAR**. + +The earlier content-bound `526`-frame review slice remains available as +historical laboratory evidence. Its deterministic triage contains `37` +attention frames in `21` episodes and four high-priority frames. The selected +frame can switch to the prior-plane residual view; source frame `1254` has +`36/98` evaluated cells above the `0.16 m` band and none below it. This +localizes a positive residual structure without naming an object or granting +safety authority. The same estimator now runs behind `missioncore.k1-local-surface-shadow-runtime/v1`, a capacity-two @@ -165,20 +166,22 @@ latest-wins queue with a bounded diagnostic result ring. LAB E27 passed a were consumed, maximum queue depth was `1/2`, no frame was replaced, p95 processing latency was `14.777 ms`, and state, point classes, step candidates and scalar results matched the immutable replay derivative exactly. This is a -recorded-source-paced execution gate only; physical K1 worker binding, +recorded-source-paced execution gate only; physical K1 acquisition, free-space, commands, navigation and safety authority remain unavailable. -The physical shadow seam is now implemented but not yet physically qualified. -The authenticated external worker can bind decoded map-frame `lio_pcl` and -`lio_pose` in either arrival order through -`missioncore.k1-local-surface-pose-binder/v1`, then feed the accepted estimator +The authenticated external-worker seam is now implemented and replay-qualified. +It binds decoded map-frame `lio_pcl` and `lio_pose` in either arrival order +through +`missioncore.k1-local-surface-pose-binder/v1`, then feeds the accepted estimator without blocking camera cadence. Point input and estimator work are -capacity-two latest-wins queues; pose history and result history are bounded. -The pinned `e28_physical_k1_local_surface_profile.json` rejects the run unless -at least 100 point frames bind and exact accounting, pose misses, queue drops, -p95 result age, profile identity and shadow-only authority all pass. No -physical acceptance result is claimed until that profile is executed against -an attached K1 on the NVIDIA worker. +bounded separately: the authenticated ingress has LiDAR depth eight, while the +freshness-first estimator remains capacity two. LAB E28 replayed the complete +recording at 1× through the actual NVIDIA worker: all `4,570` LiDAR and `4,598` +pose events arrived with zero gaps, all point frames bound, and `4,206` +results were persisted at `9.1945 Hz`. Explicitly superseded work was +`364/4,570` (`7.965%`), p95 result age was `89.530 ms`, and all acceptance +checks passed. Physical K1 acquisition remains a separate commissioning test; +it is not required for replay algorithm development. The complete RELLIS-3D v1.1 release is now admitted there and its full `2,413`-frame validation split is available in **Полигон → Датасеты**. The diff --git a/docs/13_LIDAR_WORKER_PRODUCT_AND_ROADMAP.md b/docs/13_LIDAR_WORKER_PRODUCT_AND_ROADMAP.md index 001d74e..63fa225 100644 --- a/docs/13_LIDAR_WORKER_PRODUCT_AND_ROADMAP.md +++ b/docs/13_LIDAR_WORKER_PRODUCT_AND_ROADMAP.md @@ -5,7 +5,7 @@ Status: accepted architecture plan; L0/L1 implemented; L2 diagnostic A/B complete; full GOOSE and RELLIS qualification complete; L2.6d K1 replay local-surface temporal qualification, operator triage and prior-plane residual explainability implemented; L2.6e recorded-source-paced bounded shadow -qualified; physical K1 shadow next +qualified; E28 complete worker replay accepted; dynamic-observation layer next Scope: passively received real-time K1 point/pose evidence, immutable replay and future live shadow processing Explicitly out of scope: K1 firmware modification, a new onboard exporter, new @@ -423,8 +423,10 @@ Dataset expansion is no longer the next gate. - [x] Replay the same profile through a bounded latest-wins shadow queue at the recorded 1× source rate; no K1 command, navigation or safety authority is added. -- [ ] Bind that runtime to physical live K1 point/pose evidence through the - existing authenticated external-worker seam. +- [x] Bind the complete raw replay to the normalizer, point/pose binder and + bounded estimator through the authenticated external-worker seam. +- [ ] Commission the same seam against physical live K1 evidence only when + hardware/network behavior becomes the test subject. The implemented `missioncore.k1-local-surface/v1` derivative is reproducible through `experiments/perception/run_k1_local_surface.py` and is exposed @@ -437,36 +439,40 @@ local-discontinuity candidates. The candidates are not semantic curb recognition or ground truth. The React contract is provider-neutral; K1 remains a bound backend source rather than UI implementation knowledge. -The complete RAVNOVES00 run covered all `526/526` available LiDAR samples. -There were zero stale-pose, insufficient-surface or failed-fit frames. Observed -diagnostic distributions are: derived sensor-to-surface height `1.295 m` p50, -roughness `0.054 m` p95, slope `3.172°` p95 and pose-binding age `19.113 ms` -p95. These values describe this recording only; they are not calibration, -ground truth or a navigation gate. +The complete RAVNOVES00 pack contains `4,489` timeline frames and `9,207,270` +points. `3,928` frames carry available LiDAR evidence and all `3,928` produced +a valid result; `561` timeline frames are explicitly unavailable. There were +zero stale-pose, insufficient-surface or failed-fit frames. Observed diagnostic +distributions are: derived sensor-to-surface height `1.2816 m` p50, roughness +`0.06736 m` p95, slope `4.1895°` p95 and pose-binding age `21.8545 ms` p95. +These values describe this recording only; they are not calibration, ground +truth or a navigation gate. The L2.6b qualification scores each available frame against a local plane built only from the preceding TTL window; the frame being scored is excluded from -the prediction input. It produced `525` independent next-frame samples. The -distribution of per-frame median absolute residual has `0.038543 m` p50 and -`0.050398 m` p95. Prediction inlier fraction has `0.96354` p50 and `0.95133` -mean. The temporal comparison detected `12/525` strict-threshold jumps; height -delta is `0.022887 m` p95, slope delta `0.09572°` p95 and roughness delta -`0.003317 m` p95. +the prediction input. It produced `3,927` independent next-frame samples. The +distribution of per-frame median absolute residual has `0.04065 m` p50 and +`0.07304 m` p95. Prediction inlier fraction has `0.94643` p50 and `0.91937` +mean. The temporal comparison detected `205/3,927` strict-threshold jumps; +height delta is `0.02891 m` p95, slope delta `0.12410°` p95 and roughness +delta `0.00520 m` p95. The central surface is therefore repeatable at roughly four-centimetre median error on this replay. The tail is not yet planner evidence: the distribution -of per-frame p95 residual has `0.41496 m` p95 and reaches `1.083 m`. Step/curb -candidates occur in every available frame (`68.5` cells and `83` points p50), +of per-frame p95 residual has `0.66193 m` p95 and reaches `1.735 m`. Step/curb +candidates occur in every available frame (`75` cells and `99` points p50), which proves that the review layer is active but also that it is broad and still requires independent review. Free space remains unavailable; none of these metrics grants navigation, command or safety authority. -The `missioncore.k1-local-surface-review/v1` triage contract selects a frame +The earlier content-bound `526`-frame review generation is retained for +operator evidence and regression history. Its +`missioncore.k1-local-surface-review/v1` triage contract selects a frame when its prediction p95 residual is at least `0.45 m`, prediction inlier fraction falls below `85%`, or one of the content-bound temporal thresholds is crossed. It groups observations separated by no more than two frames into one episode and ranks threshold exceedance without changing the source artifact. -RAVNOVES00 produced `37` review frames in `21` episodes: `18` prediction-tail +That reviewed slice produced `37` review frames in `21` episodes: `18` prediction-tail frames, `14` inlier drops and `12` height transitions. Four frames have a normalized attention score of at least `2.0`. @@ -517,7 +523,7 @@ immutable replay derivative: zero state, point-class or step-candidate mismatches and zero scalar delta. The accepted result is `k1-local-surface-shadow-04f14d8c580f74cbd5b0a452867563ebc6b3ef93d872129e4918680932253ab7`. -The authenticated external worker now contains the next bounded seam. +The authenticated external worker now contains the accepted bounded seam. `missioncore.k1-local-surface-pose-binder/v1` accepts decoded map-frame `lio_pcl` and `lio_pose` in either arrival order, binds by bounded worker host-arrival time, and publishes only matched pairs to the capacity-two @@ -527,18 +533,28 @@ counts every point frame; the final report records pose-binding misses, binder/runtime overflow, exact queue accounting, processing failures and p95 result age. -The pinned physical profile is -`experiments/perception/worker/e28_physical_k1_local_surface_profile.json`. -It requires at least `100` bound frames, at most `5%` pose misses, `1%` binder -point replacement, `1%` runtime replacement and `80 ms` p95 worker-local -result age. The existing D-only PowerShell worker launcher accepts this -physical profile and verifies that the minimal hash-addressed package contains -the local-surface geometry, binder/runtime and ground primitives. +The pinned profile is +`experiments/perception/worker/e28_worker_replay_local_surface_profile.json`. +It separates a lossless depth-eight authenticated LiDAR ingress from the +freshness-first capacity-two estimator. The acceptance gate requires at least +`9 Hz`, no more than `100 ms` p95 result age, at most `8%` explicitly +superseded estimator inputs, at most `5%` pose misses, at most `1%` binder +replacement, exact queue accounting and shadow-only authority. -This is implementation readiness, not a physical result. The next gate is to -build and stage the new content-addressed worker package on D, execute the -profile against an attached K1 and retain the generated report. Commands, -free-space, navigation and safety authority remain disabled. +LAB E28 executed the complete recording on `mission-gpu` through the normal +authenticated wire. All `4,570` LiDAR and `4,598` pose events arrived with +zero sequence gaps; all `4,570` point frames bound with no miss or binder +replacement. The worker persisted `4,206` results, explicitly superseded +`364/4,570` inputs (`7.965%`), delivered `9.1945 Hz`, and measured +`89.530 ms` full p95 result age. Processing p95 was `45.470 ms`; failures were +zero. The accepted result is +`e28-local-surface-wire-e1b0902d502267aff5f0bf78365b24967f98c83d7c1e6bb0f9dc2934c0684ceb`. + +This is a worker replay result, not a physical acquisition result. The next +gate is a separate dynamic-observation layer plus explicit staleness/deadline +health. Physical K1 execution is deferred until hardware-clock, field-network +or acquisition behavior is the test subject. Commands, free-space, navigation +and safety authority remain disabled. Exit: one immutable K1 session yields both a persistent reconstruction and a bounded local world state without hard-coded terrain height or scanner-side @@ -563,10 +579,12 @@ independent gate without increasing unsafe false-free or false-dynamic output. - [x] Add a provider-neutral bounded LiDAR local-surface queue independent of camera cadence and qualify it at recorded 1× source pace. -- [x] Bind authenticated decoded physical `lio_pcl + lio_pose` to that queue - with bounded buffers, exact accounting and a pinned physical acceptance +- [x] Bind authenticated decoded replay `lio_pcl + lio_pose` to that queue + with bounded buffers, exact accounting and a pinned worker acceptance profile. -- [ ] Run the accepted K1 local-surface/local-map profile on the NVIDIA worker. +- [x] Run the complete accepted K1 local-surface profile on the NVIDIA worker. +- [ ] Commission physical K1 acquisition separately; do not block replay + development on device presence. - [ ] Fuse K1 geometric evidence with E26 camera evidence as independent sources; a LiDAR-native detector remains optional. - [ ] Publish `agree`, `single-source`, `conflict` and `unknown`; unknown remains @@ -625,15 +643,18 @@ natural-ground recall from `51.76%` to `76.22%` and stays below `23.41 ms` p95 across 961 frames. Full RELLIS qualification covers `2,413` validation frames and rejects Patchwork++ because the small Ground-IoU gain came with unacceptable obstacle loss. Public-dataset ground qualification is therefore complete -enough for the current decision. The first K1 local-surface replay slice now -covers all available `RAVNOVES00` samples and is visible in the operator -interface. Leave-current-frame-out temporal qualification now covers `525` -samples: the median surface error is stable, while the p95 tail remains too -large for a free-space claim. Deterministic triage has reduced the first manual -inspection set to four high-priority frames in two episodes. Prior-plane -residual explainability now shows that episode `09` is a localized positive -structure rather than symmetric plane drift. The highest-value immediate work -is therefore the physical K1 run through the now-implemented bounded -live-shadow seam, followed by a separate dynamic-observation layer. Nvblox, +enough for the current decision. The complete K1 local-surface derivative now +covers `4,489` timeline frames, including `3,928` available LiDAR frames, and +is visible in the operator interface. Leave-current-frame-out temporal +qualification covers `3,927` samples: the median surface error is stable, +while the p95 tail remains too large for a free-space claim. The retained +review generation reduces its first manual inspection set to four +high-priority frames in two episodes. Prior-plane residual explainability +shows that episode `09` is a localized positive structure rather than +symmetric plane drift. The complete worker replay proves the bounded +live-shadow seam at `9.1945 Hz` and `89.530 ms` p95 result age without a +connected scanner. The highest-value immediate work is therefore the separate +dynamic-observation and staleness-health layer. Physical K1 commissioning +follows only when hardware behavior is in scope. Nvblox, raw-scan detectors and alternative SLAM remain optional later gates because the current report contract does not carry their required ray/timing semantics. diff --git a/docs/14_LIDAR_DATASET_GATEWAY.md b/docs/14_LIDAR_DATASET_GATEWAY.md index 8d41f2f..21c2277 100644 --- a/docs/14_LIDAR_DATASET_GATEWAY.md +++ b/docs/14_LIDAR_DATASET_GATEWAY.md @@ -49,11 +49,13 @@ Implemented now: - a content-addressed `missioncore.k1-local-surface/v1` replay derivative over immutable `RAVNOVES00`, with dynamic height/slope/roughness/confidence, explicit pose-binding age and conservative observed occupied/unknown policy; -- leave-current-frame-out next-frame qualification over `525` samples, - temporal jump evidence and unverified local-discontinuity candidates; +- a complete `4,489`-frame timeline derivative with `3,928` valid LiDAR + results, `9,207,270` points and leave-current-frame-out qualification over + `3,927` samples; - deterministic `missioncore.k1-local-surface-review/v1` replay triage: - `37` attention frames grouped into `21` episodes, with four - high-priority frames and direct source-frame navigation; + the retained `526`-frame review generation has `37` attention frames grouped + into `21` episodes, with four high-priority frames and direct source-frame + navigation; - point/cell-aligned prediction evidence in `missioncore.k1-local-surface-frame/v2`: the prior-only plane, current lower-cell coordinates, signed residual and derived inlier mask are @@ -61,6 +63,9 @@ Implemented now: - a provider-neutral read-only local-surface view in **Парк → Диагностика LiDAR**, synchronized to the five existing RAVNOVES00 scene selectors and a clickable complete-recording timeline; +- a complete recorded-host-paced 1× worker replay through authenticated + ingress, K1 normalization, point/pose binding and a bounded latest-wins + surface estimator: `4,206` results at `9.1945 Hz` and `89.530 ms` p95 age; - worker storage admission for `D:\NDC_MISSIONCORE\datasets` and `/mnt/d/NDC_MISSIONCORE/datasets`; - a dedicated, honest dataset catalog in **Полигон → Датасеты**; @@ -93,11 +98,11 @@ Not implemented: - no RELLIS ROS bag admission, continuous synchronized playback or production promotion; - no ray-cleared free-space or planner-authoritative rolling occupancy map; -- the accepted K1 local-surface profile now passes a 15-second - recorded-source-paced bounded shadow gate; authenticated physical - `lio_pcl + lio_pose` binding and its pinned acceptance profile are - implemented, but no attached-K1 run has qualified them yet. The residual - overlay remains non-authoritative. +- the accepted K1 local-surface profile now passes both the historical + 15-second equivalence gate and the complete external-worker replay gate. + Physical K1 acquisition, hardware-clock and field-network commissioning + remain separate and intentionally unqualified. The residual overlay remains + non-authoritative. ## Product surface boundary diff --git a/experiments/perception/LAB_E28_REPORT_2026-07-26.md b/experiments/perception/LAB_E28_REPORT_2026-07-26.md new file mode 100644 index 0000000..ba8b574 --- /dev/null +++ b/experiments/perception/LAB_E28_REPORT_2026-07-26.md @@ -0,0 +1,171 @@ +# LAB E28 — complete K1 replay through the external worker + +Date: 2026-07-26 + +State: **accepted for replay shadow diagnostics** + +## Scope + +E28 qualifies the existing K1 local-surface estimator on the external +`mission-gpu` worker without reconnecting the physical scanner. The complete +immutable `RAVNOVES00` recording is reproduced at recorded 1× host-arrival +pace through the normal authenticated binary WebSocket, K1 normalizer, +point/pose binder and bounded latest-wins estimator. + +This is not a physical acquisition, free-space, navigation or safety +acceptance. No K1 firmware change, device command or scanner write was used. + +## Source identity + +- session: `20260720T065719Z_viewer_live`; +- selection: complete recording, no look-ahead; +- combined source span: `469.908994 s`; +- camera frames: `4,489`; +- raw K1 LiDAR events: `4,570`; +- raw K1 pose events: `4,598`; +- MQTT capture SHA-256: + `70da0edad5cbf0e89b6e2355c0aac294f8c36a67249cd33b8918353fd88c83af`; +- camera index SHA-256: + `e029815a60ad9fbfedb6169142c7449df2b119a51d1ce001f08806e04eb0be14`. + +The source process admitted every selected event. Its LiDAR ingress reached a +maximum depth of `4/8` with zero replacement. Release lag was `5.051 ms` p95 +and source wall time was `470.169 s`, or `1.00055×` ideal. + +## Full passive derivative + +Before the wire qualification, the complete content-addressed E10 pack was +processed offline with the same pinned surface profile: + +- model: + `k1-local-surface-23762244c8bdb97de26fb721ac957d7a00bc9a63571ac4cfa4be19c4effc7d55`; +- timeline frames: `4,489`; +- available/valid LiDAR frames: `3,928`; +- unavailable frames: `561`; +- admitted points: `9,207,270`; +- stale-pose, insufficient-surface and fit-failed frames: `0`; +- sensor height: `1.2816 m` p50; +- pose-binding age: `21.8545 ms` p95; +- slope: `4.1895°` p95; +- roughness: `0.06736 m` p95; +- prior-only next-frame median residual: `0.04065 m` p50 and + `0.07304 m` p95; +- distribution of per-frame p95 residual: `0.66193 m` p95; +- strict temporal jumps: `205/3,927`. + +The immutable NPZ SHA-256 is +`f57eb2485b6cef47f2a97a2d9ff1aa9fd9265fe1eb69cd5852d12f39e13b8bc6`. +The current report SHA-256 is +`e342ecd50a1302bf1fd2250f2dfc8d4ba2909e311ced1d2651696dd200bbf0ca`. +This derivative is the first item returned to +**Парк → Диагностика LiDAR**. + +## Measured queue decision + +The recording contains real host-arrival bursts: among `4,569` LiDAR +inter-arrival intervals, `457` are below `25 ms` and `2,245` are below +`80 ms`. Requiring a sequential stateful estimator to process every burst +increment increased result age without increasing the product update rate. + +Rejected attempts were retained: + +| Execution | Queue | Result | Finding | +| --- | ---: | --- | --- | +| Mac, complete 10× stress | 2 | rejected | `851` results; `3,711` runtime replacements; `169.94 ms` bounded p95 age | +| Mac, complete 1× | 2 | rejected | old ingress depth lost `6` LiDAR events; `296/4,564` runtime replacements | +| Mac, complete 1× | 8 | rejected | transport became lossless and runtime replacement fell to `38/4,570`, but full p95 age rose to `128.00 ms` | +| worker, 60 s 1× | 8 | rejected diagnostic slice | `581/585` results; p95 age `163.06 ms` | +| worker, 60 s 1× | 2 | rejected under the old gate | `557/585` results, `9.28 Hz`; p95 age `87.73 ms` | + +The accepted policy therefore separates two bounds: + +- ingress depth `8` preserves the complete authenticated transport; +- estimator depth `2` keeps a fresh local-world result and explicitly counts + superseded work. + +Acceptance is expressed in product quantities: at least `9 Hz`, no more than +`100 ms` p95 result age and no more than `8%` superseded estimator inputs. +Transport and binder accounting remain exact; no threshold hides unknown +events. + +## Accepted worker run + +Result: +`e28-local-surface-wire-e1b0902d502267aff5f0bf78365b24967f98c83d7c1e6bb0f9dc2934c0684ceb`. + +| Measurement | Result | +| --- | ---: | +| Camera / LiDAR / pose events received | `4,489 / 4,570 / 4,598` | +| Ingress sequence gaps | `0` | +| Point frames bound / missed / binder-dropped | `4,570 / 0 / 0` | +| Estimator results persisted | `4,206` | +| Explicitly superseded estimator inputs | `364` (`7.965%`) | +| Effective local-surface rate | `9.1945 Hz` | +| Sensor decode p50 / p95 | `1.672 / 16.776 ms` | +| Surface processing p50 / p95 | `23.689 / 45.470 ms` | +| Full result age p50 / p95 | `37.849 / 89.530 ms` | +| Runtime or evidence-writer failures | `0 / 0` | +| Evidence queue maximum depth | `1/32` | +| Worker wall time | `468.042 s` | + +All 21 acceptance checks passed. Every processed result was written to +`local-surface-frames.jsonl` with hashes of its point-class, point-height and +step-candidate arrays. The report copied back from D has SHA-256 +`8116d88f903a8c743a3b1c59984a7b4688aaa5fb67e6d2f849a03d3479ce1b97`. + +The maximum result-age tail reached `1,275.662 ms`. E28 does not hide that +tail: p95 passes the shadow-diagnostic gate, while any future planner-facing +stage must add deadline-miss/staleness health and a stricter independent +acceptance. + +## Worker and storage boundary + +- worker: `DESKTOP-OPJ8J04` / `mission-gpu`; +- task-controlled root: `D:\NDC_MISSIONCORE`; +- result: + `D:\NDC_MISSIONCORE\runtime\derived\e28-local-surface\worker-full-1x-q2-gate-v2`; +- worker package: + `e15-worker-package-77e6537cf347c7e03f286a6235c33fc3072403b3ddf64f24749c9ed0e066be73`; +- package manifest SHA-256: + `a530fea8c703caa3ddefc49580a5311206e833b6c8b9bebf48d3e5242babeaf7`; +- runner SHA-256: + `be055a2c75686844b6d9de9d787ac922d2094643d39d9db29502321805fff5d9`; +- profile SHA-256: + `86814d28544138f905f73d14ef3ad58e4815a0dad60022945076eb88c68f0d30`; +- container image: + `sha256:58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794`. + +Package, runner, profile, secret repository and result remained on D. Package, +runner, profile, media runtime and token repository were mounted read-only; +only the E28 result mount was writable. The container was read-only, +capability-free and `no-new-privileges`. + +The source stayed on Mac loopback. A temporary reverse SSH tunnel exposed it +only as worker loopback `127.0.0.1:18031`; no LAN listener was created. The +bearer token was read from a private file and never passed on the command line. +The tunnel and one-shot container were gone after the run. + +## Decision + +The complete saved K1 evidence is sufficient for current algorithm and worker +qualification. A physically attached K1 is not required for each software +iteration. + +Accepted now: + +- complete passive K1 replay on the actual external worker; +- lossless authenticated transport and point/pose binding; +- bounded freshness-first local-surface output at greater than `9 Hz`; +- current operator review of the full offline derivative. + +Still unavailable: + +- raw MID-360 scan semantics, ray-cleared free space and traversability; +- dynamic-object classification; +- hardware-clock, field-network and physical acquisition commissioning; +- command, navigation or safety authority. + +The next implementation gate is a separate dynamic-observation layer over the +accepted local surface, followed by explicit staleness/deadline health. +Physical K1 execution remains a later commissioning test for transport and +hardware behavior, not a prerequisite for this development loop. diff --git a/experiments/perception/run_e12_shadow_transport_replay.py b/experiments/perception/run_e12_shadow_transport_replay.py index d89e378..3e40b8d 100644 --- a/experiments/perception/run_e12_shadow_transport_replay.py +++ b/experiments/perception/run_e12_shadow_transport_replay.py @@ -86,7 +86,10 @@ def _percentiles(values: list[float]) -> dict[str, float | int]: } -def _camera_events(epoch_root: Path, duration_seconds: float) -> Iterator[ReplayEvent]: +def _camera_events( + epoch_root: Path, + duration_seconds: float | None, +) -> Iterator[ReplayEvent]: index_path = epoch_root / "index.jsonl" if not index_path.is_file(): raise RuntimeError("camera index is missing") @@ -96,7 +99,11 @@ def _camera_events(epoch_root: Path, duration_seconds: float) -> Iterator[Replay raise RuntimeError("camera replay interval is empty") first_record = json.loads(first_line) start_epoch_ns = int(first_record["host_epoch_ns"]) - end_epoch_ns = start_epoch_ns + int(duration_seconds * 1_000_000_000) + end_epoch_ns = ( + None + if duration_seconds is None + else start_epoch_ns + int(duration_seconds * 1_000_000_000) + ) init_payload = (epoch_root / "init.mp4").read_bytes() yield ReplayEvent( modality="camera-init", @@ -109,7 +116,7 @@ def _camera_events(epoch_root: Path, duration_seconds: float) -> Iterator[Replay for line in itertools.chain((first_line,), stream): record = json.loads(line) epoch_ns = int(record["host_epoch_ns"]) - if epoch_ns > end_epoch_ns: + if end_epoch_ns is not None and epoch_ns > end_epoch_ns: break payload_path = epoch_root / str(record["path"]) payload = payload_path.read_bytes() @@ -130,14 +137,21 @@ def _camera_events(epoch_root: Path, duration_seconds: float) -> Iterator[Replay def _mqtt_events( capture_path: Path, *, - start_epoch_ns: int, - duration_seconds: float, + start_epoch_ns: int | None, + duration_seconds: float | None, ) -> Iterator[ReplayEvent]: - end_epoch_ns = start_epoch_ns + int(duration_seconds * 1_000_000_000) + end_epoch_ns = ( + None + if start_epoch_ns is None or duration_seconds is None + else start_epoch_ns + int(duration_seconds * 1_000_000_000) + ) for message in iter_replay_messages(capture_path): - if message.received_at_epoch_ns < start_epoch_ns: + if ( + start_epoch_ns is not None + and message.received_at_epoch_ns < start_epoch_ns + ): continue - if message.received_at_epoch_ns > end_epoch_ns: + if end_epoch_ns is not None and message.received_at_epoch_ns > end_epoch_ns: break if message.topic.endswith("/lio_pcl") or message.topic == "RealtimePointcloud": modality: Literal["lidar", "pose"] = "lidar" @@ -169,17 +183,22 @@ def _wait_for(predicate: object, *, timeout_seconds: float, label: str) -> None: def run(args: argparse.Namespace) -> dict[str, object]: experiment_id = str(getattr(args, "experiment_id", _EXPERIMENT_ID)) speed = float(getattr(args, "speed", 1.0)) + duration_seconds = float(args.duration_seconds) if ( - experiment_id not in {"e12", "e21", "e23"} + experiment_id not in {"e12", "e21", "e23", "e28"} or not math.isfinite(speed) or not 0.1 <= speed <= 10 + or not math.isfinite(duration_seconds) + or duration_seconds < 0 ): raise RuntimeError("replay experiment identity or speed is invalid") + complete_recording = duration_seconds == 0 + selected_duration = None if complete_recording else duration_seconds repository_root = Path(args.repository_root).expanduser().resolve() session_root = Path(args.session_root).expanduser().resolve() epoch_root = session_root / "media" / "sensor.camera.right" / "epoch-1" capture_path = session_root / "captures" / "mqtt_live" / "mqtt.raw.k1mqtt" - camera = iter(_camera_events(epoch_root, args.duration_seconds)) + camera = iter(_camera_events(epoch_root, selected_duration)) first_camera = next(camera, None) first_frame = next(camera, None) if first_camera is None or first_frame is None or first_frame.modality != "camera-frame": @@ -188,8 +207,8 @@ def run(args: argparse.Namespace) -> dict[str, object]: mqtt = iter( _mqtt_events( capture_path, - start_epoch_ns=first_frame.epoch_ns, - duration_seconds=args.duration_seconds, + start_epoch_ns=(None if complete_recording else first_frame.epoch_ns), + duration_seconds=selected_duration, ) ) first_mqtt = next(mqtt, None) @@ -300,6 +319,7 @@ def run(args: argparse.Namespace) -> dict[str, object]: "e12": REPORT_SCHEMA, "e21": "missioncore.e21-replay-source-report/v1", "e23": "missioncore.e23-replay-source-report/v1", + "e28": "missioncore.e28-worker-replay-source-report/v1", }[experiment_id] peak_rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss peak_rss_mib = peak_rss / (1024 * 1024 if sys.platform == "darwin" else 1024) @@ -312,7 +332,14 @@ def run(args: argparse.Namespace) -> dict[str, object]: "session": session_root.name, "camera_source": "sensor.camera.right", "camera_epoch": 1, - "duration_seconds": args.duration_seconds, + "selection": ( + "complete-recording" + if complete_recording + else "bounded-duration" + ), + "duration_seconds": ( + None if complete_recording else duration_seconds + ), "selected_span_seconds": source_span_seconds, "speed": speed, "pacing_clock": "recorded-host-arrival", @@ -379,7 +406,11 @@ def _arguments() -> argparse.Namespace: parser.add_argument("--port", type=int, default=8012) parser.add_argument("--duration-seconds", type=float, default=15.0) parser.add_argument("--speed", type=float, default=1.0) - parser.add_argument("--experiment-id", choices=("e12", "e21", "e23"), default="e12") + parser.add_argument( + "--experiment-id", + choices=("e12", "e21", "e23", "e28"), + default="e12", + ) parser.add_argument("--consumer-timeout-seconds", type=float, default=30.0) parser.add_argument("--completion-timeout-seconds", type=float, default=30.0) return parser.parse_args() diff --git a/experiments/perception/worker/Invoke-E15ShadowInference.ps1 b/experiments/perception/worker/Invoke-E15ShadowInference.ps1 index fedd94f..ca5278d 100644 --- a/experiments/perception/worker/Invoke-E15ShadowInference.ps1 +++ b/experiments/perception/worker/Invoke-E15ShadowInference.ps1 @@ -136,12 +136,16 @@ if ( $live = Get-Content -LiteralPath $liveProfile -Raw | ConvertFrom-Json if ( $live.schema_version -ne "missioncore.e15-shadow-inference-profile/v1" -or - $live.mode -notin @("replay-shadow-gate", "physical-shadow-gate") -or + $live.mode -notin @( + "replay-shadow-gate", + "worker-replay-gate", + "physical-shadow-gate" + ) -or [bool]$live.authority.commands_enabled -or [bool]$live.authority.navigation_or_safety_accepted -or $live.transport.pyav_version -ne "18.0.0" ) { throw "LAB E15/E28 shadow authority contract changed" } -if ($live.mode -eq "physical-shadow-gate") { +if ($live.mode -in @("worker-replay-gate", "physical-shadow-gate")) { foreach ($relative in @( "k1link\compute\lidar_local_surface_geometry.py", "k1link\compute\lidar_local_surface_shadow.py", @@ -156,6 +160,18 @@ if ($live.mode -eq "physical-shadow-gate") { $live.local_surface.profile_id -ne "k1-vendor-map-dynamic-local-surface/v1" ) { throw "LAB E28 local-surface profile contract changed" } } +if ( + $live.mode -eq "worker-replay-gate" -and ( + $live.replay_source.session_id -ne "20260720T065719Z_viewer_live" -or + $live.replay_source.selection -ne "complete-recording" -or + [double]$live.replay_source.speed -ne 1.0 -or + [bool]$live.replay_source.look_ahead -or + [int]$live.local_surface.point_queue_capacity -ne 2 -or + [double]$live.local_surface.acceptance.minimum_effective_fps -ne 9.0 -or + [double]$live.local_surface.acceptance.maximum_runtime_drop_fraction -ne 0.08 -or + [double]$live.local_surface.acceptance.maximum_p95_result_age_ms -ne 100.0 + ) +) { throw "LAB E28 complete-recording replay contract changed" } if ($stabilityProfile) { $stability = Get-Content -LiteralPath $stabilityProfile -Raw | ConvertFrom-Json if ( diff --git a/experiments/perception/worker/e28_physical_k1_local_surface_profile.json b/experiments/perception/worker/e28_worker_replay_local_surface_profile.json similarity index 78% rename from experiments/perception/worker/e28_physical_k1_local_surface_profile.json rename to experiments/perception/worker/e28_worker_replay_local_surface_profile.json index 3d3b5c4..0bb51c6 100644 --- a/experiments/perception/worker/e28_physical_k1_local_surface_profile.json +++ b/experiments/perception/worker/e28_worker_replay_local_surface_profile.json @@ -1,6 +1,6 @@ { "schema_version": "missioncore.e15-shadow-inference-profile/v1", - "mode": "physical-shadow-gate", + "mode": "worker-replay-gate", "authority": { "commands_enabled": false, "navigation_or_safety_accepted": false @@ -14,6 +14,17 @@ "calibration_slot": "camera_1", "calibration_sha256": "05f3ad9b38b3a4fc95388a8ec83da83c745e217709e51787b3d5aad0969f6fa9" }, + "replay_source": { + "session_id": "20260720T065719Z_viewer_live", + "display_name": "RAVNOVES00", + "selection": "complete-recording", + "speed": 1.0, + "minimum_source_span_seconds": 450.0, + "expected_camera_frames": 4489, + "expected_lidar_events": 4570, + "expected_pose_events": 4598, + "look_ahead": false + }, "transport": { "wire_schema": "missioncore.live-perception-wire/v1", "camera_media": "persistent-fmp4-pyav", @@ -46,15 +57,16 @@ "retention_seconds": 3.0, "result_capacity": 8, "acceptance": { - "minimum_bound_frames": 100, + "minimum_bound_frames": 4500, + "minimum_effective_fps": 9.0, "maximum_pose_miss_fraction": 0.05, "maximum_point_drop_fraction": 0.01, - "maximum_runtime_drop_fraction": 0.01, - "maximum_p95_result_age_ms": 80.0 + "maximum_runtime_drop_fraction": 0.08, + "maximum_p95_result_age_ms": 100.0 } }, "acceptance": { - "minimum_camera_frames": 140, + "minimum_camera_frames": 4400, "detector_minimum_effective_fps": 9.5, "detector_maximum_drop_fraction": 0.01, "semantic_minimum_effective_fps": 1.8, diff --git a/experiments/perception/worker/run_e15_shadow_inference.py b/experiments/perception/worker/run_e15_shadow_inference.py index cca5574..7193dc5 100644 --- a/experiments/perception/worker/run_e15_shadow_inference.py +++ b/experiments/perception/worker/run_e15_shadow_inference.py @@ -177,10 +177,16 @@ def read_live_profile(path: Path) -> tuple[dict[str, Any], str]: scheduling = profile.get("scheduling") temporal = profile.get("temporal") local_surface = profile.get("local_surface") + replay_source = profile.get("replay_source") acceptance = profile.get("acceptance") if ( profile.get("schema_version") != PROFILE_SCHEMA - or profile.get("mode") not in {"replay-shadow-gate", "physical-shadow-gate"} + or profile.get("mode") + not in { + "replay-shadow-gate", + "worker-replay-gate", + "physical-shadow-gate", + } or not all( isinstance(value, dict) for value in (source, authority, transport, scheduling, temporal, acceptance) @@ -213,6 +219,34 @@ def read_live_profile(path: Path) -> tuple[dict[str, Any], str]: or not 1 <= float(temporal.get("maximum_pose_point_delta_ms", 0)) <= 1000 ): raise RuntimeError("LAB E15 bounded runtime contract is invalid") + if profile.get("mode") == "worker-replay-gate": + replay_integer_contract = { + "expected_camera_frames": 4489, + "expected_lidar_events": 4570, + "expected_pose_events": 4598, + } + if ( + not isinstance(replay_source, dict) + or replay_source.get("session_id") + != "20260720T065719Z_viewer_live" + or replay_source.get("display_name") != "RAVNOVES00" + or replay_source.get("selection") != "complete-recording" + or float(replay_source.get("speed", 0)) != 1.0 + or float(replay_source.get("minimum_source_span_seconds", 0)) + < 450 + or replay_source.get("look_ahead") is not False + or any( + not isinstance(replay_source.get(key), int) + or isinstance(replay_source.get(key), bool) + or replay_source.get(key) != expected + for key, expected in replay_integer_contract.items() + ) + ): + raise RuntimeError( + "LAB E28 complete-recording worker replay contract is invalid" + ) + elif replay_source is not None: + raise RuntimeError("LAB E15 non-replay profile carries replay source state") if local_surface is not None: from k1link.compute.lidar_local_surface_geometry import ( DEFAULT_K1_LOCAL_SURFACE_PROFILE, @@ -247,7 +281,8 @@ def read_live_profile(path: Path) -> tuple[dict[str, Any], str]: else None ) if ( - profile.get("mode") != "physical-shadow-gate" + profile.get("mode") + not in {"worker-replay-gate", "physical-shadow-gate"} or not isinstance(local_surface, dict) or local_surface.get("enabled") is not True or local_surface.get("profile_id") @@ -281,9 +316,11 @@ def read_live_profile(path: Path) -> tuple[dict[str, Any], str]: local_acceptance.get("maximum_p95_result_age_ms", 0) ) <= 0 + or float(local_acceptance.get("minimum_effective_fps", 0)) + <= 0 ): raise RuntimeError( - "LAB E28 physical local-surface profile contract is invalid" + "LAB E28 worker local-surface profile contract is invalid" ) fractions = ( "detector_maximum_drop_fraction", @@ -334,6 +371,7 @@ def _local_surface_acceptance_checks( results = runtime.get("results") runtime_profile = runtime.get("profile") result_age = results.get("result_age_ms") if isinstance(results, dict) else None + delivery = runtime.get("delivery") if not isinstance(points, dict): points = {} if not isinstance(poses, dict): @@ -346,6 +384,8 @@ def _local_surface_acceptance_checks( runtime_profile = {} if not isinstance(result_age, dict): result_age = {} + if not isinstance(delivery, dict): + delivery = {} point_published = int(points.get("published", 0)) point_bound = int(points.get("bound", 0)) @@ -406,6 +446,10 @@ def _local_surface_acceptance_checks( "local_surface_maximum_runtime_drop_fraction": runtime_dropped / max(1, runtime_published) <= float(acceptance["maximum_runtime_drop_fraction"]), + "local_surface_minimum_effective_fps": float( + delivery.get("effective_fps", 0) + ) + >= float(acceptance["minimum_effective_fps"]), "local_surface_zero_runtime_failures": result_failed == 0, "local_surface_maximum_p95_result_age_ms": ( isinstance(p95_result_age, (int, float)) diff --git a/experiments/perception/worker/run_e28_local_surface_wire.py b/experiments/perception/worker/run_e28_local_surface_wire.py new file mode 100644 index 0000000..c4ead89 --- /dev/null +++ b/experiments/perception/worker/run_e28_local_surface_wire.py @@ -0,0 +1,615 @@ +#!/usr/bin/env python3 +"""Qualify the K1 local-surface runtime through the authenticated replay wire.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import os +import queue +import select +import threading +import time +from collections import Counter +from contextlib import suppress +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from run_e12_shadow_transport_probe import ( + _connect, + _decode_event, + _read_server_frame, + _send_client_frame, +) + +from k1link.compute.lidar_local_surface_geometry import ( + DEFAULT_K1_LOCAL_SURFACE_PROFILE, +) +from k1link.compute.lidar_local_surface_shadow import ( + K1LocalSurfaceShadowCoordinator, + K1LocalSurfaceShadowResult, +) +from k1link.data_plane import DecodedPointCloudView, DecodedPoseView +from k1link.device_plugins.xgrids_k1.protocol.normalizer import ( + normalize_k1_message, +) + +REPORT_SCHEMA = "missioncore.e28-local-surface-wire-report/v1" +PROFILE_SCHEMA = "missioncore.e15-shadow-inference-profile/v1" + + +@dataclass(frozen=True, slots=True) +class _WireMessage: + sequence: int + topic: str + payload: bytes + received_at_epoch_ns: int + received_monotonic_ns: int | None + source: str = "replay" + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + while block := stream.read(1024 * 1024): + digest.update(block) + return digest.hexdigest() + + +def _canonical(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode() + + +def _read_profile(path: Path) -> tuple[dict[str, Any], str]: + resolved = path.resolve(strict=True) + profile = json.loads(resolved.read_text(encoding="utf-8-sig")) + if not isinstance(profile, dict): + raise RuntimeError("LAB E28 worker replay profile is not an object") + replay = profile.get("replay_source") + local = profile.get("local_surface") + authority = profile.get("authority") + expected_profile_sha256 = hashlib.sha256( + _canonical(DEFAULT_K1_LOCAL_SURFACE_PROFILE.to_dict()) + ).hexdigest() + acceptance = local.get("acceptance") if isinstance(local, dict) else None + if ( + profile.get("schema_version") != PROFILE_SCHEMA + or profile.get("mode") != "worker-replay-gate" + or authority + != { + "commands_enabled": False, + "navigation_or_safety_accepted": False, + } + or not isinstance(replay, dict) + or replay.get("session_id") != "20260720T065719Z_viewer_live" + or replay.get("selection") != "complete-recording" + or float(replay.get("speed", 0)) != 1.0 + or replay.get("look_ahead") is not False + or not isinstance(local, dict) + or local.get("enabled") is not True + or local.get("profile_id") != DEFAULT_K1_LOCAL_SURFACE_PROFILE.profile_id + or local.get("profile_sha256") != expected_profile_sha256 + or not isinstance(acceptance, dict) + or float(acceptance.get("minimum_effective_fps", 0)) <= 0 + or float(acceptance.get("maximum_p95_result_age_ms", 0)) <= 0 + or any( + not 0 <= float(acceptance.get(key, -1)) <= 1 + for key in ( + "maximum_pose_miss_fraction", + "maximum_point_drop_fraction", + "maximum_runtime_drop_fraction", + ) + ) + ): + raise RuntimeError("LAB E28 worker replay profile contract is invalid") + return profile, _sha256(resolved) + + +def _fraction(numerator: int, denominator: int) -> float: + return numerator / max(1, denominator) + + +def _read_shadow_token(repository_root: Path) -> str: + path = ( + repository_root.resolve(strict=True) + / ".runtime" + / "live-perception" + / "shadow-worker.token" + ) + if not path.is_file() or path.is_symlink(): + raise RuntimeError("LAB E28 shadow token file is unavailable") + token = path.read_text(encoding="ascii").strip() + if not 40 <= len(token) <= 128 or not token.isascii(): + raise RuntimeError("LAB E28 shadow token is invalid") + return token + + +def _distribution(values: list[float]) -> dict[str, float | int | None]: + if not values: + return { + "sample_count": 0, + "p50": None, + "p95": None, + "maximum": None, + } + ordered = sorted(values) + + def percentile(fraction: float) -> float: + position = (len(ordered) - 1) * fraction + lower = math.floor(position) + upper = math.ceil(position) + if lower == upper: + return ordered[lower] + return ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower) + + return { + "sample_count": len(ordered), + "p50": percentile(0.5), + "p95": percentile(0.95), + "maximum": ordered[-1], + } + + +def run(args: argparse.Namespace) -> dict[str, Any]: + if ( + not math.isfinite(args.source_speed) + or not 0.1 <= args.source_speed <= 10 + or not math.isfinite(args.max_duration_seconds) + or args.max_duration_seconds <= 0 + or not math.isfinite(args.socket_timeout_seconds) + or args.socket_timeout_seconds <= 0 + ): + raise RuntimeError("LAB E28 worker runtime bounds are invalid") + profile, profile_sha256 = _read_profile(args.profile) + replay = profile["replay_source"] + local = profile["local_surface"] + local_acceptance = local["acceptance"] + output = args.output.resolve() + output.mkdir(mode=0o700, parents=True, exist_ok=False) + frames_path = output / "local-surface-frames.jsonl" + frame_count = 0 + all_processing_ms: list[float] = [] + all_result_age_ms: list[float] = [] + evidence_failures: list[str] = [] + evidence_queue_capacity = 32 + evidence_queue_maximum_depth = 0 + evidence_queue: queue.Queue[object] = queue.Queue( + maxsize=evidence_queue_capacity + ) + evidence_stop = object() + + with frames_path.open("x", encoding="utf-8", newline="\n") as frame_stream: + + def write_evidence() -> None: + nonlocal frame_count + failed = False + while True: + item = evidence_queue.get() + if item is evidence_stop: + return + if failed: + continue + try: + if not isinstance(item, K1LocalSurfaceShadowResult): + raise RuntimeError( + "LAB E28 evidence queue item is invalid" + ) + document = item.document() + document["point_evidence"] = { + "point_count": int(item.point_class.shape[0]), + "point_class_sha256": hashlib.sha256( + item.point_class.tobytes(order="C") + ).hexdigest(), + "point_height_sha256": hashlib.sha256( + item.point_height_m.tobytes(order="C") + ).hexdigest(), + "point_step_candidate_sha256": hashlib.sha256( + item.point_step_candidate.tobytes(order="C") + ).hexdigest(), + "source_alignment": "raw-k1-message-sequence", + } + frame_stream.write( + json.dumps( + document, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + + "\n" + ) + frame_count += 1 + except Exception as exc: + evidence_failures.append( + f"{type(exc).__name__}: {str(exc)[:240]}" + ) + failed = True + + evidence_thread = threading.Thread( + target=write_evidence, + name="e28-local-surface-evidence", + daemon=True, + ) + evidence_thread.start() + + def observe(result: K1LocalSurfaceShadowResult) -> None: + nonlocal evidence_queue_maximum_depth + if evidence_failures: + raise RuntimeError("LAB E28 evidence writer failed") + try: + evidence_queue.put_nowait(result) + except queue.Full as exc: + raise RuntimeError( + "LAB E28 bounded evidence queue overflowed" + ) from exc + evidence_queue_maximum_depth = max( + evidence_queue_maximum_depth, + evidence_queue.qsize(), + ) + all_processing_ms.append(result.processing_ms) + all_result_age_ms.append(result.result_age_ms) + + coordinator = K1LocalSurfaceShadowCoordinator( + point_capacity=int(local["point_queue_capacity"]), + pose_capacity=int(local["pose_buffer_capacity"]), + future_pose_wait_ms=float(local["future_pose_wait_ms"]), + retention_seconds=float(local["retention_seconds"]), + result_capacity=int(local["result_capacity"]), + result_observer=observe, + ) + token = _read_shadow_token(args.repository_root) + connection = None + stream = None + counts: Counter[str] = Counter() + payload_bytes: Counter[str] = Counter() + sensor_decode_ms: list[float] = [] + first_ingress_sequence: int | None = None + last_ingress_sequence: int | None = None + ingress_sequence_gaps = 0 + session_id: str | None = None + session_end_seen = False + first_source_epoch_ns: int | None = None + last_source_epoch_ns: int | None = None + timed_out = False + failures: list[str] = [] + started = time.perf_counter() + try: + connection, stream = _connect( + args.host, + args.port, + args.path, + token, + args.socket_timeout_seconds, + ) + deadline = time.monotonic() + args.max_duration_seconds + while time.monotonic() < deadline: + readable, _, _ = select.select( + [connection], + [], + [], + min(0.25, max(0.0, deadline - time.monotonic())), + ) + if not readable: + continue + opcode, frame = _read_server_frame(stream) + if opcode == 0x8: + break + if opcode == 0x9: + _send_client_frame(stream, 0xA, frame) + continue + if opcode != 0x2: + raise RuntimeError(f"LAB E28 received unexpected websocket opcode {opcode}") + header, payload = _decode_event(frame) + sequence = int(header["ingress_sequence"]) + if last_ingress_sequence is not None: + if sequence <= last_ingress_sequence: + raise RuntimeError("LAB E28 ingress sequence is not increasing") + ingress_sequence_gaps += max( + 0, + sequence - last_ingress_sequence - 1, + ) + if first_ingress_sequence is None: + first_ingress_sequence = sequence + last_ingress_sequence = sequence + event_session_id = str(header["session_id"]) + if session_id is None: + session_id = event_session_id + coordinator.begin_session(session_id) + elif session_id != event_session_id: + raise RuntimeError("LAB E28 replay session identity changed") + modality = str(header["modality"]) + counts[modality] += 1 + payload_bytes[modality] += len(payload) + if modality != "control": + captured_at_epoch_ns = int(header["captured_at_epoch_ns"]) + first_source_epoch_ns = ( + captured_at_epoch_ns + if first_source_epoch_ns is None + else min(first_source_epoch_ns, captured_at_epoch_ns) + ) + last_source_epoch_ns = ( + captured_at_epoch_ns + if last_source_epoch_ns is None + else max(last_source_epoch_ns, captured_at_epoch_ns) + ) + if modality == "control": + control = json.loads(payload) + if control.get("event") == "session-end": + session_end_seen = True + break + continue + if modality not in {"lidar", "pose"}: + continue + decode_started = time.perf_counter() + normalized = normalize_k1_message( + _WireMessage( + sequence=int(header["source_sequence"]), + topic=str(header["source_id"]), + payload=payload, + received_at_epoch_ns=int(header["captured_at_epoch_ns"]), + received_monotonic_ns=int(header["received_monotonic_ns"]), + ), + processing_started_monotonic_ns=time.monotonic_ns(), + ) + sensor_decode_ms.append((time.perf_counter() - decode_started) * 1000) + if modality == "lidar" and isinstance( + normalized, + DecodedPointCloudView, + ): + coordinator.publish_point_cloud(normalized) + elif modality == "pose" and isinstance( + normalized, + DecodedPoseView, + ): + coordinator.publish_pose(normalized) + else: + raise RuntimeError("LAB E28 known sensor modality did not normalize") + else: + timed_out = True + except Exception as exc: + failures.append(f"{type(exc).__name__}: {str(exc)[:240]}") + finally: + try: + coordinator.close(timeout_seconds=30) + except Exception as exc: + failures.append(f"{type(exc).__name__}: {str(exc)[:240]}") + if stream is not None: + with suppress(Exception): + _send_client_frame(stream, 0x8, b"") + with suppress(Exception): + stream.close() + if connection is not None: + with suppress(Exception): + connection.close() + evidence_queue.put(evidence_stop) + evidence_thread.join(timeout=30) + if evidence_thread.is_alive(): + failures.append("LAB E28 evidence writer did not stop") + failures.extend(evidence_failures) + frame_stream.flush() + os.fsync(frame_stream.fileno()) + + snapshot = coordinator.snapshot() + binder = snapshot["binder"] + runtime = snapshot["runtime"] + if not isinstance(binder, dict) or not isinstance(runtime, dict): + raise RuntimeError("LAB E28 local-surface runtime did not initialize") + points = binder["points"] + queue_state = runtime["queue"] + results = runtime["results"] + delivery = runtime["delivery"] + if ( + not isinstance(points, dict) + or not isinstance(queue_state, dict) + or not isinstance(results, dict) + or not isinstance(delivery, dict) + ): + raise RuntimeError("LAB E28 local-surface telemetry is invalid") + point_published = int(points["published"]) + point_bound = int(points["bound"]) + point_missed = int(points["missed"]) + point_dropped = int(points["dropped_overflow"]) + runtime_published = int(queue_state["published"]) + runtime_consumed = int(queue_state["consumed"]) + runtime_dropped = int(queue_state["dropped_overflow"]) + full_processing_distribution = _distribution(all_processing_ms) + full_result_age_distribution = _distribution(all_result_age_ms) + p95_result_age = full_result_age_distribution["p95"] + source_span_seconds = ( + 0.0 + if first_source_epoch_ns is None or last_source_epoch_ns is None + else (last_source_epoch_ns - first_source_epoch_ns) / 1_000_000_000 + ) + checks = { + "exact_source_speed": math.isclose( + args.source_speed, + float(replay["speed"]), + rel_tol=0, + abs_tol=1e-9, + ), + "complete_recording_span": source_span_seconds + >= float(replay["minimum_source_span_seconds"]), + "exact_camera_event_count": counts["camera-frame"] == int(replay["expected_camera_frames"]), + "exact_lidar_event_count": counts["lidar"] == int(replay["expected_lidar_events"]), + "exact_pose_event_count": counts["pose"] == int(replay["expected_pose_events"]), + "session_end_seen": session_end_seen, + "zero_ingress_sequence_gaps": ingress_sequence_gaps == 0, + "minimum_bound_frames": point_bound >= int(local_acceptance["minimum_bound_frames"]), + "binder_accounting": point_bound + point_missed + point_dropped + int(points["depth"]) + == point_published, + "binder_to_runtime_accounting": point_bound == runtime_published, + "maximum_pose_miss_fraction": _fraction( + point_missed, + point_published, + ) + <= float(local_acceptance["maximum_pose_miss_fraction"]), + "maximum_point_drop_fraction": _fraction( + point_dropped, + point_published, + ) + <= float(local_acceptance["maximum_point_drop_fraction"]), + "runtime_accounting": runtime_consumed + runtime_dropped + int(queue_state["depth"]) + == runtime_published, + "maximum_runtime_drop_fraction": _fraction( + runtime_dropped, + runtime_published, + ) + <= float(local_acceptance["maximum_runtime_drop_fraction"]), + "minimum_effective_fps": float(delivery["effective_fps"]) + >= float(local_acceptance["minimum_effective_fps"]), + "zero_runtime_failures": int(results["failed"]) == 0, + "persisted_frame_accounting": frame_count == int(results["published"]), + "maximum_p95_result_age_ms": ( + isinstance(p95_result_age, int | float) + and not isinstance(p95_result_age, bool) + and math.isfinite(float(p95_result_age)) + and float(p95_result_age) <= float(local_acceptance["maximum_p95_result_age_ms"]) + ), + "shadow_authority_only": snapshot["authority"] + == { + "commands_enabled": False, + "navigation_or_safety_accepted": False, + }, + "zero_failures": not failures, + "zero_timeout": not timed_out, + } + accepted = all(checks.values()) + report = { + "schema_version": REPORT_SCHEMA, + "created_at_utc": datetime.now(UTC) + .isoformat(timespec="milliseconds") + .replace("+00:00", "Z"), + "state": "accepted" if accepted else "rejected", + "ground_truth": False, + "identity": { + "profile_sha256": profile_sha256, + "session_id": session_id, + "source_session_id": replay["session_id"], + "runner_sha256": _sha256(Path(__file__).resolve(strict=True)), + }, + "source": { + "selection": "complete-recording", + "speed": args.source_speed, + "source_span_seconds": source_span_seconds, + "look_ahead": False, + }, + "transport": { + "counts": dict(sorted(counts.items())), + "payload_bytes": dict(sorted(payload_bytes.items())), + "first_ingress_sequence": first_ingress_sequence, + "last_ingress_sequence": last_ingress_sequence, + "ingress_sequence_gaps": ingress_sequence_gaps, + "session_end_seen": session_end_seen, + "timed_out": timed_out, + }, + "local_surface": snapshot, + "worker": { + "wall_seconds": time.perf_counter() - started, + "sensor_decode_ms": { + **_distribution(sensor_decode_ms), + }, + "local_surface_processing_ms": full_processing_distribution, + "local_surface_result_age_ms": full_result_age_distribution, + "persisted_frames": frame_count, + "evidence_queue": { + "capacity": evidence_queue_capacity, + "maximum_depth": evidence_queue_maximum_depth, + "depth": evidence_queue.qsize(), + "failures": len(evidence_failures), + }, + }, + "artifacts": { + "local_surface_frames": { + "path": frames_path.name, + "byte_length": frames_path.stat().st_size, + "sha256": _sha256(frames_path), + } + }, + "acceptance": { + "accepted": accepted, + "checks": checks, + "commands_enabled": False, + "navigation_or_safety_accepted": False, + }, + "failures": failures, + "limitations": [ + "Recorded K1 evidence is replayed through the authenticated wire; " + "this is not a physical acquisition test.", + "Recorded host-arrival timing is preserved; there is no hardware clock qualification.", + "Observed points do not prove free or traversable unknown space.", + ], + } + report_sha256 = hashlib.sha256(_canonical(report)).hexdigest() + report["result_id"] = f"e28-local-surface-wire-{report_sha256}" + report_path = output / "report.json" + report_path.write_bytes( + json.dumps( + report, + ensure_ascii=False, + indent=2, + sort_keys=True, + allow_nan=False, + ).encode() + + b"\n" + ) + return report + + +def _arguments() -> argparse.Namespace: + script = Path(__file__).resolve() + root = script.parents[3] if len(script.parents) > 3 else Path.cwd() + parser = argparse.ArgumentParser() + parser.add_argument( + "--repository-root", + type=Path, + default=root, + ) + parser.add_argument( + "--profile", + type=Path, + default=Path(__file__).with_name("e28_worker_replay_local_surface_profile.json"), + ) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8028) + parser.add_argument( + "--path", + default=("/api/v1/device-plugins/nodedc.device.xgrids-lixelkity-k1/live-perception-shadow"), + ) + parser.add_argument("--socket-timeout-seconds", type=float, default=30.0) + parser.add_argument("--max-duration-seconds", type=float, default=600.0) + parser.add_argument("--source-speed", type=float, default=1.0) + parser.add_argument("--output", type=Path, required=True) + return parser.parse_args() + + +def main() -> int: + report = run(_arguments()) + print( + json.dumps( + { + "result_id": report["result_id"], + "state": report["state"], + "source": report["source"], + "transport": report["transport"], + "worker": report["worker"], + "acceptance": report["acceptance"], + }, + ensure_ascii=False, + indent=2, + sort_keys=True, + ) + ) + return 0 if report["acceptance"]["accepted"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/k1link/compute/lidar_local_surface.py b/src/k1link/compute/lidar_local_surface.py index 241e4fa..191ef36 100644 --- a/src/k1link/compute/lidar_local_surface.py +++ b/src/k1link/compute/lidar_local_surface.py @@ -1327,8 +1327,8 @@ def build_k1_local_surface( "status": "replay-experiment-only", "production_promotion": False, "next_gate": ( - "run the accepted profile through a bounded latest-wins live-shadow " - "queue without commands, free-space or safety authority" + "qualify a separate dynamic-observation layer; physical K1 " + "acquisition remains a later commissioning gate" ), }, "authority": { diff --git a/src/k1link/compute/lidar_local_surface_shadow.py b/src/k1link/compute/lidar_local_surface_shadow.py index 428f7e5..dedf325 100644 --- a/src/k1link/compute/lidar_local_surface_shadow.py +++ b/src/k1link/compute/lidar_local_surface_shadow.py @@ -4,6 +4,7 @@ import math import threading import time from collections import Counter, deque +from collections.abc import Callable from dataclasses import asdict, dataclass from typing import Any, Final, Literal @@ -767,6 +768,7 @@ class K1LocalSurfaceShadowRuntime: profile: K1LocalSurfaceProfile = DEFAULT_K1_LOCAL_SURFACE_PROFILE, queue_capacity: int = 2, result_capacity: int = 8, + result_observer: Callable[[K1LocalSurfaceShadowResult], None] | None = None, ) -> None: if ( not session_id @@ -779,6 +781,7 @@ class K1LocalSurfaceShadowRuntime: self.profile = profile self._queue = LatestWinsQueue[K1LocalSurfaceShadowInput](queue_capacity) self._result_capacity = result_capacity + self._result_observer = result_observer self._results: deque[K1LocalSurfaceShadowResult] = deque(maxlen=result_capacity) self._result_dropped = 0 self._processed = 0 @@ -788,6 +791,8 @@ class K1LocalSurfaceShadowRuntime: self._result_age_ms: deque[float] = deque(maxlen=512) self._last_error: str | None = None self._inflight = False + self._first_published_monotonic_ns: int | None = None + self._last_published_monotonic_ns: int | None = None self._condition = threading.Condition() self._closed = False self._estimator = K1LocalSurfaceShadowEstimator(profile) @@ -802,6 +807,9 @@ class K1LocalSurfaceShadowRuntime: with self._condition: if self._closed: raise RuntimeError("K1 local-surface shadow runtime is closed") + if self._first_published_monotonic_ns is None: + self._first_published_monotonic_ns = value.published_monotonic_ns + self._last_published_monotonic_ns = value.published_monotonic_ns self._queue.publish(value) def publish_views( @@ -849,6 +857,16 @@ class K1LocalSurfaceShadowRuntime: queue_snapshot = self._queue.snapshot() with self._condition: latest = self._results[-1] if self._results else None + source_span_seconds = ( + 0.0 + if self._first_published_monotonic_ns is None + or self._last_published_monotonic_ns is None + else ( + self._last_published_monotonic_ns + - self._first_published_monotonic_ns + ) + / 1_000_000_000 + ) return { "schema_version": K1_LOCAL_SURFACE_SHADOW_SCHEMA, "mode": "live-shadow-diagnostic-only", @@ -856,6 +874,14 @@ class K1LocalSurfaceShadowRuntime: "profile": self.profile.to_dict(), "queue_policy": "bounded-latest-wins", "queue": asdict(queue_snapshot), + "delivery": { + "source_span_seconds": source_span_seconds, + "effective_fps": ( + self._processed / source_span_seconds + if source_span_seconds > 0 + else 0.0 + ), + }, "results": { "capacity": self._result_capacity, "depth": len(self._results), @@ -891,7 +917,9 @@ class K1LocalSurfaceShadowRuntime: self._inflight = True try: result = self._estimator.process(value) - except (LidarGroundError, ValueError, np.linalg.LinAlgError) as exc: + if self._result_observer is not None: + self._result_observer(result) + except Exception as exc: with self._condition: self._failed += 1 self._last_error = type(exc).__name__ @@ -922,10 +950,12 @@ class K1LocalSurfaceShadowCoordinator: future_pose_wait_ms: float = 25.0, retention_seconds: float = 3.0, result_capacity: int = 8, + result_observer: Callable[[K1LocalSurfaceShadowResult], None] | None = None, ) -> None: self.profile = profile self._queue_capacity = point_capacity self._result_capacity = result_capacity + self._result_observer = result_observer self._binder = K1LocalSurfacePoseBinder( maximum_pose_binding_ms=profile.maximum_pose_binding_ms, point_capacity=point_capacity, @@ -953,6 +983,7 @@ class K1LocalSurfaceShadowCoordinator: profile=self.profile, queue_capacity=self._queue_capacity, result_capacity=self._result_capacity, + result_observer=self._result_observer, ) self._session_id = session_id diff --git a/src/k1link/compute/live_perception.py b/src/k1link/compute/live_perception.py index 1b7be17..033b34f 100644 --- a/src/k1link/compute/live_perception.py +++ b/src/k1link/compute/live_perception.py @@ -363,7 +363,7 @@ class LivePerceptionIngress: "control": 4, "camera-init": 1, "camera-frame": 2, - "lidar": 2, + "lidar": 8, "pose": 16, } _MAX_PAYLOAD_BYTES: Final[dict[LiveIngressModality, int]] = { diff --git a/tests/test_e15_shadow_inference.py b/tests/test_e15_shadow_inference.py index 0e86a0e..b29815f 100644 --- a/tests/test_e15_shadow_inference.py +++ b/tests/test_e15_shadow_inference.py @@ -30,6 +30,28 @@ def _module() -> object: sys.path.pop(0) +def _e28_module() -> object: + worker = ( + Path(__file__).resolve().parents[1] + / "experiments" + / "perception" + / "worker" + ) + sys.path.insert(0, str(worker)) + try: + spec = importlib.util.spec_from_file_location( + "e28_local_surface_wire_test", + worker / "run_e28_local_surface_wire.py", + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + finally: + sys.path.pop(0) + + def test_e15_profile_pins_replay_shadow_authority_and_bounded_runtime() -> None: module = _module() profile_path = ( @@ -84,20 +106,31 @@ def test_e15_profile_rejects_command_authority(tmp_path: Path) -> None: module.read_live_profile(changed) -def test_e28_profile_pins_physical_k1_local_surface_gate() -> None: +def test_e28_profile_pins_complete_recording_worker_local_surface_gate() -> None: module = _module() profile_path = ( Path(__file__).resolve().parents[1] / "experiments" / "perception" / "worker" - / "e28_physical_k1_local_surface_profile.json" + / "e28_worker_replay_local_surface_profile.json" ) profile, digest = module.read_live_profile(profile_path) assert len(digest) == 64 - assert profile["mode"] == "physical-shadow-gate" + assert profile["mode"] == "worker-replay-gate" + assert profile["replay_source"] == { + "session_id": "20260720T065719Z_viewer_live", + "display_name": "RAVNOVES00", + "selection": "complete-recording", + "speed": 1.0, + "minimum_source_span_seconds": 450.0, + "expected_camera_frames": 4489, + "expected_lidar_events": 4570, + "expected_pose_events": 4598, + "look_ahead": False, + } assert profile["local_surface"] == { "enabled": True, "profile_id": "k1-vendor-map-dynamic-local-surface/v1", @@ -110,11 +143,12 @@ def test_e28_profile_pins_physical_k1_local_surface_gate() -> None: "retention_seconds": 3.0, "result_capacity": 8, "acceptance": { - "minimum_bound_frames": 100, + "minimum_bound_frames": 4500, + "minimum_effective_fps": 9.0, "maximum_pose_miss_fraction": 0.05, "maximum_point_drop_fraction": 0.01, - "maximum_runtime_drop_fraction": 0.01, - "maximum_p95_result_age_ms": 80.0, + "maximum_runtime_drop_fraction": 0.08, + "maximum_p95_result_age_ms": 100.0, }, } assert profile["authority"] == { @@ -130,17 +164,63 @@ def test_e28_profile_rejects_unpinned_local_surface(tmp_path: Path) -> None: / "experiments" / "perception" / "worker" - / "e28_physical_k1_local_surface_profile.json" + / "e28_worker_replay_local_surface_profile.json" ) value = json.loads(source.read_text()) value["local_surface"]["profile_sha256"] = "0" * 64 changed = tmp_path / "unpinned-local-surface.json" changed.write_text(json.dumps(value)) - with pytest.raises(RuntimeError, match="E28 physical local-surface"): + with pytest.raises(RuntimeError, match="E28 worker local-surface"): module.read_live_profile(changed) +def test_e28_wire_runner_accepts_the_same_complete_recording_profile() -> None: + module = _e28_module() + profile_path = ( + Path(__file__).resolve().parents[1] + / "experiments" + / "perception" + / "worker" + / "e28_worker_replay_local_surface_profile.json" + ) + + profile, digest = module._read_profile(profile_path) + + assert len(digest) == 64 + assert profile["mode"] == "worker-replay-gate" + assert profile["replay_source"]["selection"] == "complete-recording" + assert profile["replay_source"]["expected_lidar_events"] == 4570 + assert profile["local_surface"]["point_queue_capacity"] == 2 + assert profile["authority"]["commands_enabled"] is False + + +def test_e28_wire_distribution_uses_the_complete_sample() -> None: + module = _e28_module() + + assert module._distribution([]) == { + "sample_count": 0, + "p50": None, + "p95": None, + "maximum": None, + } + assert module._distribution([10.0, 20.0, 30.0, 40.0]) == { + "sample_count": 4, + "p50": 25.0, + "p95": 38.5, + "maximum": 40.0, + } + + +def test_e28_wire_reads_the_private_token_from_a_file(tmp_path: Path) -> None: + module = _e28_module() + token_root = tmp_path / ".runtime" / "live-perception" + token_root.mkdir(parents=True) + (token_root / "shadow-worker.token").write_text(f"{'a' * 43}\n") + + assert module._read_shadow_token(tmp_path) == "a" * 43 + + def test_e28_local_surface_acceptance_requires_exact_bounded_accounting() -> None: module = _module() config = { @@ -149,6 +229,7 @@ def test_e28_local_surface_acceptance_requires_exact_bounded_accounting() -> Non "pose_buffer_capacity": 16, "acceptance": { "minimum_bound_frames": 100, + "minimum_effective_fps": 9.0, "maximum_pose_miss_fraction": 0.05, "maximum_point_drop_fraction": 0.01, "maximum_runtime_drop_fraction": 0.01, @@ -179,6 +260,10 @@ def test_e28_local_surface_acceptance_requires_exact_bounded_accounting() -> Non }, "runtime": { "closed": True, + "delivery": { + "source_span_seconds": 10.0, + "effective_fps": 10.0, + }, "profile": { "profile_id": "k1-vendor-map-dynamic-local-surface/v1", }, @@ -214,6 +299,14 @@ def test_e28_local_surface_acceptance_requires_exact_bounded_accounting() -> Non ] is False ) + snapshot["runtime"]["results"]["result_age_ms"]["p95"] = 40.0 + snapshot["runtime"]["delivery"]["effective_fps"] = 8.99 + assert ( + module._local_surface_acceptance_checks(snapshot, config)[ + "local_surface_minimum_effective_fps" + ] + is False + ) def test_persistent_worker_request_is_single_run_d_backed_and_token_bounded( diff --git a/tests/test_e21_realtime_envelope.py b/tests/test_e21_realtime_envelope.py index 6cd88aa..a800cf9 100644 --- a/tests/test_e21_realtime_envelope.py +++ b/tests/test_e21_realtime_envelope.py @@ -55,6 +55,48 @@ def test_camera_replay_loads_segments_lazily_and_checks_each_digest(tmp_path: Pa next(events) +def test_camera_replay_complete_recording_mode_has_no_duration_cutoff( + tmp_path: Path, +) -> None: + module = _module( + "e28_complete_source_replay_test", + "experiments/perception/run_e12_shadow_transport_replay.py", + ) + epoch = tmp_path / "epoch-1" + segments = epoch / "segments" + segments.mkdir(parents=True) + (epoch / "init.mp4").write_bytes(b"init") + rows = [] + for sequence, epoch_ns in ((1, 100_000_000), (2, 2_100_000_000)): + payload = f"frame-{sequence}".encode() + path = segments / f"{sequence}.m4s" + path.write_bytes(payload) + rows.append( + { + "host_epoch_ns": epoch_ns, + "host_monotonic_ns": epoch_ns, + "length": len(payload), + "path": f"segments/{sequence}.m4s", + "sequence": sequence, + "sha256": hashlib.sha256(payload).hexdigest(), + } + ) + (epoch / "index.jsonl").write_text( + "".join(json.dumps(row) + "\n" for row in rows), + encoding="utf-8", + ) + + bounded = list(module._camera_events(epoch, 1.0)) + complete = list(module._camera_events(epoch, None)) + + assert [event.modality for event in bounded] == ["camera-init", "camera-frame"] + assert [event.modality for event in complete] == [ + "camera-init", + "camera-frame", + "camera-frame", + ] + + def test_e21_trend_reports_accumulating_age_and_stable_memory() -> None: module = _module( "e21_analyzer_test", diff --git a/tests/test_lidar_local_surface.py b/tests/test_lidar_local_surface.py index 8a3dd9b..0e22644 100644 --- a/tests/test_lidar_local_surface.py +++ b/tests/test_lidar_local_surface.py @@ -360,6 +360,8 @@ def test_k1_local_surface_shadow_matches_replay_and_stays_non_authoritative( assert snapshot["queue"]["published"] == len(published) assert snapshot["queue"]["consumed"] == len(published) assert snapshot["queue"]["dropped_overflow"] == 0 + assert snapshot["delivery"]["source_span_seconds"] > 0 + assert snapshot["delivery"]["effective_fps"] > 0 assert snapshot["results"]["failed"] == 0 assert snapshot["occupancy_policy"]["absence_of_points_means_free"] is False assert snapshot["authority"]["commands_enabled"] is False diff --git a/tests/test_live_perception.py b/tests/test_live_perception.py index f127116..7af23cb 100644 --- a/tests/test_live_perception.py +++ b/tests/test_live_perception.py @@ -31,7 +31,7 @@ def test_live_ingress_keeps_modalities_separately_bounded_and_ordered() -> None: received_monotonic_ns=200 + sequence, payload=f"camera-{sequence}".encode(), ) - for sequence in range(3): + for sequence in range(9): assert ingress.publish( modality="lidar", source_id="lixel/application/report/lio_pcl", @@ -53,7 +53,7 @@ def test_live_ingress_keeps_modalities_separately_bounded_and_ordered() -> None: snapshot = ingress.snapshot() assert snapshot["queues"]["camera-frame"]["depth"] == 2 assert snapshot["queues"]["camera-frame"]["dropped_overflow"] == 3 - assert snapshot["queues"]["lidar"]["depth"] == 2 + assert snapshot["queues"]["lidar"]["depth"] == 8 assert snapshot["queues"]["lidar"]["dropped_overflow"] == 1 assert snapshot["queues"]["pose"]["depth"] == 16 assert snapshot["queues"]["pose"]["dropped_overflow"] == 4 @@ -69,7 +69,9 @@ def test_live_ingress_keeps_modalities_separately_bounded_and_ordered() -> None: 3, 4, ] - assert [event.source_sequence for event in events if event.modality == "lidar"] == [1, 2] + assert [event.source_sequence for event in events if event.modality == "lidar"] == list( + range(1, 9) + ) assert [event.source_sequence for event in events if event.modality == "pose"] == list( range(4, 20) ) diff --git a/tests/test_live_perception_synchronizer.py b/tests/test_live_perception_synchronizer.py index 34fb2b1..3fcdd4a 100644 --- a/tests/test_live_perception_synchronizer.py +++ b/tests/test_live_perception_synchronizer.py @@ -183,7 +183,11 @@ def test_local_surface_pose_binder_fails_closed_after_pose_deadline() -> None: def test_local_surface_shadow_coordinator_closes_exact_accounting() -> None: - coordinator = K1LocalSurfaceShadowCoordinator(result_capacity=2) + observed = [] + coordinator = K1LocalSurfaceShadowCoordinator( + result_capacity=2, + result_observer=observed.append, + ) coordinator.begin_session("physical-k1-shadow-test") assert coordinator.publish_pose(_pose(1, 1_000_000_000)) == 0 assert coordinator.publish_point_cloud(_points(1, 1_010_000_000)) == 1 @@ -196,3 +200,5 @@ def test_local_surface_shadow_coordinator_closes_exact_accounting() -> None: assert snapshot["runtime"]["queue"]["consumed"] == 1 assert snapshot["runtime"]["results"]["failed"] == 0 assert snapshot["authority"]["navigation_or_safety_accepted"] is False + assert len(observed) == 1 + assert observed[0].frame_index == 1