From 029b486c678c1980313e98bf7cc8221f4ae5083b Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 5 Aug 2026 13:57:27 +0300 Subject: [PATCH] feat(perception): add bounded reference graph --- ...E_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md | 39 +- src/k1link/perception/__init__.py | 36 + src/k1link/perception/baseline.py | 89 ++- src/k1link/perception/graph.py | 659 ++++++++++++++++++ src/k1link/perception/graph_contracts.py | 151 ++++ src/k1link/perception/graph_validation.py | 91 +++ src/k1link/perception/providers.py | 42 +- src/k1link/perception/recorded_source.py | 324 +++++++++ tests/test_perception_architecture.py | 17 +- tests/test_perception_contracts.py | 5 +- tests/test_perception_graph.py | 544 +++++++++++++++ 11 files changed, 1979 insertions(+), 18 deletions(-) create mode 100644 src/k1link/perception/graph.py create mode 100644 src/k1link/perception/graph_contracts.py create mode 100644 src/k1link/perception/graph_validation.py create mode 100644 src/k1link/perception/recorded_source.py create mode 100644 tests/test_perception_graph.py diff --git a/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md b/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md index 9a07c07..6237c5e 100644 --- a/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md +++ b/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md @@ -2,7 +2,7 @@ Date: 2026-08-05 -Status: planned; implementation not started +Status: in progress; M4.0–M4.2 implemented Audit base: `1b3e0b3` on `feat/simulation-polygon-s1` @@ -632,8 +632,9 @@ M4.0 is closed by the executable baseline RAVNOVES00 and the exact source-fusion, E32, E33, E34, E35 and E46J result documents, KB4 calibration, valid-FOV, YOLOX model/config/profile identities and the observed E15 Worker 006 rollback process. `k1link.perception.baseline` -validates the profile and resolves all six immutable evidence documents by file -SHA-256, schema, result and identity. The reusable/historical split is frozen in +validates the profile, the camera summary/index, the source-pack manifest/artifact +and all six immutable evidence documents by file SHA-256, schema, result and +identity. The reusable/historical split is frozen in `config/perception/m4-reuse-inventory-v1.json` and enforced by an import-boundary test. @@ -650,11 +651,33 @@ M4.1 is closed by `k1link.perception.contracts`, `providers` and `adapters`: ownership, range without current qualified points, held-as-current, missing-LiDAR-as-free and physical/command authority in replay. -Validation at this increment: 17 focused M4 tests, 60 related geometry/temporal/ -telemetry tests and the complete Python suite (`1205 passed, 1 skipped`). Scoped -Ruff and strict mypy pass for `src/k1link/perception`. Repository-wide Ruff and -mypy remain red only in pre-existing legacy experiment/web modules; those errors -were not hidden or expanded into this product boundary. +### 2026-08-05 — M4.2 + +M4.2 is closed by `k1link.perception.graph` and `recorded_source`: + +- `ReferencePerceptionGraphV1` composes source, detector, geometry, temporal, + motion and threat providers behind one source-neutral entry point; +- detector, geometry, temporal and threat queues are all capacity-bound and use + explicit latest-wins supersession under overload; +- each admitted sequence closes once as delivered, superseded, stale, rejected, + failed or unavailable; a missing terminal is converted to an observable graph + failure instead of disappearing; +- provider output is cross-validated against the source frame, exact proposal/ + point/component ownership and the no-current-LiDAR/no-metric-occupancy rule; +- run and stage lifecycle reuse the admitted pipeline telemetry emitter; +- restart, cancellation and shutdown are deterministic and leave no graph + threads alive; +- `RecordedRavnoves00Source` reads timing and modality availability directly from + the immutable source LiDAR pack, not from an E/L result wrapper, verifies its + SHA-256, and exposes identical packets at 1.0× or uncapped pace; +- the live adapter is defined only as the same packet-entry protocol and retains + zero physical acceptance in M4. + +Validation at this increment: 26 focused M4 tests, 69 focused-and-related source/ +geometry/temporal/degradation/telemetry tests and the complete Python suite +(`1214 passed, 1 skipped`). Scoped Ruff and strict mypy pass for +`src/k1link/perception`. The canonical GUI remained on `127.0.0.1:8000`; no +duplicate development server or Worker 006 mutation was introduced. ## Implementation order diff --git a/src/k1link/perception/__init__.py b/src/k1link/perception/__init__.py index a1f6a29..815619c 100644 --- a/src/k1link/perception/__init__.py +++ b/src/k1link/perception/__init__.py @@ -49,6 +49,18 @@ from .contracts import ( TimestampBundle, validate_exclusive_point_ownership, ) +from .graph import ( + GRAPH_RESULT_SCHEMA, + REFERENCE_GRAPH_ID, + TERMINAL_OUTCOME_SCHEMA, + DeliveredFrame, + GraphExecutionError, + GraphRunResult, + GraphState, + ReferencePerceptionGraphV1, + TerminalOutcome, + TerminalOutcomeType, +) from .providers import ( REFERENCE_GRAPH_CONFIG_SCHEMA, DetectorProvider, @@ -60,10 +72,18 @@ from .providers import ( ProviderRole, QueuePolicy, ReferencePerceptionGraphConfig, + SourcePacket, SourceProvider, TemporalStateProvider, ThreatProvider, ) +from .recorded_source import ( + LiveSourceAdapter, + RecordedFrameReference, + RecordedRavnoves00Source, + RecordedSourceError, + ReplayPacing, +) __all__ = [ "BASELINE_PROFILE_ID", @@ -119,7 +139,23 @@ __all__ = [ "ProviderRole", "QueuePolicy", "ReferencePerceptionGraphConfig", + "SourcePacket", "SourceProvider", "TemporalStateProvider", "ThreatProvider", + "GRAPH_RESULT_SCHEMA", + "REFERENCE_GRAPH_ID", + "TERMINAL_OUTCOME_SCHEMA", + "DeliveredFrame", + "GraphExecutionError", + "GraphRunResult", + "GraphState", + "ReferencePerceptionGraphV1", + "TerminalOutcome", + "TerminalOutcomeType", + "LiveSourceAdapter", + "RecordedFrameReference", + "RecordedRavnoves00Source", + "RecordedSourceError", + "ReplayPacing", ] diff --git a/src/k1link/perception/baseline.py b/src/k1link/perception/baseline.py index 2adb702..f7a1cb3 100644 --- a/src/k1link/perception/baseline.py +++ b/src/k1link/perception/baseline.py @@ -19,6 +19,14 @@ REUSE_INVENTORY_SCHEMA: Final = "missioncore.perception-reuse-inventory/v1" BASELINE_PROFILE_ID: Final = "m4-ravnoves00-recorded-realtime/v1" BASELINE_SOURCE_ID: Final = "RAVNOVES00" BASELINE_SESSION_ID: Final = "20260720T065719Z_viewer_live" +BASELINE_CAMERA_SOURCE_ID: Final = "sensor.camera.right" +BASELINE_RECORDED_JOB_ID: Final = "recorded-camera-602ac89026ed12978619801d" +BASELINE_SOURCE_PACK_ID: Final = ( + "e10-lidar-pack-576c994a6c814e2592dd6240ace3902a5db94843312c759a73ba0c9166157d2b" +) +BASELINE_SOURCE_PACK_SHA256: Final = ( + "0685d24219d8236caf8b7f1685e93f6d6b59e7fd015a768d88a92bbe8b154944" +) _SHA256 = re.compile(r"^[a-f0-9]{64}$") _EXPECTED_EVIDENCE_ROLES: Final = { @@ -48,6 +56,19 @@ _EVIDENCE_KEYS: Final = { "identity_sha256", "file_sha256", } +_SOURCE_KEYS: Final = { + "source_id", + "session_id", + "camera_source_id", + "recorded_job_id", + "frame_count", + "duration_seconds", + "frame_rate", + "modalities", + "camera_stream_sha256", + "source_pack_id", + "source_pack_artifact_sha256", +} class BaselineContractError(ValueError): @@ -90,10 +111,20 @@ def load_m4_baseline(path: Path) -> BaselineProfile: raise BaselineContractError("baseline profile identity changed") source = _object(document.get("source"), "source") + _exact_keys(source, _SOURCE_KEYS, "source") if source.get("source_id") != BASELINE_SOURCE_ID: raise BaselineContractError("M4 source must remain RAVNOVES00") if source.get("session_id") != BASELINE_SESSION_ID: raise BaselineContractError("M4 source session identity changed") + if source.get("camera_source_id") != BASELINE_CAMERA_SOURCE_ID: + raise BaselineContractError("M4 camera source identity changed") + if source.get("recorded_job_id") != BASELINE_RECORDED_JOB_ID: + raise BaselineContractError("M4 recorded job identity changed") + if source.get("source_pack_id") != BASELINE_SOURCE_PACK_ID: + raise BaselineContractError("M4 source pack identity changed") + if source.get("source_pack_artifact_sha256") != BASELINE_SOURCE_PACK_SHA256: + raise BaselineContractError("M4 source pack artifact identity changed") + _digest(source.get("camera_stream_sha256"), "camera stream digest") modalities = _string_array(source.get("modalities"), "source modalities") if set(modalities) != {"image", "registered-point-increment", "pose"}: raise BaselineContractError("baseline source must bind image, points and pose") @@ -143,7 +174,7 @@ def verify_m4_baseline(repository_root: Path, profile: BaselineProfile) -> Basel """Resolve every immutable evidence document and verify its exact digest.""" root = repository_root.resolve() - verified: list[str] = [] + verified = list(_verify_source_artifacts(root, profile)) for item in profile.evidence: evidence_path = (root / item.relative_path).resolve() if root not in evidence_path.parents: @@ -173,6 +204,62 @@ def verify_m4_baseline(repository_root: Path, profile: BaselineProfile) -> Basel ) +def _verify_source_artifacts(root: Path, profile: BaselineProfile) -> tuple[str, ...]: + source = _object(profile.document.get("source"), "source") + recorded_job_id = _string(source.get("recorded_job_id"), "recorded job id") + camera_source_id = _string(source.get("camera_source_id"), "camera source id") + source_pack_id = _string(source.get("source_pack_id"), "source pack id") + expected_frames = _integer(source.get("frame_count"), "source frame count") + camera_root = ( + root + / ".runtime/compute-jobs" + / recorded_job_id + / "input/camera" + / camera_source_id + / "epoch-1" + ) + summary_path = camera_root / "summary.json" + index_path = camera_root / "index.jsonl" + summary = _read_object(summary_path) + if summary.get("schema_version") != "missioncore.camera-recording/v1": + raise BaselineContractError("camera recording summary schema changed") + if summary.get("source_id") != camera_source_id: + raise BaselineContractError("camera recording source identity changed") + if summary.get("status") != "complete": + raise BaselineContractError("camera recording is incomplete") + if _integer(summary.get("media_segment_count"), "camera segment count") != expected_frames: + raise BaselineContractError("camera recording frame count changed") + if summary.get("stream_sha256") != source.get("camera_stream_sha256"): + raise BaselineContractError("camera recording stream digest changed") + if _file_sha256(index_path) != summary.get("index_sha256"): + raise BaselineContractError("camera recording index digest changed") + + pack_root = root / ".runtime/compute-experiments/e10/lidar-packs" / source_pack_id + pack_manifest_path = pack_root / "manifest.json" + pack_artifact_path = pack_root / "lidar-pack.npz" + pack_manifest = _read_object(pack_manifest_path) + if pack_manifest.get("schema_version") != "missioncore.e10-lidar-replay-pack/v1": + raise BaselineContractError("source pack schema changed") + if pack_manifest.get("pack_id") != source_pack_id: + raise BaselineContractError("source pack identity changed") + identity = _object(pack_manifest.get("identity"), "source pack identity") + if _integer(identity.get("frame_count"), "source pack frame count") != expected_frames: + raise BaselineContractError("source pack frame count changed") + artifact = _object(pack_manifest.get("artifact"), "source pack artifact") + expected_pack_sha = _string( + source.get("source_pack_artifact_sha256"), + "source pack artifact digest", + ) + if artifact.get("sha256") != expected_pack_sha: + raise BaselineContractError("source pack manifest digest changed") + if _file_sha256(pack_artifact_path) != expected_pack_sha: + raise BaselineContractError("source pack artifact digest changed") + return tuple( + str(path.relative_to(root)) + for path in (summary_path, index_path, pack_manifest_path, pack_artifact_path) + ) + + def validate_reuse_inventory(path: Path) -> dict[str, object]: """Validate the M4 primitive/wrapper split used by architecture tests.""" diff --git a/src/k1link/perception/graph.py b/src/k1link/perception/graph.py new file mode 100644 index 0000000..e09c6c8 --- /dev/null +++ b/src/k1link/perception/graph.py @@ -0,0 +1,659 @@ +"""Persistent bounded reference graph for object-centric recorded-realtime CV.""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from contextlib import AbstractContextManager +from dataclasses import dataclass, replace +from queue import Empty, Full, Queue +from threading import Event, Lock, Thread +from typing import TypeVar, cast + +from k1link.compute.pipeline_telemetry import ( + PipelineStageOutcome, + PipelineTelemetryEmitter, + PipelineTelemetryIdentity, + PipelineTelemetrySink, +) + +from .baseline import BASELINE_PROFILE_ID, BASELINE_SOURCE_ID +from .contracts import ( + LocalObstacleMap, + ObjectProposal2D, + ObstacleObservation, + SourceAccounting, + TemporalObstacle, + TemporalState, + validate_exclusive_point_ownership, +) +from .graph_contracts import ( + GRAPH_RESULT_SCHEMA, + REFERENCE_GRAPH_ID, + TERMINAL_OUTCOME_SCHEMA, + DeliveredFrame, + GraphExecutionError, + GraphRunResult, + GraphState, + TerminalOutcome, + TerminalOutcomeType, + build_graph_run_result, +) +from .graph_validation import ( + validate_observations, + validate_proposals, + validate_temporal, + validate_threats, +) +from .providers import ( + DetectorProvider, + GeometryAssociationProvider, + MotionProvider, + ProviderRole, + ReferencePerceptionGraphConfig, + SourcePacket, + SourceProvider, + TemporalStateProvider, + ThreatProvider, +) + + +@dataclass(frozen=True, slots=True) +class _Detected: + packet: SourcePacket + proposals: tuple[ObjectProposal2D, ...] + + +@dataclass(frozen=True, slots=True) +class _Associated: + packet: SourcePacket + proposals: tuple[ObjectProposal2D, ...] + observations: tuple[ObstacleObservation, ...] + + +@dataclass(frozen=True, slots=True) +class _Temporal: + packet: SourcePacket + proposals: tuple[ObjectProposal2D, ...] + associated_proposal_ids: frozenset[str] + obstacles: tuple[TemporalObstacle, ...] + + +@dataclass(frozen=True, slots=True) +class _StopSignal: + pass + + +_STOP = _StopSignal() +_DetectorItem = SourcePacket | _StopSignal +_GeometryItem = _Detected | _StopSignal +_TemporalItem = _Associated | _StopSignal +_ThreatItem = _Temporal | _StopSignal +_QueueItem = SourcePacket | _Detected | _Associated | _Temporal | _StopSignal +_QueueItemT = TypeVar("_QueueItemT", bound=_QueueItem) + + +class ReferencePerceptionGraphV1: + """Run one source through bounded provider stages with exact terminal accounting.""" + + def __init__( + self, + *, + config: ReferencePerceptionGraphConfig, + source: SourceProvider, + detector: DetectorProvider, + geometry: GeometryAssociationProvider, + temporal: TemporalStateProvider, + motion: MotionProvider, + threat: ThreatProvider, + telemetry_identity: PipelineTelemetryIdentity | None = None, + telemetry_sink: PipelineTelemetrySink | None = None, + clock_ns: Callable[[], int] = time.monotonic_ns, + ) -> None: + if config.graph_id != REFERENCE_GRAPH_ID: + raise GraphExecutionError("reference graph id is incompatible") + if config.source_profile_id != BASELINE_PROFILE_ID: + raise GraphExecutionError("reference source profile is incompatible") + if (telemetry_identity is None) is not (telemetry_sink is None): + raise GraphExecutionError("telemetry identity and sink must be configured together") + self.config = config + self.source = source + self.detector = detector + self.geometry = geometry + self.temporal = temporal + self.motion = motion + self.threat = threat + self.telemetry_identity = telemetry_identity + self.telemetry_sink = telemetry_sink + self._clock_ns = clock_ns + self._state = GraphState.CREATED + self._state_lock = Lock() + self._result_lock = Lock() + self._cancel_event = Event() + self._threads: list[Thread] = [] + self._outcomes: dict[int, TerminalOutcome] = {} + self._deliveries: list[DeliveredFrame] = [] + self._admitted: dict[int, SourcePacket] = {} + self._admitted_at_ns: dict[int, int] = {} + self._validate_provider_pins() + + @property + def state(self) -> GraphState: + with self._state_lock: + return self._state + + @property + def worker_threads_alive(self) -> int: + return sum(thread.is_alive() for thread in self._threads) + + def cancel(self) -> None: + self._cancel_event.set() + + def run(self) -> GraphRunResult: + self._begin_run() + started_ns = self._now() + detector_queue: Queue[_DetectorItem] = Queue(self._capacity("detector")) + geometry_queue: Queue[_GeometryItem] = Queue(self._capacity("geometry")) + temporal_queue: Queue[_TemporalItem] = Queue(self._capacity("temporal")) + threat_queue: Queue[_ThreatItem] = Queue(self._capacity("threat")) + self._threads = [ + Thread( + target=self._detector_loop, + args=(detector_queue, geometry_queue), + name="m4-detector", + daemon=True, + ), + Thread( + target=self._geometry_loop, + args=(geometry_queue, temporal_queue), + name="m4-geometry", + daemon=True, + ), + Thread( + target=self._temporal_loop, + args=(temporal_queue, threat_queue), + name="m4-temporal", + daemon=True, + ), + Thread( + target=self._threat_loop, + args=(threat_queue,), + name="m4-threat", + daemon=True, + ), + ] + for thread in self._threads: + thread.start() + run_emitter = self._run_emitter() + if run_emitter is not None: + run_emitter.run("started") + self._set_state(GraphState.RUNNING) + source_failed = False + try: + for packet in self.source.packets(self._cancel_event): + if self._cancel_event.is_set(): + break + if not self._admit(packet): + continue + self._put_latest(detector_queue, packet, "detector") + except Exception: + source_failed = True + finally: + self._set_state(GraphState.STOPPING) + self._put_stop(detector_queue, "detector") + + self._join_workers() + self._close_accounting() + final_state = ( + GraphState.CANCELLED + if self._cancel_event.is_set() + else GraphState.FAILED + if source_failed + else GraphState.STOPPED + ) + self._set_state(final_state) + result = self._result(final_state) + if run_emitter is not None: + duration_ms = max(0.0, (self._now() - started_ns) / 1_000_000) + if final_state is GraphState.STOPPED: + run_emitter.run("completed", duration_ms=duration_ms, exit_code=0) + else: + run_emitter.run( + "failed", + duration_ms=duration_ms, + exit_code=1, + error_type=( + "GraphCancelled" if final_state is GraphState.CANCELLED else "SourceFailure" + ), + ) + return result + + def _begin_run(self) -> None: + with self._state_lock: + if self._state not in { + GraphState.CREATED, + GraphState.STOPPED, + GraphState.CANCELLED, + GraphState.FAILED, + }: + raise GraphExecutionError("reference graph is already running") + self._state = GraphState.STARTING + self._cancel_event = Event() + with self._result_lock: + self._outcomes.clear() + self._deliveries.clear() + self._admitted.clear() + self._admitted_at_ns.clear() + + def _admit(self, packet: SourcePacket) -> bool: + envelope = packet.envelope + with self._result_lock: + if envelope.sequence in self._admitted: + return False + self._admitted[envelope.sequence] = packet + self._admitted_at_ns[envelope.sequence] = self._now() + if envelope.source_id != BASELINE_SOURCE_ID: + self._terminal(packet, TerminalOutcomeType.REJECTED, "source", "source-not-admitted") + return False + if not envelope.image.available: + self._terminal(packet, TerminalOutcomeType.UNAVAILABLE, "source", "image-unavailable") + return False + if envelope.source_age_ns > self._deadline("detector"): + self._terminal(packet, TerminalOutcomeType.STALE, "source", "source-deadline-exceeded") + return False + return True + + def _detector_loop( + self, + incoming: Queue[_DetectorItem], + outgoing: Queue[_GeometryItem], + ) -> None: + while True: + item = incoming.get() + try: + if isinstance(item, _StopSignal): + self._put_stop(outgoing, "geometry") + return + if self._cancelled(item): + continue + if self._expired(item, "detector"): + continue + with self._stage(item, "detector", 1) as stage: + proposals = self.detector.detect(item) + validate_proposals(item, proposals) + if stage is not None: + stage.output_count = len(proposals) + self._put_latest(outgoing, _Detected(item, proposals), "geometry") + except Exception as exc: + if not isinstance(item, _StopSignal): + self._failed(item, "detector", exc) + finally: + incoming.task_done() + + def _geometry_loop( + self, + incoming: Queue[_GeometryItem], + outgoing: Queue[_TemporalItem], + ) -> None: + while True: + item = incoming.get() + try: + if isinstance(item, _StopSignal): + self._put_stop(outgoing, "temporal") + return + if self._cancelled(item.packet): + continue + if self._expired(item.packet, "geometry"): + continue + with self._stage(item.packet, "geometry", len(item.proposals)) as stage: + observations = self.geometry.associate(item.packet, item.proposals) + validate_observations(item.packet, item.proposals, observations) + validate_exclusive_point_ownership(observations) + if stage is not None: + stage.output_count = len(observations) + self._put_latest( + outgoing, + _Associated(item.packet, item.proposals, observations), + "temporal", + ) + except Exception as exc: + if not isinstance(item, _StopSignal): + self._failed(item.packet, "geometry", exc) + finally: + incoming.task_done() + + def _temporal_loop( + self, + incoming: Queue[_TemporalItem], + outgoing: Queue[_ThreatItem], + ) -> None: + while True: + item = incoming.get() + try: + if isinstance(item, _StopSignal): + self._put_stop(outgoing, "threat") + return + if self._cancelled(item.packet): + continue + if self._expired(item.packet, "temporal"): + continue + with self._stage(item.packet, "temporal", len(item.observations)) as stage: + obstacles = self.temporal.update(item.packet, item.observations) + obstacles = self.motion.estimate(item.packet, obstacles) + validate_temporal(item.packet, obstacles) + if stage is not None: + stage.output_count = len(obstacles) + self._put_latest( + outgoing, + _Temporal( + item.packet, + item.proposals, + frozenset( + proposal_id + for observation in item.observations + if observation.occupied_support + for proposal_id in observation.proposal_ids + ), + obstacles, + ), + "threat", + ) + except Exception as exc: + if not isinstance(item, _StopSignal): + self._failed(item.packet, "temporal", exc) + finally: + incoming.task_done() + + def _threat_loop(self, incoming: Queue[_ThreatItem]) -> None: + while True: + item = incoming.get() + try: + if isinstance(item, _StopSignal): + return + if self._cancelled(item.packet): + continue + if self._expired(item.packet, "threat"): + continue + obstacle_map = self._obstacle_map(item) + with self._stage(item.packet, "threat", len(item.obstacles)) as stage: + threats = self.threat.assess(obstacle_map) + validate_threats(obstacle_map, threats) + if stage is not None: + stage.output_count = len(threats) + delivery = DeliveredFrame( + sequence=item.packet.envelope.sequence, + obstacle_map=obstacle_map, + threats=threats, + ) + with self._result_lock: + self._deliveries.append(delivery) + self._terminal( + item.packet, + TerminalOutcomeType.DELIVERED, + "threat", + "object-payload-delivered", + ) + except Exception as exc: + if not isinstance(item, _StopSignal): + self._failed(item.packet, "threat", exc) + finally: + incoming.task_done() + + def _obstacle_map(self, item: _Temporal) -> LocalObstacleMap: + packet = item.packet + now_ns = self._now() + with self._result_lock: + admitted_at_ns = self._admitted_at_ns[packet.envelope.sequence] + occupied = tuple( + obstacle for obstacle in item.obstacles if obstacle.state is TemporalState.CURRENT + ) + unknown = tuple( + obstacle for obstacle in item.obstacles if obstacle.state is not TemporalState.CURRENT + ) + camera_uncertainty = tuple( + proposal + for proposal in item.proposals + if proposal.proposal_id not in item.associated_proposal_ids + ) + return LocalObstacleMap( + source_id=packet.envelope.source_id, + session_id=packet.envelope.session_id, + frame_id=packet.envelope.frame_id, + graph_id=self.config.graph_id, + generated_monotonic_ns=now_ns, + output_age_ns=( + packet.envelope.source_age_ns + max(0, now_ns - admitted_at_ns) + ), + occupied=occupied, + unknown=unknown, + camera_uncertainty=camera_uncertainty, + accounting=SourceAccounting(1, 1, 0, 0), + ) + + def _cancelled(self, packet: SourcePacket) -> bool: + if not self._cancel_event.is_set(): + return False + self._terminal(packet, TerminalOutcomeType.FAILED, "graph", "graph-cancelled") + return True + + def _expired(self, packet: SourcePacket, stage_id: str) -> bool: + with self._result_lock: + admitted_at_ns = self._admitted_at_ns[packet.envelope.sequence] + execution_age_ns = max(0, self._now() - admitted_at_ns) + if packet.envelope.source_age_ns + execution_age_ns <= self._deadline(stage_id): + return False + self._terminal( + packet, + TerminalOutcomeType.STALE, + stage_id, + "stage-deadline-exceeded", + ) + return True + + def _failed(self, packet: SourcePacket, stage_id: str, exc: Exception) -> None: + self._terminal( + packet, + TerminalOutcomeType.FAILED, + stage_id, + f"provider-error-{type(exc).__name__.lower()}", + ) + + def _terminal( + self, + packet: SourcePacket, + outcome: TerminalOutcomeType, + stage_id: str, + reason: str, + ) -> None: + envelope = packet.envelope + terminal = TerminalOutcome( + source_id=envelope.source_id, + session_id=envelope.session_id, + frame_id=envelope.frame_id, + sequence=envelope.sequence, + outcome=outcome, + stage_id=stage_id, + reason=reason, + ) + with self._result_lock: + self._outcomes.setdefault(envelope.sequence, terminal) + + def _put_latest( + self, + queue: Queue[_QueueItemT], + item: _QueueItemT, + stage_id: str, + ) -> None: + while True: + try: + queue.put_nowait(item) + return + except Full: + pass + try: + dropped = queue.get_nowait() + except Empty: + continue + try: + if not isinstance(dropped, _StopSignal): + packet = _packet_from_item(dropped) + self._terminal( + packet, + TerminalOutcomeType.SUPERSEDED, + stage_id, + "bounded-queue-latest-wins", + ) + finally: + queue.task_done() + + def _put_stop(self, queue: Queue[_QueueItemT], stage_id: str) -> None: + timeout = self._terminal_timeout(stage_id) / 1_000_000_000 + try: + queue.put(cast(_QueueItemT, _STOP), timeout=timeout) + except Full: + self._put_latest(queue, cast(_QueueItemT, _STOP), stage_id) + + def _join_workers(self) -> None: + timeout_seconds = max( + 1.0, + max(policy.terminal_timeout_ns for policy in self.config.queues) + / 1_000_000_000 + * 2, + ) + for thread in self._threads: + thread.join(timeout_seconds) + if self.worker_threads_alive: + self._set_state(GraphState.FAILED) + raise GraphExecutionError("reference graph left an orphan worker thread") + + def _close_accounting(self) -> None: + with self._result_lock: + missing = sorted(set(self._admitted) - set(self._outcomes)) + packets = [self._admitted[sequence] for sequence in missing] + for packet in packets: + self._terminal( + packet, + TerminalOutcomeType.FAILED, + "graph", + "terminal-accounting-gap", + ) + with self._result_lock: + if set(self._admitted) != set(self._outcomes): + raise GraphExecutionError("terminal accounting did not close") + + def _result(self, state: GraphState) -> GraphRunResult: + with self._result_lock: + outcomes = tuple(self._outcomes[key] for key in sorted(self._outcomes)) + deliveries = tuple(sorted(self._deliveries, key=lambda item: item.sequence)) + admitted_count = len(self._admitted) + return build_graph_run_result( + graph_id=self.config.graph_id, + source_profile_id=self.config.source_profile_id, + state=state, + admitted_count=admitted_count, + outcomes=outcomes, + deliveries=deliveries, + ) + + def _stage( + self, + packet: SourcePacket, + stage_id: str, + input_count: int, + ) -> AbstractContextManager[PipelineStageOutcome | _NullStage]: + emitter = self._frame_emitter(packet) + if emitter is None: + return _NullStage() + return emitter.stage(stage_id, input_count=input_count) + + def _run_emitter(self) -> PipelineTelemetryEmitter | None: + if self.telemetry_identity is None or self.telemetry_sink is None: + return None + return PipelineTelemetryEmitter( + identity=self.telemetry_identity, + sink=self.telemetry_sink, + clock_ns=self._clock_ns, + ) + + def _frame_emitter(self, packet: SourcePacket) -> PipelineTelemetryEmitter | None: + if self.telemetry_identity is None or self.telemetry_sink is None: + return None + identity = replace( + self.telemetry_identity, + request_id=packet.envelope.frame_id, + frame_index=packet.envelope.sequence, + ) + return PipelineTelemetryEmitter( + identity=identity, + sink=self.telemetry_sink, + clock_ns=self._clock_ns, + ) + + def _validate_provider_pins(self) -> None: + actual = { + ProviderRole.SOURCE: self.source.provider_id, + ProviderRole.DETECTOR: self.detector.provider_id, + ProviderRole.GEOMETRY: self.geometry.provider_id, + ProviderRole.TEMPORAL: self.temporal.provider_id, + ProviderRole.MOTION: self.motion.provider_id, + ProviderRole.THREAT: self.threat.provider_id, + } + pinned = {pin.role: pin.provider_id for pin in self.config.providers} + if actual != pinned: + raise GraphExecutionError("configured provider identities do not match runtime") + + def _capacity(self, stage_id: str) -> int: + return next(policy.capacity for policy in self.config.queues if policy.stage_id == stage_id) + + def _deadline(self, stage_id: str) -> int: + return next( + policy.deadline_ns + for policy in self.config.queues + if policy.stage_id == stage_id + ) + + def _terminal_timeout(self, stage_id: str) -> int: + return next( + policy.terminal_timeout_ns + for policy in self.config.queues + if policy.stage_id == stage_id + ) + + def _now(self) -> int: + return int(self._clock_ns()) + + def _set_state(self, state: GraphState) -> None: + with self._state_lock: + self._state = state + + +class _NullStage: + output_count: int | None = None + + def __enter__(self) -> _NullStage: + return self + + def __exit__(self, *args: object) -> None: + return None + + +def _packet_from_item(item: object) -> SourcePacket: + if isinstance(item, SourcePacket): + return item + if isinstance(item, (_Detected, _Associated, _Temporal)): + return item.packet + raise GraphExecutionError("queue contained an incompatible item") + + +__all__ = [ + "GRAPH_RESULT_SCHEMA", + "REFERENCE_GRAPH_ID", + "TERMINAL_OUTCOME_SCHEMA", + "DeliveredFrame", + "GraphExecutionError", + "GraphRunResult", + "GraphState", + "ReferencePerceptionGraphV1", + "TerminalOutcome", + "TerminalOutcomeType", +] diff --git a/src/k1link/perception/graph_contracts.py b/src/k1link/perception/graph_contracts.py new file mode 100644 index 0000000..cd52eb5 --- /dev/null +++ b/src/k1link/perception/graph_contracts.py @@ -0,0 +1,151 @@ +"""Immutable lifecycle and result documents for the reference perception graph.""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from enum import StrEnum +from typing import Final + +from .contracts import LocalObstacleMap, ThreatAssessment + +GRAPH_RESULT_SCHEMA: Final = "missioncore.reference-perception-graph-result/v1" +TERMINAL_OUTCOME_SCHEMA: Final = "missioncore.perception-terminal-outcome/v1" +REFERENCE_GRAPH_ID: Final = "reference-perception-graph/v1" +_SAFE_IDENTIFIER: Final = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$") + + +class GraphExecutionError(RuntimeError): + """The graph cannot preserve its bounded lifecycle or product accounting.""" + + +class GraphState(StrEnum): + CREATED = "created" + STARTING = "starting" + RUNNING = "running" + STOPPING = "stopping" + STOPPED = "stopped" + CANCELLED = "cancelled" + FAILED = "failed" + + +class TerminalOutcomeType(StrEnum): + DELIVERED = "delivered" + SUPERSEDED = "superseded" + STALE = "stale" + REJECTED = "rejected" + FAILED = "failed" + UNAVAILABLE = "unavailable" + + +@dataclass(frozen=True, slots=True) +class TerminalOutcome: + source_id: str + session_id: str + frame_id: str + sequence: int + outcome: TerminalOutcomeType + stage_id: str + reason: str + + def __post_init__(self) -> None: + for value in (self.source_id, self.session_id, self.frame_id, self.stage_id, self.reason): + if _SAFE_IDENTIFIER.fullmatch(value) is None: + raise GraphExecutionError("terminal outcome contains an unsafe identifier") + if self.sequence < 0: + raise GraphExecutionError("terminal sequence must be nonnegative") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": TERMINAL_OUTCOME_SCHEMA, + "source_id": self.source_id, + "session_id": self.session_id, + "frame_id": self.frame_id, + "sequence": self.sequence, + "outcome": self.outcome.value, + "stage_id": self.stage_id, + "reason": self.reason, + } + + +@dataclass(frozen=True, slots=True) +class DeliveredFrame: + sequence: int + obstacle_map: LocalObstacleMap + threats: tuple[ThreatAssessment, ...] + + def canonical_dict(self) -> dict[str, object]: + obstacle_map = self.obstacle_map.to_dict() + obstacle_map.pop("generated_monotonic_ns") + obstacle_map.pop("output_age_ns") + return { + "sequence": self.sequence, + "obstacle_map": obstacle_map, + "threats": [item.to_dict() for item in self.threats], + } + + +@dataclass(frozen=True, slots=True) +class GraphRunResult: + graph_id: str + source_profile_id: str + state: GraphState + admitted_count: int + terminal_outcomes: tuple[TerminalOutcome, ...] + deliveries: tuple[DeliveredFrame, ...] + canonical_payload_sha256: str + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": GRAPH_RESULT_SCHEMA, + "graph_id": self.graph_id, + "source_profile_id": self.source_profile_id, + "state": self.state.value, + "admitted_count": self.admitted_count, + "terminal_outcomes": [item.to_dict() for item in self.terminal_outcomes], + "deliveries": [item.canonical_dict() for item in self.deliveries], + "canonical_payload_sha256": self.canonical_payload_sha256, + } + + +def build_graph_run_result( + *, + graph_id: str, + source_profile_id: str, + state: GraphState, + admitted_count: int, + outcomes: tuple[TerminalOutcome, ...], + deliveries: tuple[DeliveredFrame, ...], +) -> GraphRunResult: + canonical = { + "graph_id": graph_id, + "source_profile_id": source_profile_id, + "terminal_outcomes": [item.to_dict() for item in outcomes], + "deliveries": [item.canonical_dict() for item in deliveries], + } + payload = json.dumps(canonical, sort_keys=True, separators=(",", ":")).encode() + return GraphRunResult( + graph_id=graph_id, + source_profile_id=source_profile_id, + state=state, + admitted_count=admitted_count, + terminal_outcomes=outcomes, + deliveries=deliveries, + canonical_payload_sha256=hashlib.sha256(payload).hexdigest(), + ) + + +__all__ = [ + "GRAPH_RESULT_SCHEMA", + "REFERENCE_GRAPH_ID", + "TERMINAL_OUTCOME_SCHEMA", + "DeliveredFrame", + "GraphExecutionError", + "GraphRunResult", + "GraphState", + "TerminalOutcome", + "TerminalOutcomeType", + "build_graph_run_result", +] diff --git a/src/k1link/perception/graph_validation.py b/src/k1link/perception/graph_validation.py new file mode 100644 index 0000000..f376927 --- /dev/null +++ b/src/k1link/perception/graph_validation.py @@ -0,0 +1,91 @@ +"""Cross-provider fail-closed validation for the reference perception graph.""" + +from __future__ import annotations + +from .contracts import ( + LocalObstacleMap, + ObjectProposal2D, + ObstacleObservation, + TemporalObstacle, + TemporalState, + ThreatAssessment, +) +from .graph_contracts import GraphExecutionError +from .providers import SourcePacket + + +def validate_proposals( + packet: SourcePacket, + proposals: tuple[ObjectProposal2D, ...], +) -> None: + envelope = packet.envelope + proposal_ids = [proposal.proposal_id for proposal in proposals] + if len(set(proposal_ids)) != len(proposal_ids): + raise GraphExecutionError("detector proposal identities are duplicated") + if any( + proposal.source_id != envelope.source_id or proposal.frame_id != envelope.frame_id + for proposal in proposals + ): + raise GraphExecutionError("detector proposal escaped its source frame") + + +def validate_observations( + packet: SourcePacket, + proposals: tuple[ObjectProposal2D, ...], + observations: tuple[ObstacleObservation, ...], +) -> None: + envelope = packet.envelope + proposal_ids = {proposal.proposal_id for proposal in proposals} + observation_ids = [observation.observation_id for observation in observations] + occupancy_keys = [observation.occupancy_key for observation in observations] + if len(set(observation_ids)) != len(observation_ids): + raise GraphExecutionError("geometry observation identities are duplicated") + if len(set(occupancy_keys)) != len(occupancy_keys): + raise GraphExecutionError("geometry occupancy identities are duplicated") + if any( + observation.source_id != envelope.source_id + or observation.frame_id != envelope.frame_id + or not set(observation.proposal_ids).issubset(proposal_ids) + for observation in observations + ): + raise GraphExecutionError("geometry observation escaped its source evidence") + if not envelope.registered_point_increment.available and any( + observation.occupied_support or observation.metric_geometry is not None + for observation in observations + ): + raise GraphExecutionError("metric occupancy was invented without current LiDAR") + + +def validate_temporal( + packet: SourcePacket, + obstacles: tuple[TemporalObstacle, ...], +) -> None: + component_ids = [obstacle.component_id for obstacle in obstacles] + if len(set(component_ids)) != len(component_ids): + raise GraphExecutionError("temporal component identities are duplicated") + if not packet.envelope.registered_point_increment.available and any( + obstacle.state is TemporalState.CURRENT for obstacle in obstacles + ): + raise GraphExecutionError("current temporal occupancy was invented without LiDAR") + + +def validate_threats( + obstacle_map: LocalObstacleMap, + threats: tuple[ThreatAssessment, ...], +) -> None: + component_ids = { + obstacle.component_id for obstacle in (*obstacle_map.occupied, *obstacle_map.unknown) + } + assessment_ids = [threat.assessment_id for threat in threats] + if len(set(assessment_ids)) != len(assessment_ids): + raise GraphExecutionError("threat assessment identities are duplicated") + if any(threat.component_id not in component_ids for threat in threats): + raise GraphExecutionError("threat assessment references an unknown component") + + +__all__ = [ + "validate_observations", + "validate_proposals", + "validate_temporal", + "validate_threats", +] diff --git a/src/k1link/perception/providers.py b/src/k1link/perception/providers.py index 8a3702f..2024e77 100644 --- a/src/k1link/perception/providers.py +++ b/src/k1link/perception/providers.py @@ -6,6 +6,7 @@ import re from collections.abc import Iterator from dataclasses import dataclass from enum import StrEnum +from threading import Event from typing import Final, Protocol from .contracts import ( @@ -18,6 +19,7 @@ from .contracts import ( ) REFERENCE_GRAPH_CONFIG_SCHEMA: Final = "missioncore.reference-perception-graph-config/v1" +REFERENCE_GRAPH_STAGE_IDS: Final = frozenset({"detector", "geometry", "temporal", "threat"}) _SHA256 = re.compile(r"^[a-f0-9]{64}$") _IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$") @@ -35,6 +37,30 @@ class ProviderRole(StrEnum): THREAT = "threat" +@dataclass(frozen=True, slots=True) +class SourcePacket: + """Execution-only carrier; the graph never interprets opaque sensor payloads.""" + + envelope: SourceEnvelope + image_payload: object | None + registered_point_increment_payload: object | None + pose_payload: object | None + + def __post_init__(self) -> None: + bindings = ( + (self.envelope.image.available, self.image_payload, "image"), + ( + self.envelope.registered_point_increment.available, + self.registered_point_increment_payload, + "registered point increment", + ), + (self.envelope.pose.available, self.pose_payload, "pose"), + ) + for available, payload, label in bindings: + if available is not (payload is not None): + raise ProviderContractError(f"{label} availability and payload disagree") + + @dataclass(frozen=True, slots=True) class ProviderPin: role: ProviderRole @@ -188,8 +214,10 @@ class ReferencePerceptionGraphConfig: if len(set(roles)) != len(roles) or set(roles) != set(ProviderRole): raise ProviderContractError("graph must pin each provider role exactly once") stage_ids = [queue.stage_id for queue in self.queues] - if not stage_ids or len(set(stage_ids)) != len(stage_ids): - raise ProviderContractError("graph queue policies must be nonempty and unique") + if len(set(stage_ids)) != len(stage_ids): + raise ProviderContractError("graph queue policies must be unique") + if set(stage_ids) != REFERENCE_GRAPH_STAGE_IDS: + raise ProviderContractError("graph must bound each reference stage exactly once") def to_dict(self) -> dict[str, object]: return { @@ -223,13 +251,13 @@ class ReferencePerceptionGraphConfig: class SourceProvider(Protocol): provider_id: str - def envelopes(self) -> Iterator[SourceEnvelope]: ... + def packets(self, stop_event: Event) -> Iterator[SourcePacket]: ... class DetectorProvider(Protocol): provider_id: str - def detect(self, envelope: SourceEnvelope) -> tuple[ObjectProposal2D, ...]: ... + def detect(self, packet: SourcePacket) -> tuple[ObjectProposal2D, ...]: ... class GeometryAssociationProvider(Protocol): @@ -237,7 +265,7 @@ class GeometryAssociationProvider(Protocol): def associate( self, - envelope: SourceEnvelope, + packet: SourcePacket, proposals: tuple[ObjectProposal2D, ...], ) -> tuple[ObstacleObservation, ...]: ... @@ -247,7 +275,7 @@ class TemporalStateProvider(Protocol): def update( self, - envelope: SourceEnvelope, + packet: SourcePacket, observations: tuple[ObstacleObservation, ...], ) -> tuple[TemporalObstacle, ...]: ... @@ -257,7 +285,7 @@ class MotionProvider(Protocol): def estimate( self, - envelope: SourceEnvelope, + packet: SourcePacket, obstacles: tuple[TemporalObstacle, ...], ) -> tuple[TemporalObstacle, ...]: ... diff --git a/src/k1link/perception/recorded_source.py b/src/k1link/perception/recorded_source.py new file mode 100644 index 0000000..7dae041 --- /dev/null +++ b/src/k1link/perception/recorded_source.py @@ -0,0 +1,324 @@ +"""Recorded-realtime source adapter for the admitted RAVNOVES00 baseline.""" + +from __future__ import annotations + +import hashlib +import json +import time +from collections.abc import Callable, Iterator +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path +from threading import Event +from typing import Final, Protocol + +import numpy as np + +from .baseline import ( + BASELINE_PROFILE_ID, + BASELINE_RECORDED_JOB_ID, + BASELINE_SESSION_ID, + BASELINE_SOURCE_ID, + BASELINE_SOURCE_PACK_ID, + BASELINE_SOURCE_PACK_SHA256, +) +from .contracts import ( + ClockBasis, + ModalityOutcome, + ModalityStatus, + SourceEnvelope, + TimestampBundle, +) +from .providers import SourcePacket + +RECORDED_SOURCE_PROVIDER_ID: Final = "ravnoves00-recorded-source/v1" +RECORDED_CAMERA_JOB_ID: Final = BASELINE_RECORDED_JOB_ID +RECORDED_SOURCE_PACK_ID: Final = BASELINE_SOURCE_PACK_ID +RECORDED_SOURCE_PACK_SHA256: Final = BASELINE_SOURCE_PACK_SHA256 +RECORDED_CALIBRATION_ID: Final = "camera-1-kb4-05f3ad9b" +RECORDED_REPRESENTATION_ID: Final = "registered-map-increment-v1" +CAMERA_INDEX_SCHEMA: Final = "missioncore.camera-recording-index/v1" +DEFAULT_FRAME_COUNT: Final = 4489 + + +class RecordedSourceError(RuntimeError): + """The immutable replay source is incomplete, mismatched or ambiguous.""" + + +class ReplayPacing(StrEnum): + ONE_X = "1.0x" + UNCAPPED = "uncapped" + + +@dataclass(frozen=True, slots=True) +class RecordedFrameReference: + """Opaque reference passed to a provider without decoding sensor data.""" + + artifact_id: str + frame_index: int + content_sha256: str | None = None + + +@dataclass(frozen=True, slots=True) +class _SourceTimelineRow: + frame_index: int + source_frame_index: int + session_seconds: float + source_available: bool + + +class LiveSourceAdapter(Protocol): + """Future live source boundary; M4 does not physically accept this input.""" + + provider_id: str + + def packets(self, stop_event: Event) -> Iterator[SourcePacket]: ... + + +WaitFunction = Callable[[Event, float], bool] + + +class RecordedRavnoves00Source: + """Emit the admitted synchronized source timeline at 1.0x or uncapped speed.""" + + provider_id: str = RECORDED_SOURCE_PROVIDER_ID + + def __init__( + self, + *, + camera_index_path: Path, + source_pack_path: Path, + pacing: ReplayPacing = ReplayPacing.UNCAPPED, + expected_frame_count: int = DEFAULT_FRAME_COUNT, + expected_source_pack_sha256: str | None = RECORDED_SOURCE_PACK_SHA256, + clock_ns: Callable[[], int] = time.monotonic_ns, + wait: WaitFunction | None = None, + ) -> None: + if expected_frame_count < 1: + raise RecordedSourceError("expected frame count must be positive") + self.camera_index_path = camera_index_path.resolve() + self.source_pack_path = source_pack_path.resolve() + self.pacing = pacing + self.expected_frame_count = expected_frame_count + self.expected_source_pack_sha256 = expected_source_pack_sha256 + self._clock_ns = clock_ns + self._wait = wait or _event_wait + + @classmethod + def from_repository( + cls, + repository_root: Path, + *, + pacing: ReplayPacing = ReplayPacing.UNCAPPED, + ) -> RecordedRavnoves00Source: + root = repository_root.resolve() + camera_index = ( + root + / ".runtime/compute-jobs" + / RECORDED_CAMERA_JOB_ID + / "input/camera/sensor.camera.right/epoch-1/index.jsonl" + ) + source_pack = ( + root + / ".runtime/compute-experiments/e10/lidar-packs" + / RECORDED_SOURCE_PACK_ID + / "lidar-pack.npz" + ) + return cls( + camera_index_path=camera_index, + source_pack_path=source_pack, + pacing=pacing, + ) + + def packets(self, stop_event: Event) -> Iterator[SourcePacket]: + camera_rows = _read_jsonl(self.camera_index_path, "camera index") + timeline_rows = _read_source_timeline( + self.source_pack_path, + expected_sha256=self.expected_source_pack_sha256, + ) + if len(camera_rows) != self.expected_frame_count: + raise RecordedSourceError("camera index frame count changed") + if len(timeline_rows) != self.expected_frame_count: + raise RecordedSourceError("synchronized timeline frame count changed") + + started_ns = int(self._clock_ns()) + source_origin_ns = _source_time_ns(timeline_rows[0]) + for frame_index, (camera, timeline) in enumerate( + zip(camera_rows, timeline_rows, strict=True) + ): + if stop_event.is_set(): + return + packet = _packet(frame_index, camera, timeline) + if self.pacing is ReplayPacing.ONE_X: + target_ns = started_ns + packet.envelope.timestamps.source_ns - source_origin_ns + if not self._pace_until(stop_event, target_ns): + return + yield packet + + def _pace_until(self, stop_event: Event, target_ns: int) -> bool: + while not stop_event.is_set(): + remaining_ns = target_ns - int(self._clock_ns()) + if remaining_ns <= 0: + return True + if self._wait(stop_event, remaining_ns / 1_000_000_000): + return False + return False + + +def _packet( + frame_index: int, + camera: dict[str, object], + timeline: _SourceTimelineRow, +) -> SourcePacket: + if camera.get("schema_version") != CAMERA_INDEX_SCHEMA or camera.get("kind") != "media": + raise RecordedSourceError("camera index row is incompatible") + if _integer(camera, "sequence") != frame_index + 1: + raise RecordedSourceError("camera sequence is not contiguous") + if timeline.frame_index != frame_index: + raise RecordedSourceError("timeline frame index is not contiguous") + if timeline.source_frame_index != frame_index: + raise RecordedSourceError("source and timeline frame indices disagree") + + source_available = timeline.source_available + camera_sha = camera.get("sha256") + if not isinstance(camera_sha, str) or len(camera_sha) != 64: + raise RecordedSourceError("camera content digest is invalid") + status = _available_status() if source_available else _unavailable_status() + envelope = SourceEnvelope( + source_id=BASELINE_SOURCE_ID, + session_id=BASELINE_SESSION_ID, + frame_id=f"frame-{frame_index:06d}", + sequence=frame_index, + timestamps=TimestampBundle( + utc_ns=_integer(camera, "host_epoch_ns"), + monotonic_ns=_integer(camera, "host_monotonic_ns"), + source_ns=_source_time_ns(timeline), + clock_basis=ClockBasis.RECORDED_HOST, + ), + source_age_ns=0, + binding_reason="exact-ravnoves00-synchronized-recording", + calibration_id=RECORDED_CALIBRATION_ID, + representation_id=RECORDED_REPRESENTATION_ID, + image=_available_status(), + registered_point_increment=status, + pose=status, + ) + source_reference = ( + RecordedFrameReference(RECORDED_SOURCE_PACK_ID, frame_index) + if source_available + else None + ) + return SourcePacket( + envelope=envelope, + image_payload=RecordedFrameReference( + artifact_id=RECORDED_CAMERA_JOB_ID, + frame_index=frame_index, + content_sha256=camera_sha, + ), + registered_point_increment_payload=source_reference, + pose_payload=source_reference, + ) + + +def _available_status() -> ModalityStatus: + return ModalityStatus(True, ModalityOutcome.AVAILABLE, "recorded-source-available") + + +def _unavailable_status() -> ModalityStatus: + return ModalityStatus(False, ModalityOutcome.UNAVAILABLE, "recorded-source-unavailable") + + +def _source_time_ns(document: _SourceTimelineRow) -> int: + value = document.session_seconds + if not np.isfinite(value) or value < 0: + raise RecordedSourceError("source session time is invalid") + return round(value * 1_000_000_000) + + +def _integer(document: dict[str, object], key: str) -> int: + value = document.get(key) + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise RecordedSourceError(f"{key} must be a nonnegative integer") + return value + + +def _read_jsonl(path: Path, label: str) -> list[dict[str, object]]: + if not path.is_file(): + raise RecordedSourceError(f"{label} is missing") + documents: list[dict[str, object]] = [] + for line_number, line in enumerate(path.read_text("utf-8").splitlines(), start=1): + try: + value = json.loads(line) + except json.JSONDecodeError as exc: + raise RecordedSourceError(f"{label} line {line_number} is invalid JSON") from exc + if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): + raise RecordedSourceError(f"{label} line {line_number} is not an object") + documents.append(value) + return documents + + +def _read_source_timeline( + path: Path, + *, + expected_sha256: str | None, +) -> tuple[_SourceTimelineRow, ...]: + if not path.is_file(): + raise RecordedSourceError("source pack is missing") + if expected_sha256 is not None and _file_sha256(path) != expected_sha256: + raise RecordedSourceError("source pack digest changed") + required = { + "frame_indices", + "source_frame_indices", + "session_seconds", + "sample_available", + } + try: + with np.load(path, allow_pickle=False) as archive: + if not required.issubset(archive.files): + raise RecordedSourceError("source pack timeline arrays are incomplete") + frame_indices = np.asarray(archive["frame_indices"]) + source_frame_indices = np.asarray(archive["source_frame_indices"]) + session_seconds = np.asarray(archive["session_seconds"]) + sample_available = np.asarray(archive["sample_available"]) + except (OSError, ValueError) as exc: + raise RecordedSourceError("source pack cannot be opened") from exc + shapes = { + frame_indices.shape, + source_frame_indices.shape, + session_seconds.shape, + sample_available.shape, + } + if len(shapes) != 1 or len(frame_indices.shape) != 1: + raise RecordedSourceError("source pack timeline shapes disagree") + return tuple( + _SourceTimelineRow( + frame_index=int(frame_indices[index]), + source_frame_index=int(source_frame_indices[index]), + session_seconds=float(session_seconds[index]), + source_available=bool(sample_available[index]), + ) + for index in range(frame_indices.shape[0]) + ) + + +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _event_wait(stop_event: Event, timeout_seconds: float) -> bool: + return stop_event.wait(timeout_seconds) + + +__all__ = [ + "BASELINE_PROFILE_ID", + "LiveSourceAdapter", + "RECORDED_SOURCE_PROVIDER_ID", + "RecordedFrameReference", + "RecordedRavnoves00Source", + "RecordedSourceError", + "ReplayPacing", +] diff --git a/tests/test_perception_architecture.py b/tests/test_perception_architecture.py index 1c81dee..548a44c 100644 --- a/tests/test_perception_architecture.py +++ b/tests/test_perception_architecture.py @@ -36,7 +36,7 @@ def test_m4_baseline_is_exact_and_every_local_evidence_digest_resolves() -> None verification = verify_m4_baseline(REPOSITORY_ROOT, profile) assert verification.source_id == "RAVNOVES00" assert verification.session_id == "20260720T065719Z_viewer_live" - assert len(verification.verified_paths) == 6 + assert len(verification.verified_paths) == 10 def test_m4_baseline_cannot_silently_select_another_source(tmp_path: Path) -> None: @@ -82,3 +82,18 @@ def test_new_perception_boundary_has_no_experiment_specific_imports() -> None: ): violations[path.name] = module assert violations == {} + + +def test_reference_graph_imports_only_an_admitted_compute_primitive() -> None: + inventory = validate_reuse_inventory(REUSE_PATH) + admitted = { + item["module"] + for item in inventory["reusable_primitives"] + if isinstance(item, dict) and isinstance(item.get("module"), str) + } + compute_imports = { + module + for module in _imports(PERCEPTION_ROOT / "graph.py") + if module.startswith("k1link.compute") + } + assert compute_imports <= admitted diff --git a/tests/test_perception_contracts.py b/tests/test_perception_contracts.py index d2d907c..1c73013 100644 --- a/tests/test_perception_contracts.py +++ b/tests/test_perception_contracts.py @@ -321,7 +321,10 @@ def test_reference_graph_config_pins_all_roles_and_queue_bounds() -> None: ProviderPin(role, f"{role.value}-provider", "v1", "78a3dc2", "a" * 64) for role in ProviderRole ), - queues=(QueuePolicy("detector", 2, 80_000_000, 200_000_000),), + queues=tuple( + QueuePolicy(stage, 2, 80_000_000, 200_000_000) + for stage in ("detector", "geometry", "temporal", "threat") + ), authority=GraphAuthority(), ) assert ReferencePerceptionGraphConfig.from_dict(config.to_dict()) == config diff --git a/tests/test_perception_graph.py b/tests/test_perception_graph.py new file mode 100644 index 0000000..2164bad --- /dev/null +++ b/tests/test_perception_graph.py @@ -0,0 +1,544 @@ +from __future__ import annotations + +import json +import threading +from collections.abc import Iterator +from dataclasses import replace +from pathlib import Path +from threading import Event + +import numpy as np +import pytest + +from k1link.compute.pipeline_telemetry import PipelineTelemetryIdentity +from k1link.perception.baseline import BASELINE_PROFILE_ID +from k1link.perception.contracts import ( + BoundingRegion2D, + ClockBasis, + CorridorIntersection, + EvidenceBasis, + EvidenceCurrentness, + GridCell, + HistorySample, + LocalObstacleMap, + MetricGeometry, + ModalityOutcome, + ModalityStatus, + MotionState, + ObjectProposal2D, + ObstacleObservation, + QualificationState, + SourceEnvelope, + TemporalObstacle, + TemporalState, + ThreatAssessment, + ThreatDecision, + TimestampBundle, +) +from k1link.perception.graph import ( + GraphRunResult, + GraphState, + ReferencePerceptionGraphV1, + TerminalOutcomeType, +) +from k1link.perception.providers import ( + GraphAuthority, + ProviderPin, + ProviderRole, + QueuePolicy, + ReferencePerceptionGraphConfig, + SourcePacket, +) +from k1link.perception.recorded_source import ( + RecordedRavnoves00Source, + RecordedSourceError, + ReplayPacing, +) + + +def _status(available: bool = True) -> ModalityStatus: + return ModalityStatus( + available=available, + outcome=ModalityOutcome.AVAILABLE if available else ModalityOutcome.UNAVAILABLE, + reason="test-available" if available else "test-unavailable", + ) + + +def _packet( + sequence: int, + *, + image: bool = True, + lidar: bool = True, + source_age_ns: int = 0, +) -> SourcePacket: + envelope = SourceEnvelope( + source_id="RAVNOVES00", + session_id="20260720T065719Z_viewer_live", + frame_id=f"frame-{sequence:06d}", + sequence=sequence, + timestamps=TimestampBundle( + utc_ns=1_786_000_000_000_000_000 + sequence, + monotonic_ns=1_000 + sequence, + source_ns=35_000_000_000 + sequence * 100_000_000, + clock_basis=ClockBasis.RECORDED_HOST, + ), + source_age_ns=source_age_ns, + binding_reason="test-recorded-source", + calibration_id="camera-1-kb4-test", + representation_id="registered-map-increment-v1", + image=_status(image), + registered_point_increment=_status(lidar), + pose=_status(lidar), + ) + return SourcePacket( + envelope=envelope, + image_payload=("image", sequence) if image else None, + registered_point_increment_payload=("points", sequence) if lidar else None, + pose_payload=("pose", sequence) if lidar else None, + ) + + +class _Source: + provider_id = "test-source/v1" + + def __init__(self, packets: tuple[SourcePacket, ...]) -> None: + self.values = packets + + def packets(self, stop_event: Event) -> Iterator[SourcePacket]: + for packet in self.values: + if stop_event.is_set(): + return + yield packet + + +class _Detector: + provider_id = "test-detector/v1" + + def detect(self, packet: SourcePacket) -> tuple[ObjectProposal2D, ...]: + sequence = packet.envelope.sequence + return ( + ObjectProposal2D( + proposal_id=f"proposal-{sequence}", + source_id=packet.envelope.source_id, + frame_id=packet.envelope.frame_id, + region=BoundingRegion2D(1.0, 2.0, 20.0, 30.0), + objectness=0.9, + provider_id=self.provider_id, + model_id="objectness-test/v1", + preprocess_id="kb4-test/v1", + ), + ) + + +class _Geometry: + provider_id = "test-geometry/v1" + + def associate( + self, + packet: SourcePacket, + proposals: tuple[ObjectProposal2D, ...], + ) -> tuple[ObstacleObservation, ...]: + sequence = packet.envelope.sequence + if not packet.envelope.registered_point_increment.available: + return ( + ObstacleObservation( + observation_id=f"camera-observation-{sequence}", + occupancy_key=f"camera-uncertainty-{sequence}", + source_id=packet.envelope.source_id, + frame_id=packet.envelope.frame_id, + evidence_time_ns=packet.envelope.timestamps.source_ns, + basis=EvidenceBasis.CAMERA, + currentness=EvidenceCurrentness.CURRENT, + occupied_support=False, + source_point_ids=(), + metric_geometry=None, + proposal_ids=(proposals[0].proposal_id,), + semantic_hint=None, + reason_codes=("camera-only",), + ), + ) + return ( + ObstacleObservation( + observation_id=f"observation-{sequence}", + occupancy_key=f"occupancy-{sequence}", + source_id=packet.envelope.source_id, + frame_id=packet.envelope.frame_id, + evidence_time_ns=packet.envelope.timestamps.source_ns, + basis=EvidenceBasis.FUSED, + currentness=EvidenceCurrentness.CURRENT, + occupied_support=True, + source_point_ids=(sequence,), + metric_geometry=MetricGeometry( + coordinate_frame="map", + centroid_xyz_m=(4.0, 1.0, 0.5), + range_m=4.15, + covariance_diagonal_m2=(0.04, 0.04, 0.09), + ), + proposal_ids=(proposals[0].proposal_id,), + semantic_hint=None, + reason_codes=("current-qualified-points",), + ), + ) + + +class _Temporal: + provider_id = "test-temporal/v1" + + def update( + self, + packet: SourcePacket, + observations: tuple[ObstacleObservation, ...], + ) -> tuple[TemporalObstacle, ...]: + if not packet.envelope.registered_point_increment.available: + return () + sequence = packet.envelope.sequence + return ( + TemporalObstacle( + component_id=f"component-{sequence}", + identity_scope="ephemeral", + state=TemporalState.CURRENT, + ttl_ns=750_000_000, + last_hit_ns=packet.envelope.timestamps.source_ns, + age_ns=0, + association_basis="current-spatial-support", + history=( + HistorySample( + frame_id=packet.envelope.frame_id, + evidence_time_ns=packet.envelope.timestamps.source_ns, + centroid_xyz_m=(4.0, 1.0, 0.5), + ), + ), + cells=(GridCell(sequence, 0, 0),), + coordinate_frame="map", + last_centroid_xyz_m=(4.0, 1.0, 0.5), + motion=MotionState.UNKNOWN, + motion_confidence=0.0, + motion_reason="insufficient-history", + ), + ) + + +class _Motion: + provider_id = "test-motion/v1" + + def estimate( + self, + packet: SourcePacket, + obstacles: tuple[TemporalObstacle, ...], + ) -> tuple[TemporalObstacle, ...]: + return obstacles + + +class _Threat: + provider_id = "test-threat/v1" + + def assess(self, obstacle_map: LocalObstacleMap) -> tuple[ThreatAssessment, ...]: + occupied = obstacle_map.occupied + return tuple( + ThreatAssessment( + assessment_id=f"assessment-{item.component_id}", + component_id=item.component_id, + rig_profile_id="ravnoves00-virtual-rig/v1", + corridor_profile_id="ravnoves00-virtual-corridor/v1", + qualification=QualificationState.QUALIFIED, + relative_speed_mps=0.0, + closest_approach_m=4.0, + ttc_seconds=None, + corridor_intersection=CorridorIntersection.CLEAR, + decision=ThreatDecision.NOT_THREAT, + reason_codes=("qualified-corridor-clear",), + ) + for item in occupied + ) + + +def _config( + capacity: int = 8, + terminal_timeout_ns: int = 500_000_000, +) -> ReferencePerceptionGraphConfig: + ids = { + ProviderRole.SOURCE: _Source.provider_id, + ProviderRole.DETECTOR: _Detector.provider_id, + ProviderRole.GEOMETRY: _Geometry.provider_id, + ProviderRole.TEMPORAL: _Temporal.provider_id, + ProviderRole.MOTION: _Motion.provider_id, + ProviderRole.THREAT: _Threat.provider_id, + } + return ReferencePerceptionGraphConfig( + graph_id="reference-perception-graph/v1", + source_profile_id=BASELINE_PROFILE_ID, + providers=tuple( + ProviderPin(role, provider_id, "v1", "test-revision", "a" * 64) + for role, provider_id in ids.items() + ), + queues=tuple( + QueuePolicy(stage, capacity, 80_000_000, terminal_timeout_ns) + for stage in ("detector", "geometry", "temporal", "threat") + ), + authority=GraphAuthority(), + ) + + +class _MemoryTelemetry: + def __init__(self) -> None: + self.records: list[dict[str, object]] = [] + self._lock = threading.Lock() + + def publish(self, topic: str, payload: bytes) -> None: + with self._lock: + self.records.append({"topic": topic, "payload": json.loads(payload)}) + + +def _graph( + source: object, + *, + detector: object | None = None, + capacity: int = 8, + terminal_timeout_ns: int = 500_000_000, + telemetry: _MemoryTelemetry | None = None, +) -> ReferencePerceptionGraphV1: + identity = ( + PipelineTelemetryIdentity( + contour_id="worker-006", + agent_id="mission-core-worker", + node_id="DESKTOP-OPJ8J04", + lab_id="M4", + run_id="m4-unit-run", + source_id="RAVNOVES00", + source_package_id="ravnoves00-recorded-source/v1", + method_id="reference-perception-graph/v1", + ) + if telemetry is not None + else None + ) + return ReferencePerceptionGraphV1( + config=_config(capacity, terminal_timeout_ns), + source=source, + detector=detector or _Detector(), + geometry=_Geometry(), + temporal=_Temporal(), + motion=_Motion(), + threat=_Threat(), + telemetry_identity=identity, + telemetry_sink=telemetry, + clock_ns=lambda: 10_000, + ) + + +def test_reference_graph_closes_accounting_telemetry_and_deterministic_digest() -> None: + telemetry = _MemoryTelemetry() + graph = _graph(_Source((_packet(0), _packet(1))), telemetry=telemetry) + + first = graph.run() + second = graph.run() + + assert first.state is GraphState.STOPPED + assert first.admitted_count == 2 + assert [item.outcome for item in first.terminal_outcomes] == [ + TerminalOutcomeType.DELIVERED, + TerminalOutcomeType.DELIVERED, + ] + assert all( + delivery.obstacle_map.accounting.source_envelopes == 1 + for delivery in first.deliveries + ) + assert all(delivery.obstacle_map.free_space_claimed is False for delivery in first.deliveries) + assert first.canonical_payload_sha256 == second.canonical_payload_sha256 + assert graph.worker_threads_alive == 0 + stage_ids = { + record["payload"].get("stage_id") + for record in telemetry.records + if isinstance(record["payload"], dict) + } + assert {"detector", "geometry", "temporal", "threat"}.issubset(stage_ids) + + +def test_reference_graph_marks_unavailable_stale_and_provider_failure() -> None: + class FailingDetector(_Detector): + def detect(self, packet: SourcePacket) -> tuple[ObjectProposal2D, ...]: + if packet.envelope.sequence == 2: + raise RuntimeError("deliberate provider failure") + return super().detect(packet) + + FailingDetector.provider_id = _Detector.provider_id + packets = ( + _packet(0, image=False), + _packet(1, source_age_ns=80_000_001), + _packet(2), + _packet(3), + ) + result = _graph(_Source(packets), detector=FailingDetector()).run() + + assert [item.outcome for item in result.terminal_outcomes] == [ + TerminalOutcomeType.UNAVAILABLE, + TerminalOutcomeType.STALE, + TerminalOutcomeType.FAILED, + TerminalOutcomeType.DELIVERED, + ] + assert len(result.deliveries) == 1 + assert result.deliveries[0].sequence == 3 + + +def test_bounded_detector_queue_has_explicit_latest_wins_supersession() -> None: + detector_started = Event() + source_emitted = Event() + release_detector = Event() + + class BlockingDetector(_Detector): + def detect(self, packet: SourcePacket) -> tuple[ObjectProposal2D, ...]: + if packet.envelope.sequence == 0: + detector_started.set() + assert release_detector.wait(2) + return super().detect(packet) + + BlockingDetector.provider_id = _Detector.provider_id + + class CoordinatedSource(_Source): + def packets(self, stop_event: Event) -> Iterator[SourcePacket]: + yield self.values[0] + assert detector_started.wait(2) + yield from self.values[1:] + source_emitted.set() + + source = CoordinatedSource(tuple(_packet(index) for index in range(4))) + graph = _graph( + source, + detector=BlockingDetector(), + capacity=1, + terminal_timeout_ns=1_000_000_000, + ) + holder: list[GraphRunResult] = [] + runner = threading.Thread(target=lambda: holder.append(graph.run())) + runner.start() + assert source_emitted.wait(2) + release_detector.set() + runner.join(3) + + assert not runner.is_alive() + outcomes = {item.sequence: item.outcome for item in holder[0].terminal_outcomes} + assert outcomes[1] is TerminalOutcomeType.SUPERSEDED + assert outcomes[2] is TerminalOutcomeType.SUPERSEDED + assert outcomes[3] is TerminalOutcomeType.DELIVERED + assert set(outcomes) == {0, 1, 2, 3} + assert graph.worker_threads_alive == 0 + + +def test_cancelled_graph_closes_threads_and_can_restart() -> None: + source_waiting = Event() + + class RestartableSource(_Source): + block = True + + def packets(self, stop_event: Event) -> Iterator[SourcePacket]: + yield self.values[0] + if self.block: + source_waiting.set() + stop_event.wait(2) + + source = RestartableSource((_packet(0),)) + graph = _graph(source) + holder: list[GraphRunResult] = [] + runner = threading.Thread(target=lambda: holder.append(graph.run())) + runner.start() + assert source_waiting.wait(2) + graph.cancel() + runner.join(3) + + assert holder[0].state is GraphState.CANCELLED + assert graph.worker_threads_alive == 0 + source.block = False + restarted = graph.run() + assert restarted.state is GraphState.STOPPED + assert restarted.terminal_outcomes[0].outcome is TerminalOutcomeType.DELIVERED + assert graph.worker_threads_alive == 0 + + +def _write_recorded_fixture(root: Path, *, mismatched: bool = False) -> tuple[Path, Path]: + camera_path = root / "camera.jsonl" + timeline_path = root / "source-pack.npz" + camera_rows = [ + { + "schema_version": "missioncore.camera-recording-index/v1", + "kind": "media", + "sequence": index + 1, + "host_epoch_ns": 100 + index, + "host_monotonic_ns": 200 + index, + "sha256": f"{index + 1:064x}", + } + for index in range(2) + ] + camera_path.write_text("\n".join(json.dumps(row) for row in camera_rows) + "\n", "utf-8") + np.savez( + timeline_path, + frame_indices=np.asarray([0, 2 if mismatched else 1], dtype=np.int64), + source_frame_indices=np.asarray([0, 1], dtype=np.int64), + session_seconds=np.asarray([10.0, 10.1], dtype=np.float64), + sample_available=np.asarray([True, False], dtype=np.bool_), + ) + return camera_path, timeline_path + + +def test_recorded_source_reuses_one_timeline_for_1x_and_uncapped(tmp_path: Path) -> None: + camera_path, timeline_path = _write_recorded_fixture(tmp_path) + now = [1_000_000_000] + waits: list[float] = [] + + def wait(stop_event: Event, seconds: float) -> bool: + waits.append(seconds) + now[0] += round(seconds * 1_000_000_000) + return stop_event.is_set() + + source = RecordedRavnoves00Source( + camera_index_path=camera_path, + source_pack_path=timeline_path, + pacing=ReplayPacing.ONE_X, + expected_frame_count=2, + expected_source_pack_sha256=None, + clock_ns=lambda: now[0], + wait=wait, + ) + one_x = list(source.packets(Event())) + uncapped = list( + RecordedRavnoves00Source( + camera_index_path=camera_path, + source_pack_path=timeline_path, + pacing=ReplayPacing.UNCAPPED, + expected_frame_count=2, + expected_source_pack_sha256=None, + wait=lambda _event, _seconds: pytest.fail("uncapped replay waited"), + ).packets(Event()) + ) + + assert [packet.envelope.to_dict() for packet in one_x] == [ + packet.envelope.to_dict() for packet in uncapped + ] + assert waits == pytest.approx([0.1]) + assert one_x[1].envelope.registered_point_increment.available is False + assert one_x[1].registered_point_increment_payload is None + + +def test_recorded_source_rejects_timeline_mismatch(tmp_path: Path) -> None: + camera_path, timeline_path = _write_recorded_fixture(tmp_path, mismatched=True) + source = RecordedRavnoves00Source( + camera_index_path=camera_path, + source_pack_path=timeline_path, + expected_frame_count=2, + expected_source_pack_sha256=None, + ) + with pytest.raises(RecordedSourceError, match="timeline frame index"): + list(source.packets(Event())) + + +def test_camera_only_path_never_invents_metric_occupancy_or_free_space() -> None: + result = _graph(_Source((_packet(0, lidar=False),))).run() + delivery = result.deliveries[0] + assert delivery.obstacle_map.occupied == () + assert delivery.obstacle_map.unknown == () + assert delivery.obstacle_map.free_space_claimed is False + assert len(delivery.obstacle_map.camera_uncertainty) == 1 + + +def test_graph_config_rejects_unbounded_or_missing_stage_policy() -> None: + config = _config() + with pytest.raises(ValueError, match="each reference stage"): + replace(config, queues=config.queues[:-1])