From fd8313899b16b16a578529909b5b3c2ac5a03c0b Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 2 Sep 2026 01:09:03 +0300 Subject: [PATCH] fix(perception): separate preroll history from current sensor binding Preserve rolling points and original clocks; expose per-modality ages, held pose and rejection reasons. Saved-lineage audit changes only the first admission (75 to 76 of 128). No new model/performance claim. 384 focused Python and 62 frontend tests passed. --- .../OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md | 6 +- .../PERCEPTION_STREAM_STAGE1_2026-09-01.md | 50 +++++++ .../pilot_sensor_binding.py | 129 ++++++++++++++++++ .../run_joint_pilot.py | 48 +++---- tests/test_perception_joint_pilot.py | 118 ++++++++++++++++ 5 files changed, 323 insertions(+), 28 deletions(-) create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_sensor_binding.py diff --git a/docs/OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md b/docs/OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md index af6bf6f..c6fd28f 100644 --- a/docs/OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md +++ b/docs/OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md @@ -2,6 +2,8 @@ Дата: 2026-09-01; обновлено 2026-09-02 00:41 МСК. Текущая цель — переносимые полные профили и снижение задержек всей цепочки; допустимый лабораторный overload не блокирует разработку и не выдаётся за real-time PASS. В двух новых source-paced пилотах с перекрытием CPU и последовательного GPU этапа получены 128/128 результатов без drops; source→local receiver p95 143.32/153.44 ms, p99 198.22/194.25 ms. Среднее ожидание очереди 7.20/8.31 ms против 13.65/25.48 ms в serial-контролях. Это перспективный экспериментальный scheduler, не квалификация и не замена рабочего runtime. Этап 1 частично выполнен; этапы 2–4 не начаты. Static-obstacle слой и coarse `hard_surface` остаются достаточными для сельского прототипа; профиль не сводится к одной модели. +Последнее изменение, 2026-09-02 01:06 МСК: выполнен Git checkpoint и отдельное исправление source preroll binding. Metadata-only аудит прежних 128 кадров меняет admission только кадра 0: 75 → 76 допустимых cloud/pose пар. Это не новый модельный прогон; старые latency-результаты выше остаются последними измеренными. Следующий gate — bounded Worker pilot нового source binding, включая влияние первого нового geometry frame на последующие temporal outputs. + Текущий evidence: [отчёт этапа 1](../experiments/perception/PERCEPTION_STREAM_STAGE1_2026-09-01.md), [ADR 0049](adr/0049-stream-first-perception-profiles.md), [candidate manifest](../config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json). Запрет full-source preload закреплён в новом контракте; старый batch materializer пока не удалён и продуктовый путь не переключён. ## Цель и граница завершения @@ -230,7 +232,9 @@ EoMT — семейство ViT-моделей сегментации изобр - 2026-09-02 00:20–00:41 МСК: уточнение владельца о долгосрочном экспериментариуме внесено в план, ADR, Desktop status и candidate `experiment_policy`. Тяжёлые профили сохраняются; functional, quality и hardware/source/transport-specific performance статусы независимы. - 2026-09-02: добавлены CUDA-event/host-IPC timing и экспериментальные варианты DDRNet layout/CUDA Graph; оба не показали устойчивого выигрыша полного графа и не стали defaults. Затем реализован `--schedule overlap-cpu`: один serial GPU worker, один chronological CPU consumer, общий бюджет двух pending кадров, единый bounded учёт входов. Первый fixed-slot вариант потерял один кадр и сохранён как отрицательный результат; dynamic shared-slot вариант дважды дал 128/128 без drops. Маски, proposals, metric observations, tracks, threats, costmap и non-timing TGS output совпали с reference на всех 128 кадрах; age-based policy проверяется отдельно. - 2026-09-02: 91 focused tests PASS (12 pilot, 52 contract, 27 existing live-ingress/synchronizer/shadow), Ruff PASS, Python 3.9 child lint PASS. Семь последовательных Worker runs и версии кода сохранены в `.runtime/perception-latency-20260902T0020MSK/manifest.json`. Worker-сервисы восстановлены, Triton ready=200, local 8000=200; Frigate/Ollama exited/restart=no. Никаких продуктовых defaults/registry cutover. -- Этап 1: частично выполнен; совместный пилот и ограниченный latency experiment выполнены. Performance gate остаётся FAIL, но сам по себе больше не является запретом экспериментальной упаковки. Следующие задачи: устранить source-adapter preroll/layered-freshness разрыв, перенести измеренное расписание в общий runtime, затем измерить transport/application boundary. DDRNet/online-surface jitter остаётся отдельным profiling направлением. Нельзя закрывать whole-path gate по compute-only метрике или отсутствию drops. +- 2026-09-02 01:06 МСК: по запросу владельца сохранены все 122 изменённых/новых исходных файла семью тематическими коммитами; чистая контрольная точка `ffd6be5`. Коммиты: `d655d69` equipment/capture, `a945d66` backend lifecycle, `62d5520` Worker packaging, `5a78c99` portable UI, `4e04d06` real-time contracts, `bfe6ef4` streaming pilot, `ffd6be5` plan/evidence. Push не выполнялся; raw/weights/runtime не включены. Отдельный `05c99ef` исправляет девять legacy test failures: исторические fixtures отделены от новых installed-package pins, старый image явно отвергается; production digest checks не ослаблены. +- 2026-09-02: следующий source-binding increment отделяет первый preroll от текущих LiDAR increments, сохраняет history-only timestamps/points в rolling/TGS и добавляет поэлементные причины unavailable/stale/held. Metadata audit: первый кадр получает 4,787 current points, 7,207 preroll points остаются history-only; остальные 127 admissions и increment identities неизменны. 384 focused Python tests и 62 выбранных frontend architecture/Observatory tests PASS. Нового GPU run, frontend build, deployment или product cutover в этой итерации не было. +- Этап 1: частично выполнен; совместный пилот и ограниченный latency experiment выполнены. Performance gate остаётся FAIL, но сам по себе больше не является запретом экспериментальной упаковки. Preroll source fix проверен unit/integration и metadata-only аудитом, ещё не новым модельным прогоном. Следующие задачи: повторить bounded Worker pilot, довести выходной layered-freshness ABI, перенести расписание в общий runtime, затем измерить transport/application boundary. DDRNet/online-surface jitter остаётся отдельным profiling направлением. Нельзя закрывать whole-path gate по compute-only метрике или отсутствию drops. - Этапы 2, 3, 4: не начаты. Продуктовый backend/frontend path, active registry, canonical local 8000 и физический источник не переключались. Старый full-source materializer остаётся legacy; новый streaming transport ещё не реализован. ## Surprises / открытые вопросы diff --git a/experiments/perception/PERCEPTION_STREAM_STAGE1_2026-09-01.md b/experiments/perception/PERCEPTION_STREAM_STAGE1_2026-09-01.md index 5b4eba8..800bd67 100644 --- a/experiments/perception/PERCEPTION_STREAM_STAGE1_2026-09-01.md +++ b/experiments/perception/PERCEPTION_STREAM_STAGE1_2026-09-01.md @@ -507,3 +507,53 @@ Standalone image и приложение↔Worker transport всё ещё не стендом. Следующая работа — preroll/layered freshness, общий runtime с измеряемыми очередями и реальный binary data plane; performance FAIL на 4090 не повод выбросить профиль или остановить развитие экспериментариума. + +## Git checkpoint и source preroll binding — 2026-09-02 01:06 МСК + +Все накопленные исходные изменения сохранены семью тематическими коммитами +до чистой контрольной точки `ffd6be5`. Веса, записи, subprocess logs и `.runtime` +в Git не включались, push не выполнялся. Девять обнаруженных legacy test failures +исправлены отдельным `05c99ef`: тесты старого orchestration contract используют +конфигурационные fixtures из `7025e17`; текущие installed runtime assertions +проверяют текущие pins и отдельно отвергают прежний agent image. Runtime, +manifest readiness и production hash validators ради тестов не менялись. + +Следующий шаг плана реализован в `pilot_sensor_binding.py` и source producer: + +- Старые preroll increments больше не превращают всю первую LiDAR/pose пару + в unavailable. Для первого current frame выбираются только causal increments + внутри прежних age/skew границ; прочие сохраняются в bounded rolling/TGS. +- Идентичности history-only increments опубликованы отдельно. Их timestamps + не заменяются временем camera; ни один такой point не считается новым. +- У pose/points есть отдельные `current`/`held`/`stale`/`unavailable` состояния, + исходные ages и причины отказа пары. Held pose сохраняет исходное время; + отсутствие нового point increment по-прежнему не считается новой геометрией. +- После первой камеры admission прежних increments не фильтруется и не + смягчается. Это промежуточный source-binding ABI, не завершённая реализация + всех `LayerEvidence` на выходах полного профиля. + +Metadata-only сверка сохранённого `reference-128/scenes.jsonl`: 75 → 76 +допустимых current pairs; изменён только sequence 0. На первом кадре current +points = 4,787, history-only = 7,207; pose age = 25.267 ms, oldest current +point age = 89.667 ms, skew = 64.400 ms. Все остальные 127 admissions и +increment identities совпадают. Оставшиеся 52 unavailable кадра сохраняются: +39 с просроченной pose, 43 без нового point increment, по 4 с point/pose skew +или старым newest point; множества пересекаются. + +**Это не новый Worker/GPU прогон и не новое performance evidence.** Появление +geometry на первом кадре может изменить последующие temporal/threat outputs. +Следующий bounded модельный пилот должен проверить их и повторно измерить +latency/очереди. Предыдущие результаты и hashes не переписывались. + +Сохранённый audit: `.runtime/perception-preroll-binding-20260902/metadata-audit.json`, +SHA-256 `ffe62591aaf6e84e8b373feaceac8e4c9bc624390f7609a34045c1ab1745476f`. +Исходный scenes SHA-256 `148faacef69ccf3bf74bfdbe6f81ee13286e3578eb39ef426b503e22525e618d`. +Новый binding SHA-256 `94d1b23fff2b0833a5f2f446294fb8f1fe44fdaf7ac4d37693fd726a261107df`; +runner SHA-256 `e074ce63a79906402c5a5e6125bcffa4d2eb3dbf2b28b7b09f7b75f4a0872e66`. + +Проверки: 384 focused Python tests PASS, включая 96 realtime/pilot/existing-live +checks; 62 frontend architecture/Observatory tests PASS; Ruff изменённого +Python-кода PASS. Full backend/frontend suite и production build не выполнялись: +новых визуальных или продуктовых runtime изменений в этом шаге нет. +На Mac только небольшие fixtures и metadata audit, без моделей/load tests. +Worker-сервисы не останавливались; canonical local 8000 сохранён. diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_sensor_binding.py b/experiments/perception/worker/streaming_profile_stage1/pilot_sensor_binding.py new file mode 100644 index 0000000..ef3614d --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_sensor_binding.py @@ -0,0 +1,129 @@ +"""Causal input binding with explicit preroll history and per-modality age. + +No new inference or interpolation. The caller retains all preroll points in its +bounded rolling window; only the initial current-pair selection is narrowed. +After the first camera, the original increment admission rules are unchanged. +""" + +from dataclasses import dataclass + +POSE_AGE_NS = 100_000_000 +NEWEST_POINT_AGE_NS = 100_000_000 +OLDEST_POINT_AGE_NS = 250_000_000 +POINT_POSE_SKEW_NS = 100_000_000 + + +def increment_identity(event): + return { + "sequence": event.sequence, + "host_monotonic_ns": event.time_ns, + "points": len(event.value[0]), + } + + +def milliseconds(value): + return None if value is None else value / 1e6 + + +@dataclass(frozen=True) +class SensorBinding: + increments: tuple + history_only: tuple + pose_age_ns: int | None + newest_point_age_ns: int | None + oldest_point_age_ns: int | None + binding_age_ns: int | None + pose_state: str + points_state: str + reasons: tuple[str, ...] + + @property + def available(self): + return not self.reasons + + def document(self): + return { + "schema_version": "missioncore.pilot-sensor-binding/v1", + "pose_state": self.pose_state, + "points_state": self.points_state, + "current_pair_available": self.available, + "reason_codes": list(self.reasons), + "pose_age_ms": milliseconds(self.pose_age_ns), + "newest_point_age_ms": milliseconds(self.newest_point_age_ns), + "oldest_point_age_ms": milliseconds(self.oldest_point_age_ns), + "point_pose_skew_ms": milliseconds(self.binding_age_ns), + "preroll_history_only": [increment_identity(e) for e in self.history_only], + "preroll_history_disposition": "retained-in-bounded-rolling-window", + } + + +def bind_sensors(camera_time_ns, pose, increments, *, previous_camera_time_ns=None): + increments = tuple(increments) + if previous_camera_time_ns is not None and previous_camera_time_ns >= camera_time_ns: + raise ValueError("camera binding clock must increase") + if pose is not None and pose.time_ns > camera_time_ns: + raise ValueError("future pose cannot bind to camera") + if any(e.time_ns > camera_time_ns for e in increments): + raise ValueError("future points cannot bind to camera") + if any(a.time_ns > b.time_ns for a, b in zip(increments, increments[1:], strict=False)): + raise ValueError("point binding clock moved backwards") + + history_only = () + if previous_camera_time_ns is None: + # History used to warm rolling geometry is not one current increment. + # Preserve its identity separately; do not retimestamp or silently drop it. + selected, history = [], [] + for event in increments: + is_current = ( + pose is not None + and camera_time_ns - event.time_ns <= OLDEST_POINT_AGE_NS + and abs(event.time_ns - pose.time_ns) <= POINT_POSE_SKEW_NS + ) + (selected if is_current else history).append(event) + increments, history_only = tuple(selected), tuple(history) + + pose_age = None if pose is None else camera_time_ns - pose.time_ns + newest_age = None if not increments else camera_time_ns - increments[-1].time_ns + oldest_age = None if not increments else camera_time_ns - increments[0].time_ns + skew = ( + max(abs(e.time_ns - pose.time_ns) for e in increments) + if increments and pose is not None + else None + ) + reasons = [] + if pose is None: + pose_state = "unavailable" + reasons.append("pose-unavailable") + elif pose_age > POSE_AGE_NS: + pose_state = "stale" + reasons.append("pose-too-old") + else: + pose_state = ( + "held" + if previous_camera_time_ns is not None and pose.time_ns <= previous_camera_time_ns + else "current" + ) + if not increments or not any(len(e.value[0]) for e in increments): + points_state = "unavailable" + reasons.append("point-increment-unavailable") + else: + points_state = "current" + if newest_age > NEWEST_POINT_AGE_NS: + points_state = "stale" + reasons.append("newest-points-too-old") + if oldest_age > OLDEST_POINT_AGE_NS: + points_state = "stale" + reasons.append("oldest-points-too-old") + if skew is not None and skew > POINT_POSE_SKEW_NS: + reasons.append("point-pose-skew") + return SensorBinding( + increments, + history_only, + pose_age, + newest_age, + oldest_age, + skew, + pose_state, + points_state, + tuple(reasons), + ) diff --git a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py index 1f01c8b..bb19825 100644 --- a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py +++ b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py @@ -27,6 +27,7 @@ import numpy as np from pilot_ipc import receive, send from pilot_queue import Mailbox from pilot_scheduler import GpuStage +from pilot_sensor_binding import bind_sensors, increment_identity, milliseconds from pilot_source import SensorArchive, camera_events, merged_events @@ -66,11 +67,14 @@ def produce(args, decoder, mailbox, stop, report): full_source_prepass=False, ) pose = None + previous_camera_time = None rolling = deque() fresh = [] arrivals = Counter() release_lags = [] skipped_prefix = Counter() + binding_reasons = Counter() + preroll_history_points = 0 try: for event in merged_events(archive, args.camera_index, args.frames): if stop.is_set(): @@ -100,6 +104,12 @@ def produce(args, decoder, mailbox, stop, report): send(decoder.stdin, {"op": "next"}) decoded, raw = receive(decoder.stdout) image = np.frombuffer(raw, np.uint8).reshape(600, 800, 3) + binding = bind_sensors( + event.time_ns, pose, fresh, previous_camera_time_ns=previous_camera_time + ) + fresh = binding.increments + binding_reasons.update(binding.reasons) + preroll_history_points += sum(len(e.value[0]) for e in binding.history_only) points = np.concatenate([e.value[0] for e in fresh]) if fresh else np.empty((0, 3)) rolling_points = ( np.concatenate([e.value[0] for e in rolling]) if rolling else np.empty((0, 3)) @@ -109,20 +119,6 @@ def produce(args, decoder, mailbox, stop, report): if rolling else np.empty(0, np.int64) ) - pose_age = event.time_ns - pose.time_ns if pose else None - point_age = event.time_ns - fresh[-1].time_ns if fresh else None - oldest_age = event.time_ns - fresh[0].time_ns if fresh else None - binding_age = ( - max(abs(e.time_ns - pose.time_ns) for e in fresh) if fresh and pose else None - ) - available = bool( - len(points) - and pose - and 0 <= pose_age <= 100_000_000 - and 0 <= point_age <= 100_000_000 - and oldest_age <= 250_000_000 - and binding_age <= 100_000_000 - ) bundle = { "sequence": event.sequence, "time_ns": event.time_ns, @@ -134,24 +130,18 @@ def produce(args, decoder, mailbox, stop, report): "rolling_points": rolling_points, "rolling_times": rolling_times, "pose": pose.value if pose else None, - "available": available, - "binding_age_ms": binding_age / 1e6 if binding_age is not None else None, + "available": binding.available, + "binding_age_ms": milliseconds(binding.binding_age_ns), + "sensor_binding": binding.document(), "lineage": { "camera_index_sequence": event.value["sequence"], "camera_host_monotonic_ns": event.time_ns, "pose_sequence": pose.sequence if pose else None, "pose_host_monotonic_ns": pose.time_ns if pose else None, - "point_increments": [ - { - "sequence": e.sequence, - "host_monotonic_ns": e.time_ns, - "points": len(e.value[0]), - } - for e in fresh - ], - "pose_age_ms": pose_age / 1e6 if pose_age is not None else None, - "newest_point_age_ms": point_age / 1e6 if point_age is not None else None, - "oldest_point_age_ms": oldest_age / 1e6 if oldest_age is not None else None, + "point_increments": [increment_identity(e) for e in fresh], + "pose_age_ms": milliseconds(binding.pose_age_ns), + "newest_point_age_ms": milliseconds(binding.newest_point_age_ns), + "oldest_point_age_ms": milliseconds(binding.oldest_point_age_ns), }, "source_release_lag_ms": max(0, arrived - due) / 1e6, "decode_ms": decoded["decode_ms"], @@ -161,6 +151,7 @@ def produce(args, decoder, mailbox, stop, report): bundle["payload_bytes"] = payload_size(bundle) mailbox.put(bundle) fresh = [] + previous_camera_time = event.time_ns report["last_camera_due_ns"] = due if arrivals["camera"] >= args.frames: break @@ -170,6 +161,8 @@ def produce(args, decoder, mailbox, stop, report): report.update( arrivals=dict(arrivals), skipped_prefix=dict(skipped_prefix), + sensor_binding_reasons=dict(binding_reasons), + preroll_history_only_points=preroll_history_points, release_lag_ms=distribution(release_lags), incremental_reads=archive.counters(), window_end_monotonic_ns=time.monotonic_ns(), @@ -387,6 +380,7 @@ def run(args): scene.update( sequence=bundle["sequence"], lineage=bundle["lineage"], + sensor_binding=bundle["sensor_binding"], available=bundle["available"], original_source_ns=bundle["time_ns"], ) diff --git a/tests/test_perception_joint_pilot.py b/tests/test_perception_joint_pilot.py index de4b5df..0734455 100644 --- a/tests/test_perception_joint_pilot.py +++ b/tests/test_perception_joint_pilot.py @@ -6,6 +6,7 @@ import json import threading import zipfile from pathlib import Path +from types import SimpleNamespace import numpy as np import pytest @@ -214,3 +215,120 @@ def test_completed_gpu_and_ingress_share_two_pending_slots(pilot): assert mailbox.bytes == 0 with pytest.raises(ValueError, match="accounting"): mailbox.take_completed() + + +def _sensor(pilot, channel, milliseconds, sequence=0): + return pilot("pilot_source").SensorEvent( + milliseconds * 1_000_000, channel, sequence, (np.zeros((1, 3)), None) + ) + + +def test_preroll_history_cannot_poison_the_first_fresh_pair(pilot): + pose = _sensor(pilot, "pose", 975) + points = tuple(_sensor(pilot, "points", t, i) for i, t in enumerate((590, 738, 808, 910, 980))) + binding = pilot("pilot_sensor_binding").bind_sensors(1_000_000_000, pose, points) + assert binding.available + assert binding.increments == points[-2:] + assert binding.history_only == points[:3] + assert binding.pose_age_ns == 25_000_000 + assert binding.oldest_point_age_ns == 90_000_000 + assert binding.binding_age_ns == 65_000_000 + assert binding.document()["preroll_history_only"][0] == { + "sequence": 0, + "host_monotonic_ns": 590_000_000, + "points": 1, + } + + +def test_after_preroll_admission_does_not_silently_filter_bad_increments(pilot): + pose = _sensor(pilot, "pose", 975) + points = (_sensor(pilot, "points", 730), _sensor(pilot, "points", 980, 1)) + binding = pilot("pilot_sensor_binding").bind_sensors( + 1_000_000_000, pose, points, previous_camera_time_ns=700_000_000 + ) + assert binding.increments == points and not binding.history_only + assert not binding.available + assert set(binding.reasons) == {"oldest-points-too-old", "point-pose-skew"} + + +def test_pose_reuse_keeps_age_and_empty_increment_is_not_new_geometry(pilot): + function = pilot("pilot_sensor_binding").bind_sensors + pose = _sensor(pilot, "pose", 975) + binding = function(1_000_000_000, pose, (), previous_camera_time_ns=980_000_000) + assert binding.pose_state == "held" and binding.pose_age_ns == 25_000_000 + assert binding.points_state == "unavailable" and not binding.available + assert binding.reasons == ("point-increment-unavailable",) + expired = function(1_080_000_000, pose, (), previous_camera_time_ns=1_000_000_000) + assert expired.pose_state == "stale" and "pose-too-old" in expired.reasons + missing = function(1_000_000_000, None, ()) + assert missing.pose_state == "unavailable" and not missing.available + + +def test_sensor_binding_preserves_exact_limits_and_rejects_lookahead(pilot): + function = pilot("pilot_sensor_binding").bind_sensors + pose = _sensor(pilot, "pose", 900) + points = (_sensor(pilot, "points", 800), _sensor(pilot, "points", 900, 1)) + assert function(1_000_000_000, pose, points, previous_camera_time_ns=700_000_000).available + assert not function(1_000_000_001, pose, points, previous_camera_time_ns=700_000_000).available + with pytest.raises(ValueError, match="future pose"): + function(1_000_000_000, _sensor(pilot, "pose", 1001), ()) + with pytest.raises(ValueError, match="future points"): + function(1_000_000_000, pose, (_sensor(pilot, "points", 1001),)) + with pytest.raises(ValueError, match="backwards"): + function(1_000_000_000, pose, points[::-1]) + + +def test_producer_keeps_preroll_history_for_tgs_and_publishes_sensor_ages(pilot, monkeypatch): + runner = pilot("run_joint_pilot") + events = [ + _sensor(pilot, "points", 590), + _sensor(pilot, "pose", 975), + _sensor(pilot, "points", 980, 1), + pilot("pilot_source").SensorEvent( + 1_000_000_000, "camera", 0, {"sequence": 1, "host_epoch_ns": 2_000_000_000} + ), + ] + + class Archive: + def __init__(self, path): + pass + + def counters(self): + return {} + + def close(self): + pass + + class NoWait: + def is_set(self): + return False + + def wait(self, seconds): + return False + + monkeypatch.setattr(runner, "SensorArchive", Archive) + monkeypatch.setattr(runner, "camera_events", lambda *args: iter(events[-1:])) + monkeypatch.setattr(runner, "merged_events", lambda *args: iter(events)) + raw = io.BytesIO() + pilot("pilot_ipc").send(raw, {"decode_ms": 0.0}, bytes(600 * 800 * 3)) + raw.seek(0) + decoder = SimpleNamespace(stdin=io.BytesIO(), stdout=raw) + mailbox = pilot("pilot_queue").Mailbox() + report = {} + runner.produce( + SimpleNamespace(sensor_archive="unused", camera_index="unused", frames=1), + decoder, + mailbox, + NoWait(), + report, + ) + assert mailbox.error is None + bundle = mailbox.take() + assert bundle["available"] and len(bundle["points"]) == 1 + assert len(bundle["rolling_points"]) == 2 + np.testing.assert_array_equal(bundle["rolling_times"], [590_000_000, 980_000_000]) + assert bundle["lineage"]["point_increments"][0]["host_monotonic_ns"] == 980_000_000 + assert bundle["sensor_binding"]["preroll_history_only"][0]["host_monotonic_ns"] == 590_000_000 + assert report["preroll_history_only_points"] == 1 + mailbox.release(bundle) + assert mailbox.bytes == 0 and mailbox.take() is None