From 4e04d061917804c8a7a93f87f8364570ccd32926 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 2 Sep 2026 00:59:47 +0300 Subject: [PATCH] feat(perception): define stream-first profile and qualification contracts --- ...tion-ddrnet39-rfdetr-tgs-prototype-v1.json | 175 +++++++++ .../0049-stream-first-perception-profiles.md | 236 ++++++++++++ src/k1link/perception/realtime_contract.py | 354 ++++++++++++++++++ tests/test_perception_realtime_contract.py | 278 ++++++++++++++ 4 files changed, 1043 insertions(+) create mode 100644 config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json create mode 100644 docs/adr/0049-stream-first-perception-profiles.md create mode 100644 src/k1link/perception/realtime_contract.py create mode 100644 tests/test_perception_realtime_contract.py diff --git a/config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json b/config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json new file mode 100644 index 0000000..29e7966 --- /dev/null +++ b/config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json @@ -0,0 +1,175 @@ +{ + "schema_version": "missioncore.perception-profile-candidate/v1", + "profile_id": "k1-perception-ddrnet39-rfdetr-tgs-prototype/v1", + "display_name": "K1 Perception — DDRNet-39 + RF-DETR + TGS", + "status": "stage-1-candidate-not-installed-not-qualified", + "experiment_policy": { + "bounded_lab_overload_allowed": true, + "retain_profiles_that_fail_current_hardware_realtime": true, + "realtime_failure_blocks_experimental_packaging": false, + "qualification_scope": ["profile-version", "effective-config", "hardware", "source", "transport"], + "preserve_original_source_clock": true, + "report_drops_and_staleness": true, + "overload_is_realtime_pass": false, + "future_onboard_placement_requires_new_measurement": true + }, + "packaging": { + "image": "ndc-k1-perception-ddrnet39-rfdetr-tgs:prototype-v1", + "image_sha256": null, + "single_container": true, + "external_model_server_required": false, + "host_model_cache_required": false, + "runtime_model_download_allowed": false, + "model_environment_isolation": "supervised-processes-one-gpu-scheduler", + "common_interpreter_required": false + }, + "hardware_baseline": { + "gpu": "NVIDIA GeForce RTX 4090", + "active_profiles_per_worker": 1, + "sources_per_worker": 1, + "gpu_inference_policy": "serialized", + "other_gpu_workloads_allowed": false + }, + "cpu_execution_policy": { + "OPENBLAS_NUM_THREADS": 1, + "OMP_NUM_THREADS": 1, + "MKL_NUM_THREADS": 1, + "reason": "bounded-single-thread-numeric-libraries-before-whole-graph-qualification" + }, + "input_contract": { + "schema_version": "missioncore.perception-stream-start/v1", + "channels": ["camera", "point-cloud", "pose"], + "camera": { + "width": 800, + "height": 600, + "projection": "KB4", + "decoded_pixel_format": "bgr8", + "calibration_sha256": "05f3ad9b38b3a4fc95388a8ec83da83c745e217709e51787b3d5aad0969f6fa9", + "valid_fov_sha256": "a40cee06b7c6f69b6a09a11563dcfd237f3de833b1ccd31459e66692e528ba63" + }, + "point_cloud": { + "representation": "vendor-registered-current-increment", + "coordinate_frame": "map", + "units": "metres", + "scalar_type": "float32-little-endian", + "raw_fields_retained": true + }, + "pose": {"parent_frame": "map", "child_frame": "sensor"}, + "clock_mapping_required": true, + "source_end_or_duration_required": false, + "full_source_transfer_required": false, + "full_source_hash_pass_required": false, + "full_decode_or_png_cache_required": false, + "previous_perception_artifacts_allowed": false + }, + "stream_policy": { + "replay_speed": 1.0, + "release_clock": "original-source-timestamps", + "inference_stride": 1, + "sampling_status": "strict-every-frame-initial-baseline-no-silent-multirate", + "first_result_requires_eof": false, + "lookahead_allowed": false, + "reconnect": "new-epoch-drop-old-backlog-reacquire-codec-keyframe", + "overflow": "bounded-drop-with-ledger-degraded-and-failed-runtime-gate", + "archive_export_on_critical_path": false, + "transport_candidate": "grpc-bidirectional-protobuf-binary-payloads", + "transport_qualified": false + }, + "components": { + "segmentation": { + "model": "ddrnet_39-goose-fine-64", + "checkpoint_sha256": "b99c2838051bcd7b092fd3970aa62a77d5c0bbb809c9b9afb2ff4b0ebdaa4ee6", + "precision": "existing-fp32-baseline", + "preprocessing": "center-600-square-to-512-rgb-tensor-0-1", + "outside_crop": "undefined-not-free", + "class_count": 64, + "runtime": "python-3.9-pytorch-1.13.1-cu117-super-gradients-3.2.0" + }, + "objects": { + "model": "rf_detr_large_native_kb4", + "checkpoint_sha256": "0f4e20e19a99c0f8a62b5685f57f6c8b5c371c59081feda6752a0561a79ccf38", + "engine_sha256": "b8a40b3580edff001ec9680de68707242294ff590ab296000fae371f1083f695", + "runtime": "tensorrt-11-fp16-native-uint8", + "minimum_score": 0.25, + "minimum_box_area_pixels": 64, + "maximum_box_area_fraction": 0.5, + "minimum_valid_fov_fraction": 0.5, + "require_center_inside_valid_fov": true, + "required_existing_classes": ["person", "cat", "dog"], + "fine_static_object_classifier_required": false, + "filter_quality_status": "existing-unchanged-near-large-and-small-object-risk-unqualified" + }, + "geometry": { + "online_surface": "K1LocalSurfaceShadowEstimator", + "prepared_local_surface_allowed": false, + "generic_static_class": "static.unknown", + "detected_object_range": "median-camera-z-of-owned-current-points/v1", + "geometry_only_range": "nearest-euclidean-sensor-distance/v1", + "vehicle_clearance_claim": false, + "missing_support": "unavailable-not-zero-not-free" + }, + "motion": { + "reuse": "BoundedSpatialTemporalProvider+ClassIndependentMotionEstimator", + "semantic_class_implies_motion": false + }, + "costmap": { + "algorithm": "TRAVEL-TGS", + "revision": "95dc2fbd66a343efd9060c45a5711b6307a950a4", + "execution": "cpu", + "rolling_history_seconds": 1.0, + "states": ["UNOBSERVED", "GROUND_SUPPORT", "NONGROUND_OCCUPIED", "UNKNOWN_REJECTED"], + "policy_input": true + }, + "policy": { + "scope": "rural-prototype-advisory-only", + "hard_surface": "ALLOW-candidate-subject-to-geometry-and-freshness", + "fine_road_sidewalk_bikeway_distinction_required": false, + "effective_material_rules_required": true, + "unknown_or_occupied_overrides_material_allowance": true, + "sensor_to_vehicle_mount": "simulation-only-until-measured" + } + }, + "reference_files": [ + {"path": "config/perception/rf-detr-large-native-kb4-risk-shadow-v0.json", "sha256": "dbf4da5dbad6c3c22b1280b46ffcad81719bd183c81c263a4859847d829019b6"}, + {"path": "config/perception/lab-v1-goose-vegetation-benchmark-v1.json", "sha256": "96a427a8baae387b827ec9c0bf7ca42e3fb9114b8fa9a8671bbc9d10877670b9"}, + {"path": "config/perception/lab-v1-vegetation-provider-label-map-v1.json", "sha256": "f2b69046b6a740fd9532d2d88e7fabae7c20fb662f783c9502adc9026406f352"}, + {"path": "config/perception/lab-v1-vegetation-mission-policy-v1.json", "sha256": "b75c4ac841d7b4bcc57f7a9c8417ca2317d8ecfa499e72a9af8a8591a2ec0d35"}, + {"path": "config/perception/m49-tgs-full-shadow-v1.json", "sha256": "c2e07010aaee78259d36c057962d6bfb885349251ff7356d867e5813e632881c"}, + {"path": "config/perception/m4-geometry-association-v1.json", "sha256": "cc666c9389a5e221957faddec89584709b66918d14abaf646f1832e001421999"} + ], + "reference_usage": "reuse-model-parameters-and-algorithms-only-not-recording-ids-or-precomputed-results", + "realtime_contract": { + "schema_version": "missioncore.perception-realtime-contract/v1", + "budget_status": "engineering-targets-worker-pilot-measured-not-whole-path-qualified-not-safety-approved", + "budgets": { + "maximum_start_metadata_bytes": 65536, + "maximum_chunk_bytes": 1048576, + "maximum_inflight_bytes": 16777216, + "maximum_pending_camera_frames": 2, + "maximum_codec_preroll_ms": 500.0, + "maximum_pose_age_ms": 100.0, + "maximum_clock_uncertainty_ms": 5.0, + "maximum_release_lag_ms": 25.0, + "maximum_output_age_p95_ms": 125.0, + "maximum_output_age_p99_ms": 125.0, + "maximum_layer_age_ms": 250.0, + "maximum_first_result_ms": 1000.0, + "maximum_warmup_seconds": 120.0, + "maximum_stop_seconds": 5.0, + "maximum_vram_mib": 22000, + "maximum_rss_mib": 8192, + "maximum_backlog_growth_ms": 25.0 + }, + "budget_origin": "125ms-historical-envelope-retained-as-target-now-applied-to-whole-path; other-limits-explicit-initial-engineering-guards", + "capacity_drop_count_max": 0, + "unaccounted_input_count_max": 0, + "full_input_transfer_before_first_result_allowed": false, + "independent_quality_and_physical_live_status": "pending" + }, + "authority": { + "commands_enabled": false, + "actuation_allowed": false, + "navigation_or_safety_accepted": false, + "production_accepted": false + } +} diff --git a/docs/adr/0049-stream-first-perception-profiles.md b/docs/adr/0049-stream-first-perception-profiles.md new file mode 100644 index 0000000..e4a52c9 --- /dev/null +++ b/docs/adr/0049-stream-first-perception-profiles.md @@ -0,0 +1,236 @@ +# ADR 0049 — Stream-first full perception profiles + +Date: 2026-09-01. Status: stage-1 contract candidate; not a runtime cutover. +Execution order remains the four stages in +[the ExecPlan](../OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md). + +## Decision and owner intent + +Observatory LABs test the real-time behavior of a complete perception profile, +not just the quality of an offline export. A recording replaces the physical +source, preserving original release timing at 1×. It must not enable a different +batch implementation. The first profile contains DDRNet-39 GOOSE, RF-DETR native, +online LiDAR geometry/ranges, temporal/motion, CPU TRAVEL TGS/costmap and advisory +policy. One dedicated RTX 4090 owns one active full profile. + +The first rural prototype accepts existing coarse `hard_surface`, including +asphalt, sidewalk, bikeway and cobble. Existing `static_obstacle`/`static.unknown` +is sufficient. No new bollard classifier, extra segmenter, road/sidewalk taxonomy +or annotation campaign precedes this prototype. Other material rules remain an +explicit effective mission configuration; rural context alone is not permission +for all vegetation or soil. Geometry/unknown/freshness overrides material allowance. + +EoMT is preserved as a separate future profile. It is not loaded with DDRNet. +RF-DETR and DDRNet are two required components of the same first profile: one +supervisor and serialized GPU scheduling, not competing profile jobs. + +## CURRENT → TARGET / ownership + +| Boundary | Current evidence | Decision / stage | +| --- | --- | --- | +| Portable definitions | `portable_run_definitions.py` owns pins, resources and admission | KEEP identity/sealing; EXTEND stream capabilities and qualification in stage 3 | +| Source requirements | Recorded-only requirements include seekability, one media epoch and init digest | REPLACE as live admission criteria with channel/format/calibration/clock requirements; keep init digest as per-stream integrity | +| Job queue | `recorded_jobs.py` has durable claims but global single-active-job scope | KEEP durable queue; EXTEND worker-scoped fencing, not unbounded GPU concurrency | +| Worker agent | Blocking `execute(job)` and final publication | EXTEND incremental lifecycle and cancellation; no per-LAB agent | +| Runtime | `portable_worker_runtime.py:735` materializes before execution | NEW stream runtime in stage 2; never call this materializer in the real-time path | +| Source transport | Whole camera archive download/hash/extraction before use | REPLACE startup barrier with bounded metadata, chunks and incremental integrity | +| Installed LAB V1 | prepare → complete EoMT → complete DDRNet → assemble | Legacy evidence only; not the new profile and not real-time qualified | +| Camera/source replay | `recorded_source.py` has useful pacing and bounded prefetch, but source-specific timeline and paths | REUSE mechanisms; generic unknown-duration source adapter, no route-sized prefetch | +| Existing live ingress | `LivePerceptionIngress`, `LiveIngressEvent.wire_bytes`, `LiveSensorSynchronizer` already provide raw-first modality queues, hashes, epochs and bounded binding | REUSE, not a second acquisition system; extend worker/profile fencing and aggregate byte limits; map existing envelopes to the selected transport | +| Local surface | `RecordedGeometryStore` reads prepared NPZ | REUSE `K1LocalSurfaceShadowEstimator.process` and `lidar_local_surface_geometry`; feed current points/pose | +| Geometry/ranges | Current point association, exclusive ownership and geometry-only clusters exist | REUSE algorithms; remove recorded-store dependency and expose estimator/frame identity | +| Temporal/motion | `BoundedSpatialTemporalProvider`, `ClassIndependentMotionEstimator` | REUSE with current-run inputs and reset on epoch/lease change | +| TGS | Core algorithm reusable; old runners read a full schedule, some require exactly 4489 frames | REUSE TGS parameters/core; REPLACE schedule/file CLI with incremental input | +| Fusion | M49 TGS can be a neighboring shadow with `tgs_modifies_reference_graph_state=false` | TGS and semantics must influence the current advisory scene, not only viewer composition | +| Publication | Verified immutable result/publication/recovery exist | KEEP asynchronous final sealing; live results do not await it | +| Frontend | Setup/job/result surfaces exist, historical M49 viewer links other results | EXTEND common renderer capabilities with current-run layer references and freshness in stage 3 | +| Recording archive | Source identities and raw evidence must be retained | KEEP; create metadata/index/integrity during acquisition or incrementally, not via preflight full scan | + +No changes to the active registry, old queue, source endpoint, UI or canonical +local server are made by this ADR or the stage-1 validators. + +## Versioned handshake and observations + +`missioncore.perception-stream-start/v1` is represented by `StreamStart` in +`src/k1link/perception/realtime_contract.py`. It contains run/source/worker/epoch, +lease generation, profile/image/effective-config/calibration identities, clock +domain, input mode and channel declarations. The first profile requires camera, +point-cloud and pose. Temporary absence of a sample is an explicit modality +outcome, not removal of a required capability. + +It deliberately has no total source bytes, frame count, total duration, whole +archive, command, environment, complete member inventory or required EOF. The +initial metadata budget is 64 KiB. Large constant model assets and FOV masks +belong inside the image; calibration references resolve to bounded metadata. + +The stage-2 wire mapping must carry: + +- Run/epoch/lease binding on every message; source/channel sequence, capture + timestamp and clock domain; independently verifiable payload length/hash. +- Camera codec initialization and then incremental encoded access units, or + bounded fragments thereof. Native decoded representation is BGR8 800×600 KB4. + Do not send an entire fMP4 epoch as one observation or turn all frames into PNGs. +- Current vendor-registered map point increments with point IDs, coordinate + frame, units, scalar layout and retained raw attributes; map→sensor pose and + calibration identity. The adapter owns vendor decoding; the profile is not a + new K1 protocol implementation and cannot command the scanner. +- Explicit unavailable/gap/end/cancel messages. End is a termination event, + never a prerequisite for first inference. Final recording inventory/hash can + be sealed after consumption; per-chunk integrity is verified before use. +- Integer nanoseconds use protobuf integer fields; any JSON projection uses + decimal strings for 64-bit times rather than lossy JavaScript numbers. + +Maximum payload fragment: 1 MiB; total application inflight bytes: 16 MiB; +pending decoded camera frames: 2. Fragment reassembly is included in that budget, +with a deadline and incomplete-chunk accounting. Decoded image/tensor, cloud, +costmap and archive sinks need their own measured bounds inside the RSS budget. +Transport flow control alone is not a memory bound or a freshness policy. + +gRPC bidirectional streaming is the first transport candidate, not an already +measured implementation. HTTP control-plane operations remain. The browser does +not connect directly to model RPCs. One authenticated data-plane endpoint feeds +the supervised profile; model processes have no arbitrary external network. +Protocol/codec selection must be measured on actual payload sizes in stage 2; +changing an archive POST to gRPC without changing execution is not acceptance. +Existing `missioncore.live-perception-wire/v1` and result wire v2 are migration +inputs: reuse their semantics and raw-first producer hooks. Their modality count +caps alone permit more than the proposed 16 MiB aggregate inflight budget; an +explicit byte bound is required. No duplicate new scanner feed is introduced. + +## Time and lifecycle + +Open → validate bounded metadata/claim → warm models → Ready → start the replay +clock → observations and incremental scenes → stop/drain bounded state → close +and asynchronously seal the immutable receipt. A live source already producing +data does not queue its entire warmup history: readiness begins at a current +decodable keyframe, with the skipped interval recorded. + +Replay release uses original source intervals, not a convenient fixed 12 Hz and +not the processing speed. A slow consumer must not stretch source time to make +inference look real-time. Overload may drop bounded work to preserve liveness, +but the dropped observations remain in the ledger and fail the strict initial +performance target. Initial baseline is every camera frame, stride 1. Choosing +multirate later changes the explicit profile contract; a retained mask is not a +new inference and cannot receive a fresh capture timestamp. + +UTC identifies the run; monotonic clocks measure local durations. Cross-host +age requires a clock mapping and its uncertainty. No direct subtraction of +unrelated host monotonic clocks. The common source timeline is separate from +worker wall-clock time. Causality is bounded by the released-observation +watermark at decision time, not by whatever later samples exist in the file. +Each layer retains its own channel/sequence/time; camera anchors and pose/cloud +association times remain separately visible. Future file look-ahead and offline +nearest-pose lookup across unreleased data are forbidden. + +Lease expiry, cancellation and epoch changes fence the previous producer. No +new profile may take the GPU until old processes and work have stopped. A +reconnect creates a new epoch, discards stale backlog, resets temporal/rolling +state and reacquires codec initialization/keyframe. It never resends a stale +motor action; this profile has no motor actions at all. + +## Outputs and interpretation + +Every scene accounts for segmentation, objects, geometry, motion, costmap and +policy exactly once, with current/held/stale/unavailable state and current-run +payload identity. Existing source/object/obstacle/temporal/map/threat contracts +remain the domain vocabulary. `LayerEvidence` adds freshness/completeness checks, +not a second semantic ontology. In stage 2, layer manifests also carry their full +input reference sets; a single anchor does not replace camera/cloud/pose lineage. + +DDRNet currently crops the center 600×600 to 512×512; outside that ROI is undefined, +not hard_surface/free. RF-DETR accepts the full 800×600 native raster with its +pinned valid-FOV handling. Their coordinates must be related explicitly. + +RF-DETR's existing filter emits person/cat/dog, but minimum box area 64 px, +maximum box fraction 0.5, FOV fraction 0.5 and required valid center create known +small-animal/close-large-object edge cases. They are retained and documented, +not silently retuned or presented as proven field quality. + +Detected-object range is the median camera-Z of owned current support points; +geometry-only range is nearest Euclidean distance from the sensor. Neither is +automatically clearance from the physical vehicle body. Missing/ambiguous +support produces unavailable range, not zero/infinity. Generic static obstacles +survive absence of a detector class name. + +TGS keeps ground/occupied/rejected/unobserved separate. Ground support alone is +not a traversability or actuation decision. Policy can produce an advisory +allowed-candidate/high-cost/blocked/unknown outcome with explanation. No planner, +physical motor command, autonomous driving acceptance or mission configurator +implementation is introduced. + +## Package/dependency decision + +The candidate manifest is +`config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json`. +It is not installed into the active portable registry and has no invented image +digest. Model/checkpoint/config identities are pinned; source session IDs and old +derived output paths in reference experiments are not inherited requirements. + +Select process/environment isolation inside one future image: Python 3.12 +supervisor and geometry, RF-DETR TensorRT 11 runtime, existing Python 3.9 / Torch +1.13.1 cu117 / super-gradients 3.2.0 environment, and CPU C++ TGS. Exchange bounded +shared-memory/IPC payload references; one supervisor serializes GPU work. This +avoids a forced dependency upgrade or checkpoint conversion in stage 1. Image +assembly of the final standalone package and the full schedule pilot remain +explicit implementation evidence. A temporary common-base image passed separate +DDRNet, TensorRT, Python 3.12 geometry and C++ TGS execution probes; DDRNet masks +matched on 64 frames. This proves bounded ABI/execution compatibility in that +image, not jointly resident models, the supervisor/IPC or a complete profile. +The probes still mount pinned assets explicitly; they are not standalone proof. + +The temporary Triton base lacks Python grpc/protobuf, cv2 and TensorRT bindings; +native trtexec/server are present. The actual supervisor/transport environment +must explicitly include and validate its dependencies. Numeric-library thread +limits are pinned to one after the bounded synthetic local-surface comparison +(mean 198.0 ms default versus 63.2 ms with limits); algorithm thresholds were not +changed. Real-cloud timing, output parity and whole-graph scheduling remain gates. + +The initial DDRNet image-only probe failed before inference: the installed image +does not contain the checkpoint at its logical asset path. The bounded baseline +therefore uses an explicit read-only pinned checkpoint and runner mount. This is +valid component measurement, but fails the target's standalone packaging claim. +Super-gradients also needs a writable log directory; a bounded temporary log +mount was required. The new image must declare scratch areas explicitly without +requiring a writable host home, code checkout or implicit weight cache. + +## Preregistered engineering acceptance + +Owner clarification, 2026-09-02 MSK: bounded laboratory overload is an admissible +experimental outcome. Keep useful profiles that fail the current 4090 target; +do not block reusable runtime/packaging work solely on that performance result. +Functional execution, experimental availability, quality and real-time +qualification are separate dimensions. Qualification belongs to a measured +profile/config/hardware/source/transport combination, not permanently to a +model name. More powerful Worker/onboard placement is a future retest target, +not a claim that existing latency or internal IPC disappears automatically. +The original budgets and failed measurements below remain unchanged. No slowed +source clock, hidden drops, stale-as-current evidence or actuation is authorized. + +The manifest freezes an initial engineering candidate: p95/p99 full output age +≤125 ms, required-layer age ≤250 ms, release lag ≤25 ms, first incremental result +≤1 s after source admission, warmup ≤120 s, stop ≤5 s, VRAM ≤22000 MiB, RSS ≤8192 +MiB. First/last-window backlog growth ≤25 ms and no capacity drops, failures, +expired selected frames or unaccounted observations. These are prototype targets, +not physical braking/safety limits or claims of achieved performance. + +`ReplayMeasurements`/`realtime_failures` reject full-source preload, EOF-only +results, slowed replay, incomplete accounting, hidden missing layers and budget +violations. Declared source gaps must be verified from the input ledger; they +produce explicit degraded scenes and are not counted as fully fresh perception. +At least one fully evaluable scene is required. A component FPS figure cannot +populate this whole-path receipt. A quality comparison or successful export is +not a substitute for the real-time gate. + +Every observation receives a terminal ledger outcome. Track ingress/release, +selection, decode, inference, completion, emission, drop/expiry/failure and source +gaps separately. Store startup separately from steady state; record payload bytes, +queue high-water marks, clock error, RSS/VRAM, image/weights/config/source/equipment +identities. Export and UI encoding cannot block inference through an unlimited +queue. A receiver-side render/receipt timestamp, not just Worker completion, ends +the end-to-end interval. + +## Evidence and remaining stage-1 work + +See `experiments/perception/PERCEPTION_STREAM_STAGE1_2026-09-01.md` for measured +component results, bounded validation, Worker maintenance and remaining gates. +Neither this ADR nor passing synthetic contract tests marks the new runtime ready. diff --git a/src/k1link/perception/realtime_contract.py b/src/k1link/perception/realtime_contract.py new file mode 100644 index 0000000..ad1abe8 --- /dev/null +++ b/src/k1link/perception/realtime_contract.py @@ -0,0 +1,354 @@ +"""Stage-1, transport-neutral contract for stream-first perception. + +These executable invariants do not start models, replace the legacy queue, or +claim that a stream transport exists. Replay and live must implement the same +contract in stage 2. Existing perception objects remain the domain vocabulary. +""" + +from __future__ import annotations + +import math +import re +from dataclasses import asdict, dataclass +from typing import Final, cast + +STREAM_START_SCHEMA: Final = "missioncore.perception-stream-start/v1" +REALTIME_CONTRACT_SCHEMA: Final = "missioncore.perception-realtime-contract/v1" +REQUIRED_CHANNELS: Final = ("camera", "point-cloud", "pose") +REQUIRED_LAYERS: Final = ("segmentation", "objects", "geometry", "motion", "costmap", "policy") +_ID: Final = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$") +_SHA: Final = re.compile(r"^[a-f0-9]{64}$") + + +class RealtimeContractError(ValueError): + """The caller attempted an ambiguous, unbounded or non-causal contract.""" + + +def _integer(value: object, name: str, *, minimum: int = 0) -> int: + if type(value) is not int or value < minimum: + raise RealtimeContractError(f"{name} must be an integer >= {minimum}") + return value + + +def _number(value: object, name: str, *, minimum: float = 0.0) -> float: + if type(value) not in (int, float): + raise RealtimeContractError(f"{name} must be a finite number") + number = float(cast(int | float, value)) + if not math.isfinite(number) or number < minimum: + raise RealtimeContractError(f"{name} must be finite and >= {minimum}") + return number + + +def _identifier(value: object, name: str) -> None: + if not isinstance(value, str) or _ID.fullmatch(value) is None: + raise RealtimeContractError(f"{name} is invalid") + + +def _digest(value: object, name: str) -> None: + if not isinstance(value, str) or _SHA.fullmatch(value) is None: + raise RealtimeContractError(f"{name} must be a SHA-256 identity") + + +@dataclass(frozen=True, slots=True) +class StreamStart: + """Bounded handshake; deliberately no source duration, EOF or file inventory. + + The epoch is reset on reconnect. Source hashes belong to chunk admission + and the asynchronous final receipt, not a full-file startup verification. + Calibration/config references resolve to bounded, authenticated metadata. + """ + + run_id: str + source_id: str + worker_id: str + epoch_id: str + lease_generation: int + profile_sha256: str + image_sha256: str + effective_config_sha256: str + calibration_sha256: str + clock_domain_id: str + input_mode: str + channels: tuple[str, ...] = REQUIRED_CHANNELS + + def __post_init__(self) -> None: + for name in ("run_id", "source_id", "worker_id", "epoch_id", "clock_domain_id"): + _identifier(getattr(self, name), name) + _integer(self.lease_generation, "lease_generation", minimum=1) + for name in ( + "profile_sha256", + "image_sha256", + "effective_config_sha256", + "calibration_sha256", + ): + _digest(getattr(self, name), name) + if self.input_mode not in ("recorded-source-paced", "live"): + raise RealtimeContractError("only source-paced replay and live are admitted") + if self.channels != REQUIRED_CHANNELS: + raise RealtimeContractError("the full profile requires camera, point-cloud and pose") + + def to_dict(self) -> dict[str, object]: + value = asdict(self) + value["schema_version"] = STREAM_START_SCHEMA + value["channels"] = list(self.channels) + return value + + @classmethod + def from_dict(cls, value: object) -> StreamStart: + if not isinstance(value, dict): + raise RealtimeContractError("stream start must be an object") + fields = set(cls.__dataclass_fields__) + if set(value) != fields | {"schema_version"}: + raise RealtimeContractError("stream start fields changed; full-source fields forbidden") + if value["schema_version"] != STREAM_START_SCHEMA: + raise RealtimeContractError("stream start schema changed") + if not isinstance(value["channels"], list): + raise RealtimeContractError("channels must be an array") + arguments = {name: value[name] for name in fields} + arguments["channels"] = tuple(value["channels"]) + return cls(**arguments) + + +@dataclass(frozen=True, slots=True) +class RealtimeBudgets: + """Preregistered engineering limits, never vehicle-safety thresholds.""" + + maximum_start_metadata_bytes: int + maximum_chunk_bytes: int + maximum_inflight_bytes: int + maximum_pending_camera_frames: int + maximum_codec_preroll_ms: float + maximum_pose_age_ms: float + maximum_clock_uncertainty_ms: float + maximum_release_lag_ms: float + maximum_output_age_p95_ms: float + maximum_output_age_p99_ms: float + maximum_layer_age_ms: float + maximum_first_result_ms: float + maximum_warmup_seconds: float + maximum_stop_seconds: float + maximum_vram_mib: int + maximum_rss_mib: int + maximum_backlog_growth_ms: float + + def __post_init__(self) -> None: + for name in ( + "maximum_start_metadata_bytes", + "maximum_chunk_bytes", + "maximum_inflight_bytes", + "maximum_pending_camera_frames", + "maximum_vram_mib", + "maximum_rss_mib", + ): + _integer(getattr(self, name), name, minimum=1) + for name in set(self.__dataclass_fields__) - { + "maximum_start_metadata_bytes", + "maximum_chunk_bytes", + "maximum_inflight_bytes", + "maximum_pending_camera_frames", + "maximum_vram_mib", + "maximum_rss_mib", + }: + _number(getattr(self, name), name, minimum=0.001) + if self.maximum_chunk_bytes > self.maximum_inflight_bytes: + raise RealtimeContractError("one chunk exceeds the inflight byte budget") + if not ( + self.maximum_output_age_p95_ms + <= self.maximum_output_age_p99_ms + <= self.maximum_layer_age_ms + ): + raise RealtimeContractError("output percentile and freshness budgets disagree") + + @classmethod + def from_dict(cls, value: object) -> RealtimeBudgets: + if not isinstance(value, dict) or set(value) != set(cls.__dataclass_fields__): + raise RealtimeContractError("realtime budget fields changed") + return cls(**value) + + +@dataclass(frozen=True, slots=True) +class LayerEvidence: + """Freshness metadata wrapping existing domain payloads, not a new ontology. + + source_time_ns is mapped into the run's common source timeline. It is not + a worker clock. A retained layer must keep its original timestamp/sequence. + Network end-to-end age is measured separately, with clock uncertainty. + """ + + layer: str + epoch_id: str + source_sequence: int | None + source_time_ns: int | None + state: str + payload_sha256: str | None + + def __post_init__(self) -> None: + if self.layer not in REQUIRED_LAYERS: + raise RealtimeContractError("unknown full-profile layer") + _identifier(self.epoch_id, "epoch_id") + if self.state not in ("current", "held", "stale", "unavailable"): + raise RealtimeContractError("unknown evidence currentness") + if self.state == "unavailable": + if any( + x is not None + for x in (self.source_sequence, self.source_time_ns, self.payload_sha256) + ): + raise RealtimeContractError("unavailable evidence cannot claim a payload") + else: + _integer(self.source_sequence, "source_sequence") + _integer(self.source_time_ns, "source_time_ns") + _digest(self.payload_sha256, "payload_sha256") + + +def validate_scene_layers( + layers: tuple[LayerEvidence, ...], + *, + epoch_id: str, + source_time_ns: int, + maximum_layer_age_ms: float, +) -> None: + """Reject missing layers, cross-epoch reuse, look-ahead and relabelled stale data. + + Explicit unavailable/stale output remains valid evidence, not permission to + move. Deciding hard_surface/occupied precedence belongs to the existing + policy evaluator, whose results cannot grant actuation. + """ + _identifier(epoch_id, "epoch_id") + _integer(source_time_ns, "source_time_ns") + _number(maximum_layer_age_ms, "maximum_layer_age_ms", minimum=0.001) + if len(layers) != len(REQUIRED_LAYERS) or {x.layer for x in layers} != set(REQUIRED_LAYERS): + raise RealtimeContractError("a scene must account for every required layer exactly once") + for layer in layers: + if layer.epoch_id != epoch_id: + raise RealtimeContractError("layer belongs to another stream epoch") + if layer.source_time_ns is None: + continue + age_ns = source_time_ns - layer.source_time_ns + if age_ns < 0: + raise RealtimeContractError("future observations are not causal input") + if age_ns > maximum_layer_age_ms * 1_000_000 and layer.state in ("current", "held"): + raise RealtimeContractError("expired evidence cannot be current or held") + + +@dataclass(frozen=True, slots=True) +class ReplayMeasurements: + """Measured whole-path receipt; component timings alone cannot populate it. + + All age figures include source release, transfer, decode, graph processing + and delivery to the application. The slowest required layer determines + full-scene age. Qualification is bound to identities by the caller. + """ + + source_frames: int + declared_source_gap_frames: int + terminal_frames: int + fresh_complete_scenes: int + capacity_drops: int + expired_frames: int + failed_frames: int + unexpected_unavailable_layers: int + full_source_preloaded: bool + results_before_end_of_source: bool + replay_speed: float + maximum_release_lag_ms: float + output_age_p95_ms: float + output_age_p99_ms: float + maximum_layer_age_ms: float + first_result_ms: float + warmup_seconds: float + stop_seconds: float + peak_inflight_bytes: int + peak_pending_camera_frames: int + peak_vram_mib: int + peak_rss_mib: int + clock_uncertainty_ms: float + backlog_growth_ms: float + computed_layers: tuple[str, ...] + + def __post_init__(self) -> None: + integer_fields = { + "source_frames", + "declared_source_gap_frames", + "terminal_frames", + "fresh_complete_scenes", + "capacity_drops", + "expired_frames", + "failed_frames", + "unexpected_unavailable_layers", + "peak_inflight_bytes", + "peak_pending_camera_frames", + "peak_vram_mib", + "peak_rss_mib", + } + boolean_fields = {"full_source_preloaded", "results_before_end_of_source"} + for name in integer_fields: + _integer(getattr(self, name), name) + for name in boolean_fields: + if type(getattr(self, name)) is not bool: + raise RealtimeContractError(f"{name} must be boolean") + for name in ( + set(self.__dataclass_fields__) - integer_fields - boolean_fields - {"computed_layers"} + ): + _number(getattr(self, name), name) + if ( + self.source_frames == 0 + or self.declared_source_gap_frames > self.source_frames + or self.fresh_complete_scenes > self.terminal_frames + ): + raise RealtimeContractError("measurement frame counts are invalid") + if self.output_age_p95_ms > self.output_age_p99_ms: + raise RealtimeContractError("measurement percentiles are invalid") + if not isinstance(self.computed_layers, tuple) or any( + not isinstance(layer, str) for layer in self.computed_layers + ): + raise RealtimeContractError("computed layers must be an immutable tuple of names") + if len(set(self.computed_layers)) != len(self.computed_layers): + raise RealtimeContractError("computed layers are duplicated") + + +def realtime_failures(value: ReplayMeasurements, limits: RealtimeBudgets) -> tuple[str, ...]: + """Return every failed strict replay gate, not a product-ready flag. + + First baseline is every source camera frame, no silent stride. A later + multirate profile needs its own explicit contract and measured qualification. + Source gaps are bound to the source ledger, never inferred from model + failures. They must still produce explicit degraded scenes and terminal + accounting; no complete fresh-scene claim is allowed for those frames. + """ + failures: list[str] = [] + for failed, reason in ( + (value.full_source_preloaded, "full-source-preload"), + (not value.results_before_end_of_source, "no-incremental-result"), + (value.replay_speed != 1.0, "source-clock-not-1x"), + (value.terminal_frames != value.source_frames, "incomplete-input-accounting"), + ( + value.fresh_complete_scenes != value.source_frames - value.declared_source_gap_frames, + "incomplete-fresh-scenes", + ), + (value.fresh_complete_scenes == 0, "no-evaluable-scene"), + (value.capacity_drops != 0, "capacity-drops"), + (value.expired_frames != 0, "expired-frames"), + (value.failed_frames != 0, "failed-frames"), + (value.unexpected_unavailable_layers != 0, "required-layer-unavailable"), + (set(value.computed_layers) != set(REQUIRED_LAYERS), "full-profile-not-computed"), + ): + if failed: + failures.append(reason) + for measurement, budget in ( + ("maximum_release_lag_ms", "maximum_release_lag_ms"), + ("output_age_p95_ms", "maximum_output_age_p95_ms"), + ("output_age_p99_ms", "maximum_output_age_p99_ms"), + ("maximum_layer_age_ms", "maximum_layer_age_ms"), + ("first_result_ms", "maximum_first_result_ms"), + ("warmup_seconds", "maximum_warmup_seconds"), + ("stop_seconds", "maximum_stop_seconds"), + ("peak_inflight_bytes", "maximum_inflight_bytes"), + ("peak_pending_camera_frames", "maximum_pending_camera_frames"), + ("peak_vram_mib", "maximum_vram_mib"), + ("peak_rss_mib", "maximum_rss_mib"), + ("clock_uncertainty_ms", "maximum_clock_uncertainty_ms"), + ("backlog_growth_ms", "maximum_backlog_growth_ms"), + ): + if getattr(value, measurement) > getattr(limits, budget): + failures.append(measurement) + return tuple(failures) diff --git a/tests/test_perception_realtime_contract.py b/tests/test_perception_realtime_contract.py new file mode 100644 index 0000000..316c8ca --- /dev/null +++ b/tests/test_perception_realtime_contract.py @@ -0,0 +1,278 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import asdict, replace +from pathlib import Path + +import pytest + +from k1link.compute.live_perception import LivePerceptionIngress +from k1link.perception.realtime_contract import ( + REQUIRED_LAYERS, + LayerEvidence, + RealtimeBudgets, + RealtimeContractError, + ReplayMeasurements, + StreamStart, + realtime_failures, + validate_scene_layers, +) + +ROOT = Path(__file__).resolve().parents[1] +PROFILE = ROOT / "config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json" + + +def _profile() -> dict: + return json.loads(PROFILE.read_text()) + + +def _budgets() -> RealtimeBudgets: + return RealtimeBudgets.from_dict(_profile()["realtime_contract"]["budgets"]) + + +def _start() -> StreamStart: + return StreamStart( + run_id="run-1", + source_id="source-a", + worker_id="worker-006", + epoch_id="epoch-1", + lease_generation=1, + profile_sha256="a" * 64, + image_sha256="b" * 64, + effective_config_sha256="c" * 64, + calibration_sha256="d" * 64, + clock_domain_id="source-monotonic", + input_mode="recorded-source-paced", + ) + + +def _measurements() -> ReplayMeasurements: + return ReplayMeasurements( + source_frames=100, + declared_source_gap_frames=0, + terminal_frames=100, + fresh_complete_scenes=100, + capacity_drops=0, + expired_frames=0, + failed_frames=0, + unexpected_unavailable_layers=0, + full_source_preloaded=False, + results_before_end_of_source=True, + replay_speed=1.0, + maximum_release_lag_ms=10.0, + output_age_p95_ms=80.0, + output_age_p99_ms=100.0, + maximum_layer_age_ms=110.0, + first_result_ms=500.0, + warmup_seconds=20.0, + stop_seconds=1.0, + peak_inflight_bytes=2_000_000, + peak_pending_camera_frames=2, + peak_vram_mib=12000, + peak_rss_mib=4000, + clock_uncertainty_ms=1.0, + backlog_growth_ms=0.0, + computed_layers=REQUIRED_LAYERS, + ) + + +def _layers() -> tuple[LayerEvidence, ...]: + return tuple( + LayerEvidence(name, "epoch-1", 0, 0, "current", "f" * 64) for name in REQUIRED_LAYERS + ) + + +def test_handshake_needs_no_source_size_duration_or_end() -> None: + value = _start() + assert StreamStart.from_dict(value.to_dict()) == value + assert len(json.dumps(value.to_dict()).encode()) < _budgets().maximum_start_metadata_bytes + assert replace(value, input_mode="live").channels == value.channels + assert replace(value, worker_id="worker-007").worker_id == "worker-007" + + +def test_existing_raw_first_ingress_delivers_without_session_end() -> None: + ingress = LivePerceptionIngress() + try: + ingress.begin_session("synthetic-source") + ingress.open_consumer("synthetic-worker") + assert ingress.take_next("synthetic-worker", timeout=0).modality == "control" + assert ingress.publish( + modality="camera-frame", + source_id="camera", + source_sequence=0, + captured_at_epoch_ns=1, + received_monotonic_ns=1, + payload=b"synthetic-frame", + ) + event = ingress.take_next("synthetic-worker", timeout=0) + assert event is not None and event.payload == b"synthetic-frame" + assert ingress.snapshot()["active"] is True + finally: + ingress.close() + + +@pytest.mark.parametrize( + "field", + [ + "total_bytes", + "frame_count", + "source_duration_ns", + "source_bundle_sha256", + "source_archive", + "source_members", + "wait_for_eof", + "command", + ], +) +def test_full_recording_and_execution_fields_are_not_start_requirements(field: str) -> None: + value = _start().to_dict() + value[field] = "not-an-input-to-the-stream-handshake" + with pytest.raises(RealtimeContractError, match="fields"): + StreamStart.from_dict(value) + + +@pytest.mark.parametrize( + "change", + [ + {"lease_generation": 0}, + {"lease_generation": True}, + {"epoch_id": ""}, + {"image_sha256": "latest"}, + {"input_mode": "offline-batch"}, + {"channels": ("camera",)}, + {"calibration_sha256": None}, + ], +) +def test_invalid_handshakes_fail_closed(change: dict) -> None: + with pytest.raises(RealtimeContractError): + replace(_start(), **change) + + +def test_scene_reports_all_layers_before_any_end_marker_exists() -> None: + validate_scene_layers(_layers(), epoch_id="epoch-1", source_time_ns=0, maximum_layer_age_ms=250) + + +@pytest.mark.parametrize("layers", [_layers()[:-1], _layers()[:-1] + (_layers()[0],)]) +def test_missing_or_duplicate_layer_is_not_a_complete_profile(layers: tuple) -> None: + with pytest.raises(RealtimeContractError, match="exactly once"): + validate_scene_layers( + layers, epoch_id="epoch-1", source_time_ns=0, maximum_layer_age_ms=250 + ) + + +@pytest.mark.parametrize( + "change,reason", + [ + ({"epoch_id": "epoch-old"}, "another stream epoch"), + ({"source_time_ns": 2_000_000_000}, "future"), + ({"source_time_ns": 0, "state": "held"}, "expired"), + ], +) +def test_cross_epoch_future_and_expired_evidence_rejected(change: dict, reason: str) -> None: + layers = tuple(replace(x, source_time_ns=1_000_000_000) for x in _layers()) + layers = (replace(layers[0], **change),) + layers[1:] + with pytest.raises(RealtimeContractError, match=reason): + validate_scene_layers( + layers, epoch_id="epoch-1", source_time_ns=1_000_000_000, maximum_layer_age_ms=250 + ) + + +def test_explicit_missing_lidar_does_not_invent_geometry_or_distance() -> None: + missing = LayerEvidence("geometry", "epoch-1", None, None, "unavailable", None) + layers = tuple(missing if x.layer == "geometry" else x for x in _layers()) + validate_scene_layers(layers, epoch_id="epoch-1", source_time_ns=0, maximum_layer_age_ms=250) + with pytest.raises(RealtimeContractError, match="cannot claim"): + replace(missing, payload_sha256="e" * 64) + + +def test_synthetic_receipt_passes_only_whole_path_limits() -> None: + assert realtime_failures(_measurements(), _budgets()) == () + + +@pytest.mark.parametrize( + "change,reason", + [ + ({"full_source_preloaded": True}, "full-source-preload"), + ({"results_before_end_of_source": False}, "no-incremental-result"), + ({"replay_speed": 0.5}, "source-clock-not-1x"), + ({"terminal_frames": 101}, "incomplete-input-accounting"), + ({"fresh_complete_scenes": 50}, "incomplete-fresh-scenes"), + ({"capacity_drops": 1}, "capacity-drops"), + ({"expired_frames": 1}, "expired-frames"), + ({"failed_frames": 1}, "failed-frames"), + ({"unexpected_unavailable_layers": 1}, "required-layer-unavailable"), + ({"computed_layers": ("segmentation",)}, "full-profile-not-computed"), + ({"peak_pending_camera_frames": 3}, "peak_pending_camera_frames"), + ({"peak_inflight_bytes": 20_000_000}, "peak_inflight_bytes"), + ({"clock_uncertainty_ms": 6}, "clock_uncertainty_ms"), + ({"backlog_growth_ms": 30}, "backlog_growth_ms"), + ({"output_age_p99_ms": 130}, "output_age_p99_ms"), + ({"maximum_layer_age_ms": 300}, "maximum_layer_age_ms"), + ({"warmup_seconds": 121}, "warmup_seconds"), + ({"stop_seconds": 6}, "stop_seconds"), + ], +) +def test_fast_model_or_complete_archive_does_not_pass_realtime(change: dict, reason: str) -> None: + assert reason in realtime_failures(replace(_measurements(), **change), _budgets()) + + +def test_source_gaps_stay_visible_and_are_not_model_capacity_drops() -> None: + value = replace(_measurements(), declared_source_gap_frames=10, fresh_complete_scenes=90) + assert realtime_failures(value, _budgets()) == () + assert "incomplete-fresh-scenes" in realtime_failures( + replace(value, fresh_complete_scenes=100), _budgets() + ) + assert "no-evaluable-scene" in realtime_failures( + replace(value, declared_source_gap_frames=100, fresh_complete_scenes=0), _budgets() + ) + + +@pytest.mark.parametrize( + "change", + [ + {"output_age_p99_ms": float("nan")}, + {"clock_uncertainty_ms": float("inf")}, + {"peak_vram_mib": -1}, + {"source_frames": True}, + {"full_source_preloaded": 0}, + ], +) +def test_non_finite_or_ambiguous_measurements_rejected(change: dict) -> None: + with pytest.raises(RealtimeContractError): + replace(_measurements(), **change) + + +def test_budget_schema_rejects_silent_fields_and_inconsistent_limits() -> None: + with pytest.raises(RealtimeContractError): + RealtimeBudgets.from_dict({**asdict(_budgets()), "allow_full_download": True}) + with pytest.raises(RealtimeContractError): + replace(_budgets(), maximum_chunk_bytes=100_000_000) + + +def test_profile_pins_reused_assets_without_installing_a_fake_ready_executor() -> None: + profile = _profile() + assert profile["packaging"]["image_sha256"] is None + assert set(profile["components"]) == set(REQUIRED_LAYERS) + assert profile["stream_policy"]["inference_stride"] == 1 + assert profile["input_contract"]["full_source_transfer_required"] is False + assert profile["authority"]["actuation_allowed"] is False + assert profile["components"]["geometry"]["generic_static_class"] == "static.unknown" + assert ( + profile["components"]["policy"]["fine_road_sidewalk_bikeway_distinction_required"] is False + ) + for ref in profile["reference_files"]: + assert hashlib.sha256((ROOT / ref["path"]).read_bytes()).hexdigest() == ref["sha256"] + registry = (ROOT / "config/observatory-portable-run-definitions.json").read_text() + assert profile["profile_id"] not in registry + + +def test_experimental_overload_does_not_become_realtime_qualification() -> None: + policy = _profile()["experiment_policy"] + assert policy["bounded_lab_overload_allowed"] is True + assert policy["retain_profiles_that_fail_current_hardware_realtime"] is True + assert policy["realtime_failure_blocks_experimental_packaging"] is False + assert policy["overload_is_realtime_pass"] is False + measurements = replace(_measurements(), output_age_p95_ms=150.0, output_age_p99_ms=200.0) + failures = realtime_failures(measurements, _budgets()) + assert "output_age_p95_ms" in failures and "output_age_p99_ms" in failures