feat(perception): export bounded runtime observations outside heartbeat lane

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

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"),

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()

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()