feat(perception): define stream-first profile and qualification contracts

2026-09-02 00:59:47 +03:00
parent 5a78c997ac
commit 4e04d06191
4 changed files with 1043 additions and 0 deletions
@@ -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)