diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_lifecycle.py b/experiments/perception/worker/streaming_profile_stage1/pilot_lifecycle.py index 80a2283..e215023 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_lifecycle.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_lifecycle.py @@ -16,6 +16,7 @@ from pathlib import Path from pilot_freshness import CLOCK_DOMAIN from k1link.perception.realtime_contract import StreamStart +from k1link.perception.runtime_telemetry import RuntimeTelemetryPublisher from k1link.perception.streaming_continuity import StreamSuspended from k1link.perception.streaming_lifecycle import StreamingLifecycle from k1link.perception.worker_control import WorkerControlChannel @@ -132,8 +133,14 @@ class PilotController: ) self.thread.start() self.control = None + self.telemetry = None self.report = report try: + if getattr(args, "telemetry_snapshot", None): + self.telemetry = RuntimeTelemetryPublisher( + self.runtime, Path(args.telemetry_snapshot), + "K1 DDRNet + RF-DETR + LiDAR + TGS", + ) if control_config: self.control = WorkerControlPump( self.runtime, @@ -195,4 +202,8 @@ class PilotController: self.report["worker_control_thread_released"] = self.control.close() self.report["worker_control"] = self.control.snapshot() self.stop_renewals() - return self.runtime.close(reason) + closed = self.runtime.close(reason) + if self.telemetry: + self.report["telemetry_thread_released"] = self.telemetry.close() + self.report["telemetry_export_failures"] = self.telemetry.failures + return closed 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 32447b4..ef50201 100644 --- a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py +++ b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py @@ -921,6 +921,7 @@ if __name__ == "__main__": parser.add_argument("--controller-image-sha256") parser.add_argument("--worker-control-requests") parser.add_argument("--worker-control-responses") + parser.add_argument("--telemetry-snapshot", help="Optional host-agent current snapshot, never readiness") parser.add_argument( "--worker-readiness-mode", choices=("strict-envelope", "labelled-experiment"), diff --git a/src/k1link/perception/runtime_telemetry.py b/src/k1link/perception/runtime_telemetry.py new file mode 100644 index 0000000..de38c67 --- /dev/null +++ b/src/k1link/perception/runtime_telemetry.py @@ -0,0 +1,84 @@ +"""Lossy, bounded observability export; never a lease/readiness authority. + +The controller mounts one private directory for the existing host agent. There +is one atomic current snapshot, no per-frame journal, credential or model data. +Slow/broken filesystem I/O affects only this daemon, never the heartbeat lane. +""" + +from __future__ import annotations + +import json +import os +import threading +from datetime import UTC, datetime +from pathlib import Path + +from .streaming_lifecycle import StreamingLifecycle + +SCHEMA = "missioncore.perception-runtime-observation/v1" +MAX_BYTES = 8192 + + +def observation(runtime: StreamingLifecycle, profile_name: str) -> dict[str, object]: + snapshot = runtime.snapshot() + continuity = snapshot["input_continuity"] + phase = continuity.get("phase", "active") + state = snapshot["state"] + if state == "running" and phase in ("waiting", "synchronizing"): + state = phase + with runtime.mailbox.condition: + pending = len(runtime.mailbox.pending) + allocated = runtime.mailbox.bytes + return { + "schema_version": SCHEMA, + "observed_at_utc": datetime.now(UTC).isoformat(), + "service_state": state, + "profile_name": profile_name, + "active_request_id": runtime.start.run_id, + "input_state": phase, + "wait_reason": continuity.get("reason") or "", + "live_children": snapshot["live_children"], + "input_pauses": continuity.get("pauses", 0), + "pending_bundles": pending, + # Includes decoder scratch + admitted active inputs, not only the queue. + "buffer_bytes": allocated, + "actuation_allowed": False, + "realtime_qualified": False, + } + + +class RuntimeTelemetryPublisher: + def __init__(self, runtime: StreamingLifecycle, path: Path, profile_name: str) -> None: + if not 1 <= len(profile_name) <= 160 or not path.parent.is_dir(): + raise ValueError("telemetry needs a bounded profile name and existing directory") + self.runtime, self.path, self.profile_name = runtime, path, profile_name + self.failures = 0 + self.stop_event = threading.Event() + self.thread = threading.Thread(target=self._run, name="runtime-observability", daemon=True) + self.thread.start() + + def publish(self) -> bool: + temporary = self.path.with_name(self.path.name + ".tmp") + try: + raw = json.dumps(observation(self.runtime, self.profile_name), allow_nan=False).encode() + if len(raw) > MAX_BYTES: + raise ValueError("runtime observation exceeds bound") + with temporary.open("wb") as stream: + stream.write(raw) + os.replace(temporary, self.path) + return True + except (OSError, ValueError): + self.failures += 1 + return False + + def _run(self) -> None: + while not self.stop_event.is_set(): + self.publish() + if self.stop_event.wait(1.0): + break + self.publish() # Terminal state is exported by this thread, not close(). + + def close(self) -> bool: + self.stop_event.set() + self.thread.join(timeout=0.2) + return not self.thread.is_alive() diff --git a/tests/test_perception_runtime_telemetry.py b/tests/test_perception_runtime_telemetry.py new file mode 100644 index 0000000..4e2977d --- /dev/null +++ b/tests/test_perception_runtime_telemetry.py @@ -0,0 +1,82 @@ +"""Observability is bounded and cannot become model ownership/readiness.""" + +import json +import threading + +from k1link.perception.realtime_contract import StreamStart +from k1link.perception.runtime_telemetry import MAX_BYTES, RuntimeTelemetryPublisher, observation +from k1link.perception.streaming_lifecycle import StreamingLifecycle +from k1link.perception.streaming_queue import StreamMailbox + + +def test_observation_tracks_wait_without_renewing_or_stopping_runtime(tmp_path): + start = StreamStart( + "run", + "source", + "worker-006", + "epoch", + 1, + "a" * 64, + "b" * 64, + "c" * 64, + "d" * 64, + "clock", + "live", + ) + runtime = StreamingLifecycle( + start, + tmp_path / "lease", + StreamMailbox(), + threading.Event(), + recover_input=True, + source_clock_ns=lambda: 1000, + ) + try: + runtime.ready() + before = runtime.lease.renewals + assert observation(runtime, "Profile")["service_state"] == "running" + runtime.pause_input(start, "input-timeout") + waiting = observation(runtime, "Profile") + assert waiting["service_state"] == "waiting" + assert waiting["input_pauses"] == 1 + assert waiting["actuation_allowed"] is False + assert len(json.dumps(waiting).encode()) < MAX_BYTES + assert runtime.lease.renewals == before and not runtime.stop_event.is_set() + runtime.begin_input(start) + assert observation(runtime, "Profile")["service_state"] == "synchronizing" + publisher = RuntimeTelemetryPublisher(runtime, tmp_path / "current.json", "Profile") + try: + runtime.renew(start) + assert runtime.close() + finally: + assert publisher.close() + assert json.loads((tmp_path / "current.json").read_text())["service_state"] == "stopped" + finally: + runtime.close() + + +def test_export_failure_is_lossy_not_terminal(tmp_path): + start = StreamStart( + "run", + "source", + "worker-006", + "epoch", + 1, + "a" * 64, + "b" * 64, + "c" * 64, + "d" * 64, + "clock", + "live", + ) + runtime = StreamingLifecycle(start, tmp_path / "lease", StreamMailbox(), threading.Event()) + (tmp_path / "current.json").mkdir() # os.replace cannot replace this directory. + publisher = RuntimeTelemetryPublisher(runtime, tmp_path / "current.json", "Profile") + try: + assert publisher.close() + assert publisher.failures > 0 + runtime.ready() + runtime.renew(start) + assert not runtime.stop_event.is_set() + finally: + runtime.close()