From 097e450a8742df2acd92e2a782750c07901c6687 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 2 Sep 2026 10:14:28 +0300 Subject: [PATCH] feat(perception): enforce six-layer freshness through receipt --- ...tion-ddrnet39-rfdetr-tgs-prototype-v1.json | 31 ++ .../pilot_freshness.py | 139 ++++++++ .../streaming_profile_stage1/pilot_graph.py | 9 +- .../run_joint_pilot.py | 55 +--- src/k1link/perception/realtime_contract.py | 19 +- src/k1link/perception/realtime_scene.py | 264 ++++++++++++++++ .../perception/worker_operating_envelope.py | 153 +++++++++ tests/test_perception_realtime_scene.py | 298 ++++++++++++++++++ tests/test_worker_operating_envelope.py | 142 +++++++++ 9 files changed, 1067 insertions(+), 43 deletions(-) create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_freshness.py create mode 100644 src/k1link/perception/realtime_scene.py create mode 100644 src/k1link/perception/worker_operating_envelope.py create mode 100644 tests/test_perception_realtime_scene.py create mode 100644 tests/test_worker_operating_envelope.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 index 29e7966..c6d23b1 100644 --- a/config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json +++ b/config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json @@ -36,6 +36,37 @@ "MKL_NUM_THREADS": 1, "reason": "bounded-single-thread-numeric-libraries-before-whole-graph-qualification" }, + "operating_envelope": { + "status": "bounded-latency-reference-not-full-qualification", + "checker": "k1link.perception.worker_operating_envelope.operating_envelope_failures", + "reference_conditions": { + "envelope_id": "worker006-4090-610.47-stock-clock-reference/v1", + "gpu_name": "NVIDIA GeForce RTX 4090", + "driver_version": "610.47", + "cpu_limit_millicores": 8000, + "memory_limit_mib": 8192, + "minimum_sm_clock_mhz": 2610, + "minimum_memory_clock_mhz": 10251, + "maximum_snapshot_age_ms": 1000 + }, + "check_phase": "post-warmup-before-timed-window-plus-runtime-monitoring", + "container_may_set_host_clocks": false, + "persistent_host_clock_policy_installed": false, + "readiness_is_realtime_qualification": false, + "unknown_ownership_or_telemetry_is_pass": false, + "failed_envelope_allows_labelled_experiment": true + }, + "output_freshness_contract": { + "schema_version": "missioncore.perception-scene-freshness/v1", + "required_layers": ["segmentation", "objects", "geometry", "motion", "costmap", "policy"], + "retain_original_layer_identity": true, + "propagate_required_input_age": true, + "recheck_at_publication_receipt_and_use": true, + "clock_uncertainty_adds_to_age": true, + "unavailable_or_stale_required_layer": "no-go-advisory-only", + "history_can_independently_authorize_terrain": false, + "transport_streamstart_and_lease_fencing_required_separately": true + }, "input_contract": { "schema_version": "missioncore.perception-stream-start/v1", "channels": ["camera", "point-cloud", "pose"], diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_freshness.py b/experiments/perception/worker/streaming_profile_stage1/pilot_freshness.py new file mode 100644 index 0000000..4a68474 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_freshness.py @@ -0,0 +1,139 @@ +"""Pilot adapter for the shared six-layer ABI; source arrival clock, local receipt. + +Not a live clock synchronizer or lease authority. Domain history/unknown motion +stays in existing payloads. An unavailable layer may retain diagnostic fields in +the scene, but makes no valid payload/terrain-permission claim in the ABI. +""" + +import hashlib +import json + +from k1link.perception.realtime_contract import LayerEvidence +from k1link.perception.realtime_scene import DEPENDENCIES, SceneFreshness, derive_layer + +CLOCK_DOMAIN = "pilot-original-host-arrival" + + +def payload_digest(scene, layer): + if layer == "segmentation": + return scene["segmentation_sha256"] + fields = { + "objects": ("proposals",), + "geometry": ("observations", "surface_state", "range_estimator"), + "motion": ("tracks", "threats"), + "costmap": ("costmap_states", "costmap_material"), + "policy": ("policy_actions", "policy_counts"), + }[layer] + value = {key: scene[key] for key in fields} + return hashlib.sha256( + json.dumps(value, sort_keys=True, allow_nan=False, separators=(",", ":")).encode() + ).hexdigest() + + +def build_freshness(scene, bundle, ddr_layer, epoch_id): + stamp, sequence = bundle["time_ns"], bundle["sequence"] + segmentation = LayerEvidence( + "segmentation", + epoch_id, + ddr_layer["source_sequence"], + ddr_layer["source_host_monotonic_ns"], + "current" if ddr_layer["state"] == "current" else "held", + payload_digest(scene, "segmentation"), + ddr_layer["source_host_monotonic_ns"], + ) + objects = LayerEvidence( + "objects", epoch_id, sequence, stamp, "current", payload_digest(scene, "objects"), stamp + ) + layers = {"segmentation": segmentation, "objects": objects} + for name in ("geometry", "motion", "costmap", "policy"): + oldest = stamp + if name == "geometry": + if not bundle["available"] or scene["surface_state"] != "valid": + layers[name] = LayerEvidence(name, epoch_id, None, None, "unavailable", None) + continue + lineage = bundle["lineage"] + oldest = min( + lineage["pose_host_monotonic_ns"], + *(item["host_monotonic_ns"] for item in lineage["point_increments"]), + ) + if name == "costmap" and bundle["available"]: + # A rolling window can retain old occupied/unknown cells to prohibit + # terrain. Permissive cells must keep their own observation age. + oldest = scene["tgs_counts"]["oldest_permissive_cell_source_ns"] + layers[name] = derive_layer( + name, + epoch_id=epoch_id, + source_sequence=sequence, + source_time_ns=stamp, + payload_sha256=payload_digest(scene, name), + inputs=tuple(layers[key] for key in DEPENDENCIES[name]), + oldest_required_input_time_ns=oldest, + ) + return SceneFreshness(epoch_id, CLOCK_DOMAIN, sequence, stamp, tuple(layers.values())) + + +def assess(freshness, *, bundle, now_ns): + # Both sides of this pilot use the SAME monotonic clock. Map it back to + # original source arrival time; zero mapping uncertainty is local-only and + # is not a claim of hardware camera/LiDAR synchronization or network quality. + return freshness.assess( + epoch_id=freshness.epoch_id, + clock_domain_id=CLOCK_DOMAIN, + observed_source_time_ns=bundle["time_ns"] + now_ns - bundle["due_ns"], + clock_uncertainty_ms=0, + maximum_clock_uncertainty_ms=5, + maximum_layer_age_ms=250, + ) + + +def suppress_policy(scene): + scene["policy_actions"] = [2] * len(scene["policy_actions"]) + scene["policy_counts"] = { + "ALLOW_candidate": 0, + "HIGH_COST": 0, + "NO_GO": len(scene["policy_actions"]), + } + + +def prepare_publication(scene, bundle, ddr_layer, *, epoch_id, now_ns): + freshness = build_freshness(scene, bundle, ddr_layer, epoch_id) + checked = assess(freshness, bundle=bundle, now_ns=now_ns) + if not checked.fresh_complete: + suppress_policy(scene) + # The envelope hashes the actual guarded policy, not the discarded one. + freshness = build_freshness(scene, bundle, ddr_layer, epoch_id) + checked = assess(freshness, bundle=bundle, now_ns=now_ns) + scene["freshness"] = freshness.to_dict() + scene["freshness_at_publication"] = checked.to_dict() + scene["stale_at_publication"] = any(item.state == "stale" for item in checked.layers) + scene["oldest_required_input_age_ms"] = max( + (age for age in checked.age_upper_bound_ms if age is not None), default=None + ) + + +def validate_receipt(scene, bundle, *, epoch_id): + freshness = SceneFreshness.from_dict(scene["freshness"]) + if ( + freshness.epoch_id, + freshness.clock_domain_id, + freshness.source_sequence, + freshness.source_time_ns, + ) != (epoch_id, CLOCK_DOMAIN, bundle["sequence"], bundle["time_ns"]): + raise ValueError("collector freshness identity mismatch") + if scene["commands_enabled"] is not False or scene["actuation_allowed"] is not False: + raise ValueError("pilot result attempted control authority") + for item in freshness.layers: + if ( + item.payload_sha256 is not None + and payload_digest(scene, item.layer) != item.payload_sha256 + ): + raise ValueError("collector layer payload digest mismatch") + return freshness + + +def receipt_view(scene, checked): + """A consumer view; never mutate the hashed published evidence.""" + view = dict(scene) + if not checked.fresh_complete: + suppress_policy(view) + return view diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py b/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py index 2d0d954..d7b6243 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py @@ -444,7 +444,7 @@ class JointGraph: for index in diagnostic_cells: matching = np.flatnonzero(good & (cell_ids == index)) diagnostics[str(index)] = { - "votes": dict(zip(self.material_names, votes[index].tolist())), + "votes": dict(zip(self.material_names, votes[index].tolist(), strict=True)), "projected_ground_points": len(matching), "points_truncated": len(matching) > 128, "points": [ @@ -467,6 +467,12 @@ class JointGraph: if index is not None: cells[index] = 2 actions = self.action_lut[material, cells] + permissive_stamps = last_seen[actions != 2] + if np.any(permissive_stamps < 0): + raise ValueError("permissive TGS cell lacks source observation") + oldest_permissive = ( + int(permissive_stamps.min()) if len(permissive_stamps) else bundle["time_ns"] + ) return ( cells, actions, @@ -478,6 +484,7 @@ class JointGraph: "occupied": int(np.count_nonzero(states == 2)), "rejected": int(np.count_nonzero(states == 3)), "stale_cells": int(np.count_nonzero(stale)), + "oldest_permissive_cell_source_ns": oldest_permissive, "state_counts": dict(Counter(int(x) for x in cells)), **({"diagnostic_cells": diagnostics} if diagnostics else {}), }, 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 d49d5d6..f329a80 100644 --- a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py +++ b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py @@ -24,6 +24,7 @@ from datetime import UTC, datetime from pathlib import Path import numpy as np +from pilot_freshness import assess, prepare_publication, receipt_view, validate_receipt from pilot_ipc import receive, send from pilot_queue import Mailbox from pilot_scheduler import GpuStage @@ -482,44 +483,11 @@ def run(args): proposals=proposals, detector_ms=(gpu_done - ddr_done) / 1e6, ) - # Late evidence remains inspectable but cannot authorize terrain. - policy_age_ms = (time.monotonic_ns() - bundle["due_ns"]) / 1e6 - sensor_source_age_ms = max( - bundle["lineage"]["pose_age_ms"] or 0, - bundle["lineage"]["oldest_point_age_ms"] or 0, + freshness_started = time.monotonic_ns() + prepare_publication( + scene, bundle, ddr_layer, epoch_id=args.run_id, + now_ns=freshness_started, ) - scene["layer_freshness"] = { - "segmentation": { - **ddr_layer, - "age_at_policy_ms": policy_age_ms + ddr_layer["source_age_ms"], - }, - "detection": { - "state": "current", - "source_sequence": bundle["sequence"], - "source_host_monotonic_ns": bundle["time_ns"], - "source_age_ms": 0.0, - "age_at_policy_ms": policy_age_ms, - }, - "geometry": { - "state": "current" if bundle["available"] else "unavailable", - "source_sequence": bundle["sequence"], - "source_host_monotonic_ns": bundle["time_ns"], - "source_age_ms": sensor_source_age_ms, - "age_at_policy_ms": policy_age_ms + sensor_source_age_ms, - }, - } - scene["oldest_required_input_age_ms"] = policy_age_ms + max( - sensor_source_age_ms, - ddr_layer["source_age_ms"], - ) - scene["stale_at_publication"] = scene["oldest_required_input_age_ms"] > 250 - if scene["stale_at_publication"]: - scene["policy_actions"] = [2] * len(scene["policy_actions"]) - scene["policy_counts"] = { - "ALLOW_candidate": 0, - "HIGH_COST": 0, - "NO_GO": len(scene["policy_actions"]), - } scene.update( sequence=bundle["sequence"], lineage=bundle["lineage"], @@ -534,6 +502,10 @@ def run(args): received = json.loads(encoded) if received["sequence"] != bundle["sequence"]: raise ValueError("collector identity mismatch") + freshness = validate_receipt(received, bundle, epoch_id=args.run_id) + checked_at = time.monotonic_ns() + checked = assess(freshness, bundle=bundle, now_ns=checked_at) + view = receipt_view(received, checked) finished = time.monotonic_ns() timing = { **scene["timing_ms"], @@ -549,6 +521,7 @@ def run(args): "cpu_tail_to_receiver_ms": (finished - cpu_started) / 1e6, "compute_to_receiver_ms": (finished - begin) / 1e6, "source_due_to_receiver_ms": (finished - bundle["due_ns"]) / 1e6, + "freshness_encode_receive_ms": (finished - freshness_started) / 1e6, } result = { "sequence": bundle["sequence"], @@ -565,7 +538,9 @@ def run(args): x["metric_geometry"] is not None for x in scene["observations"] ), "tgs_counts": scene["tgs_counts"], - "policy_counts": scene["policy_counts"], + "policy_counts": view["policy_counts"], + "freshness_at_receipt": checked.to_dict(), + "policy_counts_at_publication": scene["policy_counts"], "scene_bytes": len(encoded), "scene_sha256": hashlib.sha256(encoded).hexdigest(), } @@ -711,8 +686,8 @@ def run(args): and max(s.get("cgroup_memory_mib", 99999) for s in samples) <= 8192, "all_modalities_fresh": len(results) == args.frames and all(r["available"] for r in results), - "layer_source_age_250ms": bool(results) - and all(r["ddrnet_source_age_ms"] <= 250 for r in results), + "all_six_layers_fresh_at_receipt": bool(results) + and all(r["freshness_at_receipt"]["fresh_complete"] for r in results), "network_end_to_end_qualified": False, } report["profile_realtime_qualified"] = False diff --git a/src/k1link/perception/realtime_contract.py b/src/k1link/perception/realtime_contract.py index ad1abe8..eef6a3a 100644 --- a/src/k1link/perception/realtime_contract.py +++ b/src/k1link/perception/realtime_contract.py @@ -181,6 +181,9 @@ class LayerEvidence: source_time_ns: int | None state: str payload_sha256: str | None + # Old callers omit this field. New layered results require it explicitly: + # recomputing a derived layer cannot refresh the age of its required inputs. + oldest_required_input_time_ns: int | None = None def __post_init__(self) -> None: if self.layer not in REQUIRED_LAYERS: @@ -191,13 +194,22 @@ class LayerEvidence: if self.state == "unavailable": if any( x is not None - for x in (self.source_sequence, self.source_time_ns, self.payload_sha256) + for x in ( + self.source_sequence, + self.source_time_ns, + self.payload_sha256, + self.oldest_required_input_time_ns, + ) ): 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") + if self.oldest_required_input_time_ns is not None: + _integer(self.oldest_required_input_time_ns, "oldest_required_input_time_ns") + if self.oldest_required_input_time_ns > cast(int, self.source_time_ns): + raise RealtimeContractError("required input is newer than its layer") def validate_scene_layers( @@ -223,7 +235,10 @@ def validate_scene_layers( 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 layer.source_time_ns > source_time_ns: + raise RealtimeContractError("future observations are not causal input") + stamp = layer.oldest_required_input_time_ns + age_ns = source_time_ns - (layer.source_time_ns if stamp is None else stamp) 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"): diff --git a/src/k1link/perception/realtime_scene.py b/src/k1link/perception/realtime_scene.py new file mode 100644 index 0000000..c40f12d --- /dev/null +++ b/src/k1link/perception/realtime_scene.py @@ -0,0 +1,264 @@ +"""Bounded six-layer freshness ABI; no I/O, models or control authority. + +This wraps existing domain payloads. It is not the transport/session admission +protocol: a transport must also validate StreamStart identities and lease fences. +All times here are in ONE mapped source timeline, never raw remote worker clocks. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, replace +from typing import Final + +from .realtime_contract import ( + REQUIRED_LAYERS, + LayerEvidence, + RealtimeContractError, + _identifier, + _integer, + _number, +) + +SCENE_FRESHNESS_SCHEMA: Final = "missioncore.perception-scene-freshness/v1" +_INT64_FIELDS: Final = ("source_sequence", "source_time_ns", "oldest_required_input_time_ns") + + +def _wire_integer(value: int) -> str: + _integer(value, "wire integer") + if value > 2**63 - 1: + raise RealtimeContractError("wire integer exceeds int64") + return str(value) + + +def _read_wire_integer(value: object) -> int: + if ( + not isinstance(value, str) + or not value.isascii() + or not value.isdecimal() + or len(value) > 19 + or (len(value) > 1 and value.startswith("0")) + ): + raise RealtimeContractError("int64 must be a canonical decimal string") + parsed = int(value) + _wire_integer(parsed) + return parsed + + +def _layer_document(item: LayerEvidence) -> dict[str, object]: + value = asdict(item) + for name in _INT64_FIELDS: + number = value[name] + value[name] = None if number is None else _wire_integer(number) + return value + + +# Geometry includes detector association; history may add prohibitions, not +# independently authorize terrain. This is the full profile's dependency graph. +DEPENDENCIES: Final = { + "segmentation": (), + "objects": (), + "geometry": ("objects",), + "motion": ("geometry",), + "costmap": ("segmentation", "geometry", "motion"), + "policy": ("costmap", "objects", "motion"), +} + + +def derive_layer( + layer: str, + *, + epoch_id: str, + source_sequence: int, + source_time_ns: int, + payload_sha256: str, + inputs: tuple[LayerEvidence, ...], + oldest_required_input_time_ns: int, +) -> LayerEvidence: + """Propagate missing/stale inputs and their original ages, not execution time.""" + if layer not in DEPENDENCIES or tuple(x.layer for x in inputs) != DEPENDENCIES[layer]: + raise RealtimeContractError("derived layer dependencies changed") + if any(x.epoch_id != epoch_id for x in inputs): + raise RealtimeContractError("derived input belongs to another epoch") + if any(x.state == "unavailable" for x in inputs): + return LayerEvidence(layer, epoch_id, None, None, "unavailable", None) + stamps = [oldest_required_input_time_ns] + for item in inputs: + if item.oldest_required_input_time_ns is None: + raise RealtimeContractError("required input age is missing") + stamps.append(item.oldest_required_input_time_ns) + return LayerEvidence( + layer, + epoch_id, + source_sequence, + source_time_ns, + "stale" if any(x.state == "stale" for x in inputs) else "current", + payload_sha256, + min(stamps), + ) + + +@dataclass(frozen=True, slots=True) +class FreshnessAssessment: + checked_at_source_time_ns: int + clock_uncertainty_ms: float + layers: tuple[LayerEvidence, ...] + age_upper_bound_ms: tuple[float | None, ...] + failures: tuple[str, ...] + + @property + def fresh_complete(self) -> bool: + return not self.failures + + def to_dict(self) -> dict[str, object]: + return { + "checked_at_source_time_ns": _wire_integer(self.checked_at_source_time_ns), + "clock_uncertainty_ms": self.clock_uncertainty_ms, + "layers": { + item.layer: {**_layer_document(item), "age_upper_bound_ms": age} + for item, age in zip(self.layers, self.age_upper_bound_ms, strict=True) + }, + "fresh_complete": self.fresh_complete, + "failures": list(self.failures), + "commands_enabled": False, + "actuation_allowed": False, + } + + +@dataclass(frozen=True, slots=True) +class SceneFreshness: + epoch_id: str + clock_domain_id: str + source_sequence: int + source_time_ns: int + layers: tuple[LayerEvidence, ...] + + def __post_init__(self) -> None: + _identifier(self.epoch_id, "epoch_id") + _identifier(self.clock_domain_id, "clock_domain_id") + _integer(self.source_sequence, "source_sequence") + _integer(self.source_time_ns, "source_time_ns") + if ( + not isinstance(self.layers, tuple) + or tuple(x.layer for x in self.layers) != REQUIRED_LAYERS + ): + raise RealtimeContractError("six layers required in canonical order") + indexed = {item.layer: item for item in self.layers} + for item in self.layers: + if item.epoch_id != self.epoch_id: + raise RealtimeContractError("layer belongs to another epoch") + if item.state == "unavailable": + continue + if ( + item.source_sequence is None + or item.source_time_ns is None + or item.oldest_required_input_time_ns is None + ): + raise RealtimeContractError("layer input lineage missing") + if ( + item.source_sequence > self.source_sequence + or item.source_time_ns > self.source_time_ns + ): + raise RealtimeContractError("future layer evidence") + if item.state == "current" and ( + item.source_sequence != self.source_sequence + or item.source_time_ns != self.source_time_ns + ): + raise RealtimeContractError("retained evidence must be held, not current") + if item.state == "held" and item.source_sequence == self.source_sequence: + raise RealtimeContractError("held layer must retain an earlier observation") + for name in DEPENDENCIES[item.layer]: + parent = indexed[name] + if parent.oldest_required_input_time_ns is None: + raise RealtimeContractError("available layer depends on unavailable input") + if item.oldest_required_input_time_ns > parent.oldest_required_input_time_ns: + raise RealtimeContractError("derived layer refreshed an older input") + if parent.state == "stale" and item.state != "stale": + raise RealtimeContractError("derived layer hid stale input") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": SCENE_FRESHNESS_SCHEMA, + "epoch_id": self.epoch_id, + "clock_domain_id": self.clock_domain_id, + "source_sequence": _wire_integer(self.source_sequence), + "source_time_ns": _wire_integer(self.source_time_ns), + "layers": [_layer_document(item) for item in self.layers], + } + + @classmethod + def from_dict(cls, value: object) -> SceneFreshness: + if not isinstance(value, dict) or set(value) != set(cls.__dataclass_fields__) | { + "schema_version" + }: + raise RealtimeContractError("scene freshness fields changed") + if value["schema_version"] != SCENE_FRESHNESS_SCHEMA: + raise RealtimeContractError("scene freshness schema changed") + raw = value["layers"] + if not isinstance(raw, list) or len(raw) != len(REQUIRED_LAYERS): + raise RealtimeContractError("six layers required") + if any( + not isinstance(x, dict) or set(x) != set(LayerEvidence.__dataclass_fields__) + for x in raw + ): + raise RealtimeContractError("layer fields changed") + layers = [] + for item in raw: + parsed = dict(item) + for name in _INT64_FIELDS: + parsed[name] = None if item[name] is None else _read_wire_integer(item[name]) + layers.append(LayerEvidence(**parsed)) + return cls( + value["epoch_id"], + value["clock_domain_id"], + _read_wire_integer(value["source_sequence"]), + _read_wire_integer(value["source_time_ns"]), + tuple(layers), + ) + + def assess( + self, + *, + epoch_id: str, + clock_domain_id: str, + observed_source_time_ns: int, + clock_uncertainty_ms: float, + maximum_clock_uncertainty_ms: float, + maximum_layer_age_ms: float, + ) -> FreshnessAssessment: + """Re-evaluate at publication, receipt AND use. Uncertainty adds to age. + + Unknown/unmapped clocks must not call this with a guessed zero offset. + The source adapter owns the mapping; excessive known uncertainty yields + degraded output, never a complete fresh-scene claim. + """ + if (epoch_id, clock_domain_id) != (self.epoch_id, self.clock_domain_id): + raise RealtimeContractError("epoch or mapped clock domain mismatch") + _integer(observed_source_time_ns, "observed_source_time_ns") + _number(clock_uncertainty_ms, "clock_uncertainty_ms") + _number(maximum_clock_uncertainty_ms, "maximum_clock_uncertainty_ms", minimum=0.001) + _number(maximum_layer_age_ms, "maximum_layer_age_ms", minimum=0.001) + if observed_source_time_ns < self.source_time_ns: + raise RealtimeContractError("receipt precedes source observation") + clock_failed = clock_uncertainty_ms > maximum_clock_uncertainty_ms + failures = ["clock-uncertainty"] if clock_failed else [] + evaluated, ages = [], [] + for item in self.layers: + stamp = item.oldest_required_input_time_ns + age = ( + None + if stamp is None + else (observed_source_time_ns - stamp) / 1e6 + clock_uncertainty_ms + ) + if age is not None and (age > maximum_layer_age_ms or clock_failed): + item = replace(item, state="stale") + if item.state in ("stale", "unavailable"): + failures.append(f"{item.layer}-{item.state}") + evaluated.append(item) + ages.append(age) + return FreshnessAssessment( + observed_source_time_ns, + clock_uncertainty_ms, + tuple(evaluated), + tuple(ages), + tuple(failures), + ) diff --git a/src/k1link/perception/worker_operating_envelope.py b/src/k1link/perception/worker_operating_envelope.py new file mode 100644 index 0000000..85bf459 --- /dev/null +++ b/src/k1link/perception/worker_operating_envelope.py @@ -0,0 +1,153 @@ +"""Read-only post-warmup readiness checks, not GPU management or qualification. + +An envelope is preregistered for a measured hardware/software context. A failed +check does not prohibit an explicitly labelled overload experiment. Passing is +only a prerequisite: it cannot certify clocks throughout a run, latency, model +quality, standalone packaging or network delivery. No host setters live here. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from .realtime_contract import ( + RealtimeContractError, + StreamStart, + _digest, + _identifier, + _integer, +) + + +@dataclass(frozen=True, slots=True) +class WorkerOperatingEnvelope: + envelope_id: str + gpu_name: str + driver_version: str + # Runtime versions and all model/CPU-thread settings belong to the pinned + # image/effective config in StreamStart; resource limits are explicit here. + cpu_limit_millicores: int + memory_limit_mib: int + minimum_sm_clock_mhz: int + minimum_memory_clock_mhz: int + maximum_snapshot_age_ms: int = 1000 + + def __post_init__(self) -> None: + _identifier(self.envelope_id, "envelope_id") + for field in ("gpu_name", "driver_version"): + value = getattr(self, field) + if not isinstance(value, str) or not value.strip() or len(value) > 160: + raise RealtimeContractError(f"{field} is missing or unbounded") + for field in ( + "cpu_limit_millicores", + "memory_limit_mib", + "minimum_sm_clock_mhz", + "minimum_memory_clock_mhz", + "maximum_snapshot_age_ms", + ): + _integer(getattr(self, field), field, minimum=1) + + +@dataclass(frozen=True, slots=True) +class WorkerSnapshot: + worker_id: str + clock_domain_id: str + observed_monotonic_ns: int + gpu_name: str | None + driver_version: str | None + image_sha256: str | None + effective_config_sha256: str | None + cpu_limit_millicores: int | None + memory_limit_mib: int | None + sm_clock_mhz: int | None + memory_clock_mhz: int | None + # Authoritative controller lease, NOT inferred from low GPU utilization. + gpu_owner_run_id: str | None + lease_generation: int | None + competing_gpu_clients: tuple[str, ...] | None + warmup_complete: bool | None + + def __post_init__(self) -> None: + _identifier(self.worker_id, "worker_id") + _identifier(self.clock_domain_id, "clock_domain_id") + _integer(self.observed_monotonic_ns, "observed_monotonic_ns") + for field in ("image_sha256", "effective_config_sha256"): + value = getattr(self, field) + if value is not None: + _digest(value, field) + for field in ("gpu_name", "driver_version"): + value = getattr(self, field) + if value is not None and ( + not isinstance(value, str) or not value.strip() or len(value) > 160 + ): + raise RealtimeContractError(f"{field} is invalid") + for field in ( + "cpu_limit_millicores", + "memory_limit_mib", + "sm_clock_mhz", + "memory_clock_mhz", + ): + value = getattr(self, field) + if value is not None: + # Docker 0 means unlimited; never confuse it with unknown=None. + _integer(value, field) + if self.gpu_owner_run_id is not None: + _identifier(self.gpu_owner_run_id, "gpu_owner_run_id") + if self.lease_generation is not None: + _integer(self.lease_generation, "lease_generation", minimum=1) + if self.competing_gpu_clients is not None: + if ( + not isinstance(self.competing_gpu_clients, tuple) + or len(self.competing_gpu_clients) > 64 + ): + raise RealtimeContractError("GPU client inventory is unbounded") + for client in self.competing_gpu_clients: + _identifier(client, "competing_gpu_client") + if self.warmup_complete is not None and type(self.warmup_complete) is not bool: + raise RealtimeContractError("warmup_complete must be boolean or unknown") + + +def operating_envelope_failures( + expected: WorkerOperatingEnvelope, + start: StreamStart, + observed: WorkerSnapshot, + *, + now_monotonic_ns: int, + clock_domain_id: str, +) -> tuple[str, ...]: + """Check facts collected in the Worker's clock domain by a trusted controller. + + No resource expansion, frequency lock, container stop, lease acquisition or + host mutation. Snapshot refresh and ongoing telemetry remain caller-owned. + Changing hardware/resource conditions requires a new measured envelope, not + pretending that an old qualification applies to all compatible machines. + """ + _integer(now_monotonic_ns, "now_monotonic_ns") + if observed.clock_domain_id != clock_domain_id: + raise RealtimeContractError("worker snapshot clock domain mismatch") + age = now_monotonic_ns - observed.observed_monotonic_ns + if age < 0: + raise RealtimeContractError("worker snapshot is from the future") + failures = [] + if age > expected.maximum_snapshot_age_ms * 1_000_000: + failures.append("worker-snapshot-expired") + for field in ("worker_id", "image_sha256", "effective_config_sha256"): + if getattr(observed, field) != getattr(start, field): + failures.append(f"{field}-mismatch-or-unknown") + for field in ("gpu_name", "driver_version", "cpu_limit_millicores", "memory_limit_mib"): + if getattr(observed, field) != getattr(expected, field): + failures.append(f"{field}-outside-envelope-or-unknown") + for field in ("sm_clock_mhz", "memory_clock_mhz"): + value = getattr(observed, field) + if value is None or value < getattr(expected, "minimum_" + field): + failures.append(f"{field}-below-envelope-or-unknown") + if (observed.gpu_owner_run_id, observed.lease_generation) != ( + start.run_id, + start.lease_generation, + ): + failures.append("exclusive-worker-lease-unproved") + if observed.competing_gpu_clients != (): + failures.append("competing-gpu-clients-or-inventory-unknown") + if observed.warmup_complete is not True: + failures.append("warmup-not-complete") + return tuple(failures) diff --git a/tests/test_perception_realtime_scene.py b/tests/test_perception_realtime_scene.py new file mode 100644 index 0000000..7e43b17 --- /dev/null +++ b/tests/test_perception_realtime_scene.py @@ -0,0 +1,298 @@ +"""Small synthetic scene/consumer tests; no models, network or source archives.""" + +import copy +import importlib +import json +from dataclasses import replace +from pathlib import Path + +import pytest + +from k1link.perception.realtime_contract import ( + REQUIRED_LAYERS, + LayerEvidence, + RealtimeContractError, +) +from k1link.perception.realtime_scene import DEPENDENCIES, SceneFreshness, derive_layer + +NOW = 1_000_000_000 + + +def scene(*, held_ms=0, missing=False): + layers = { + "segmentation": LayerEvidence( + "segmentation", + "epoch", + 1 if held_ms else 2, + NOW - held_ms * 1_000_000, + "held" if held_ms else "current", + "a" * 64, + NOW - held_ms * 1_000_000, + ), + "objects": LayerEvidence("objects", "epoch", 2, NOW, "current", "b" * 64, NOW), + } + for name in ("geometry", "motion", "costmap", "policy"): + if name == "geometry" and missing: + layers[name] = LayerEvidence(name, "epoch", None, None, "unavailable", None) + else: + layers[name] = derive_layer( + name, + epoch_id="epoch", + source_sequence=2, + source_time_ns=NOW, + payload_sha256="c" * 64, + inputs=tuple(layers[x] for x in DEPENDENCIES[name]), + oldest_required_input_time_ns=NOW, + ) + return SceneFreshness("epoch", "mapped-source", 2, NOW, tuple(layers.values())) + + +def assess(value, *, elapsed_ms=0, **kwargs): + arguments = dict( + epoch_id="epoch", + clock_domain_id="mapped-source", + observed_source_time_ns=NOW + elapsed_ms * 1_000_000, + clock_uncertainty_ms=0, + maximum_clock_uncertainty_ms=5, + maximum_layer_age_ms=250, + ) + arguments.update(kwargs) + return value.assess(**arguments) + + +def test_six_layer_round_trip_is_bounded_and_needs_no_eof(): + value = scene() + wire = json.dumps(value.to_dict()) + assert len(wire.encode()) < 8192 + assert SceneFreshness.from_dict(json.loads(wire)) == value + checked = assess(value, elapsed_ms=125) + assert checked.fresh_complete + assert checked.age_upper_bound_ms == (125,) * 6 + assert checked.to_dict()["actuation_allowed"] is False + + +def test_json_projection_keeps_int64_exact_above_javascript_safe_integer(): + value = scene() + large = 9_007_199_254_740_993 + value = replace( + value, + source_time_ns=large, + layers=tuple( + replace(x, source_time_ns=large, oldest_required_input_time_ns=large) + for x in value.layers + ), + ) + wire = value.to_dict() + assert wire["source_time_ns"] == str(large) + assert SceneFreshness.from_dict(json.loads(json.dumps(wire))) == value + + +@pytest.mark.parametrize("number", [1_000_000_000, "01", "-1", "1.0", str(2**63)]) +def test_wire_rejects_lossy_or_noncanonical_int64(number): + wire = scene().to_dict() + wire["source_time_ns"] = number + with pytest.raises(RealtimeContractError): + SceneFreshness.from_dict(wire) + + +def test_held_segmenter_keeps_original_identity_through_costmap_and_policy(): + value = scene(held_ms=150) + assert value.layers[0].source_sequence == 1 + assert value.layers[-1].source_sequence == 2 + assert value.layers[-1].oldest_required_input_time_ns == NOW - 150_000_000 + assert assess(value, elapsed_ms=100).fresh_complete + expired = assess(value, elapsed_ms=101) + assert expired.failures == ("segmentation-stale", "costmap-stale", "policy-stale") + assert expired.layers[-1].source_time_ns == NOW # recomputation is not a new input + + +def test_receipt_latency_cannot_inherit_fresh_publication(): + value = scene() + assert assess(value, elapsed_ms=249).fresh_complete + assert not assess(value, elapsed_ms=251).fresh_complete + assert all(x.state == "stale" for x in assess(value, elapsed_ms=251).layers) + + +def test_missing_geometry_propagates_without_fabricating_zero_distance(): + value = scene(missing=True) + checked = assess(value) + assert not checked.fresh_complete + assert [x.layer for x in checked.layers if x.state == "unavailable"] == list( + REQUIRED_LAYERS[2:] + ) + assert all(x.payload_sha256 is None for x in checked.layers[2:]) + assert checked.age_upper_bound_ms[2:] == (None,) * 4 + + +@pytest.mark.parametrize( + "field,value", + [ + ("epoch_id", "old-epoch"), + ("clock_domain_id", "raw-worker-clock"), + ("observed_source_time_ns", NOW - 1), + ("clock_uncertainty_ms", None), + ("clock_uncertainty_ms", float("nan")), + ("clock_uncertainty_ms", True), + ], +) +def test_wrong_or_unmapped_clock_rejected(field, value): + with pytest.raises(RealtimeContractError): + assess(scene(), **{field: value}) + + +def test_known_clock_uncertainty_counts_against_budget(): + assert assess(scene(), elapsed_ms=245, clock_uncertainty_ms=5).fresh_complete + assert not assess(scene(), elapsed_ms=246, clock_uncertainty_ms=5).fresh_complete + assert "clock-uncertainty" in assess(scene(), clock_uncertainty_ms=6).failures + + +@pytest.mark.parametrize( + "change", + [ + {"source_sequence": 3}, + {"source_time_ns": NOW + 1}, + {"epoch_id": "old"}, + {"source_sequence": 1}, + {"state": "held"}, + {"oldest_required_input_time_ns": None}, + ], +) +def test_false_currentness_or_future_lineage_rejected(change): + value = scene() + layers = (replace(value.layers[0], **change), *value.layers[1:]) + with pytest.raises(RealtimeContractError): + replace(value, layers=layers) + + +def test_derived_layer_cannot_erase_old_or_missing_input(): + value = scene(held_ms=150) + with pytest.raises(RealtimeContractError, match="refreshed"): + replace( + value, + layers=( + *value.layers[:-1], + replace(value.layers[-1], oldest_required_input_time_ns=NOW), + ), + ) + missing = scene(missing=True) + with pytest.raises(RealtimeContractError, match="unavailable"): + replace(missing, layers=(*missing.layers[:-1], scene().layers[-1])) + + +def test_stale_input_cannot_be_relabelled_fresh(): + value = scene() + with pytest.raises(RealtimeContractError, match="stale"): + replace(value, layers=(replace(value.layers[0], state="stale"), *value.layers[1:])) + + +@pytest.mark.parametrize( + "mutation", + [ + lambda v: v.update(command="drive"), + lambda v: v.update(schema_version="unknown"), + lambda v: v["layers"].pop(), + lambda v: v["layers"].append(v["layers"][0]), + lambda v: v["layers"][0].update(age_ms=0), + lambda v: v["layers"].__setitem__(1, v["layers"][0]), + ], +) +def test_wire_schema_is_closed(mutation): + value = scene().to_dict() + mutation(value) + with pytest.raises(RealtimeContractError): + SceneFreshness.from_dict(value) + + +@pytest.fixture +def pilot(monkeypatch): + monkeypatch.syspath_prepend( + str( + Path(__file__).resolve().parents[1] + / "experiments/perception/worker/streaming_profile_stage1" + ) + ) + return importlib.import_module("pilot_freshness") + + +def pilot_input(): + payload = dict( + segmentation_sha256="a" * 64, + proposals=[], + observations=[], + tracks=[], + threats=[], + surface_state="valid", + range_estimator={"detected": "median-camera-z"}, + costmap_states=[1, 2], + costmap_material=[1, 0], + policy_actions=[0, 2], + policy_counts={"ALLOW_candidate": 1, "HIGH_COST": 0, "NO_GO": 1}, + tgs_counts={"oldest_permissive_cell_source_ns": NOW - 100_000_000}, + commands_enabled=False, + actuation_allowed=False, + ) + bundle = dict( + time_ns=NOW, + sequence=2, + due_ns=5_000_000_000, + available=True, + lineage={ + "pose_host_monotonic_ns": NOW - 10_000_000, + "point_increments": [{"host_monotonic_ns": NOW - 20_000_000}], + }, + ) + ddr = dict(state="current", source_sequence=2, source_host_monotonic_ns=NOW) + return payload, bundle, ddr + + +def test_pilot_receiver_checks_real_payloads_and_ages_after_serialization(pilot): + payload, bundle, ddr = pilot_input() + pilot.prepare_publication( + payload, bundle, ddr, epoch_id="pilot", now_ns=bundle["due_ns"] + 50_000_000 + ) + assert payload["freshness_at_publication"]["fresh_complete"] + assert payload["policy_actions"] == [0, 2] + received = json.loads(json.dumps(payload)) + freshness = pilot.validate_receipt(received, bundle, epoch_id="pilot") + # The costmap's last observed permissive cell ages beyond 250 ms in transit. + checked = pilot.assess(freshness, bundle=bundle, now_ns=bundle["due_ns"] + 151_000_000) + view = pilot.receipt_view(received, checked) + assert checked.failures == ("costmap-stale", "policy-stale") + assert view["policy_actions"] == [2, 2] + assert received["policy_actions"] == [0, 2] # published evidence remains immutable + received["costmap_material"][0] = 2 + with pytest.raises(ValueError, match="digest"): + pilot.validate_receipt(received, bundle, epoch_id="pilot") + + +def test_empty_detection_is_valid_but_missing_lidar_is_not(pilot): + payload, bundle, ddr = pilot_input() + bundle["available"] = False + pilot.prepare_publication(payload, bundle, ddr, epoch_id="pilot", now_ns=bundle["due_ns"]) + fresh = pilot.validate_receipt(payload, bundle, epoch_id="pilot") + assert fresh.layers[1].state == "current" + assert all(x.state == "unavailable" for x in fresh.layers[2:]) + assert payload["policy_actions"] == [2, 2] + assert payload["observations"] == [] + + +def test_failed_online_surface_is_not_fresh_geometry_even_with_sensor_pair(pilot): + payload, bundle, ddr = pilot_input() + payload["surface_state"] = "fit-failed" + pilot.prepare_publication(payload, bundle, ddr, epoch_id="pilot", now_ns=bundle["due_ns"]) + fresh = pilot.validate_receipt(payload, bundle, epoch_id="pilot") + assert all(x.state == "unavailable" for x in fresh.layers[2:]) + assert payload["policy_actions"] == [2, 2] + + +def test_pilot_rejects_wrong_run_sequence_and_control_authority(pilot): + payload, bundle, ddr = pilot_input() + pilot.prepare_publication(payload, bundle, ddr, epoch_id="pilot", now_ns=bundle["due_ns"]) + with pytest.raises(ValueError, match="identity"): + pilot.validate_receipt(payload, bundle, epoch_id="old-pilot") + with pytest.raises(ValueError, match="identity"): + pilot.validate_receipt(payload, {**bundle, "sequence": 3}, epoch_id="pilot") + changed = copy.deepcopy(payload) + changed["commands_enabled"] = True + with pytest.raises(ValueError, match="authority"): + pilot.validate_receipt(changed, bundle, epoch_id="pilot") diff --git a/tests/test_worker_operating_envelope.py b/tests/test_worker_operating_envelope.py new file mode 100644 index 0000000..7baf50c --- /dev/null +++ b/tests/test_worker_operating_envelope.py @@ -0,0 +1,142 @@ +"""Synthetic readiness checks; passing must never mutate a Worker or grant control.""" + +import json +from dataclasses import replace +from pathlib import Path + +import pytest + +from k1link.perception.realtime_contract import RealtimeContractError, StreamStart +from k1link.perception.worker_operating_envelope import ( + WorkerOperatingEnvelope, + WorkerSnapshot, + operating_envelope_failures, +) + + +def start(): + return StreamStart( + "run-1", + "source-1", + "worker-006", + "epoch-1", + 1, + "a" * 64, + "b" * 64, + "c" * 64, + "d" * 64, + "source-clock", + "recorded-source-paced", + ) + + +def envelope(): + return WorkerOperatingEnvelope( + "synthetic-4090-fixed/v1", "RTX 4090", "610.47", 8000, 8192, 2610, 10251 + ) + + +def snapshot(): + return WorkerSnapshot( + "worker-006", + "worker-monotonic", + 1_000_000_000, + "RTX 4090", + "610.47", + "b" * 64, + "c" * 64, + 8000, + 8192, + 2610, + 10251, + "run-1", + 1, + (), + True, + ) + + +def check(value, **kwargs): + options = {"now_monotonic_ns": 1_100_000_000, "clock_domain_id": "worker-monotonic"} + options.update(kwargs) + return operating_envelope_failures(envelope(), start(), value, **options) + + +def test_exact_measured_conditions_pass_only_readiness(): + value = snapshot() + assert check(value) == () + assert value == snapshot() + assert "qualified" not in value.__dataclass_fields__ + + +def test_candidate_preregisters_conditions_without_installing_host_policy(): + path = ( + Path(__file__).resolve().parents[1] + / "config/perception/k1-perception-ddrnet39-rfdetr-tgs-prototype-v1.json" + ) + profile = json.loads(path.read_text()) + value = profile["operating_envelope"] + parsed = WorkerOperatingEnvelope(**value["reference_conditions"]) + assert parsed.minimum_memory_clock_mhz == 10251 + assert not value["container_may_set_host_clocks"] + assert not value["readiness_is_realtime_qualification"] + assert value["failed_envelope_allows_labelled_experiment"] + assert profile["packaging"]["image_sha256"] is None + + +@pytest.mark.parametrize( + "field,value,reason", + [ + ("worker_id", "worker-007", "worker_id-mismatch"), + ("gpu_name", "RTX 5090", "gpu_name-outside"), + ("driver_version", "new-driver", "driver_version-outside"), + ("image_sha256", "e" * 64, "image_sha256-mismatch"), + ("effective_config_sha256", "e" * 64, "effective_config_sha256-mismatch"), + ("sm_clock_mhz", 450, "sm_clock_mhz-below"), + ("memory_clock_mhz", 405, "memory_clock_mhz-below"), + ("memory_clock_mhz", None, "memory_clock_mhz-below"), + ("cpu_limit_millicores", 4000, "cpu_limit_millicores-outside"), + ("memory_limit_mib", 0, "memory_limit_mib-outside"), + ("cpu_limit_millicores", None, "cpu_limit_millicores-outside"), + ("gpu_owner_run_id", None, "exclusive-worker-lease"), + ("gpu_owner_run_id", "old-run", "exclusive-worker-lease"), + ("lease_generation", 2, "exclusive-worker-lease"), + ("competing_gpu_clients", ("other-model",), "competing-gpu-clients"), + ("competing_gpu_clients", None, "competing-gpu-clients"), + ("warmup_complete", None, "warmup-not-complete"), + ("warmup_complete", False, "warmup-not-complete"), + ], +) +def test_auto_clocks_wrong_identity_unknown_or_busy_never_pass(field, value, reason): + assert any(x.startswith(reason) for x in check(replace(snapshot(), **{field: value}))) + + +def test_low_utilization_does_not_prove_ownership_and_idle_does_not_prove_warmup(): + failures = check(replace(snapshot(), gpu_owner_run_id=None, warmup_complete=False)) + assert failures == ("exclusive-worker-lease-unproved", "warmup-not-complete") + + +def test_snapshot_age_and_clock_domain_cannot_be_ignored(): + assert check(snapshot(), now_monotonic_ns=2_000_000_000) == () + assert check(snapshot(), now_monotonic_ns=2_000_000_001) == ("worker-snapshot-expired",) + with pytest.raises(RealtimeContractError, match="clock domain"): + check(snapshot(), clock_domain_id="mac-monotonic") + with pytest.raises(RealtimeContractError, match="future"): + check(snapshot(), now_monotonic_ns=999_999_999) + + +@pytest.mark.parametrize( + "change", + [ + {"sm_clock_mhz": True}, + {"memory_limit_mib": -1}, + {"warmup_complete": 1}, + {"image_sha256": "latest"}, + {"lease_generation": 0}, + {"competing_gpu_clients": ["unbounded-list"]}, + {"competing_gpu_clients": ("client",) * 65}, + ], +) +def test_invalid_facts_fail_closed(change): + with pytest.raises(RealtimeContractError): + replace(snapshot(), **change)