feat(perception): gate streaming lifecycle on controller readiness

experiments/perception/worker/streaming_profile_stage1/pilot_readiness_probe.py

@@ -0,0 +1,206 @@
+"""CPU-only Linux lifecycle fault probe with explicitly SYNTHETIC GPU facts.
+
+No models, sensor files, GPU devices or host management. This proves bounded
+controller/supervisor behavior, not actual NVML inventory collection or latency
+of the perception graph. The fixture lease is NOT the physical Worker lease.
+"""
+
+import argparse
+import hashlib
+import json
+import os
+import select
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from dataclasses import asdict, replace
+from datetime import UTC, datetime
+from pathlib import Path
+
+from k1link.perception.realtime_contract import StreamStart
+from k1link.perception.streaming_lifecycle import StreamingLifecycle
+from k1link.perception.streaming_queue import StreamMailbox
+from k1link.perception.worker_lease import WorkerLeaseError
+from k1link.perception.worker_operating_envelope import WorkerOperatingEnvelope, WorkerSnapshot
+from k1link.perception.worker_readiness import WorkerReadinessMonitor
+
+
+def trial(root, lease_root, generation, scenario):
+    mode = "labelled-experiment" if scenario == "allowed-overload" else "strict-envelope"
+    envelope = WorkerOperatingEnvelope(
+        "synthetic-4090/v1", "synthetic-RTX4090", "synthetic-driver", 1000, 256, 2610, 10251
+    )
+    config = {"mode": mode, "envelope": asdict(envelope), "scenario": scenario}
+    start = StreamStart(
+        f"fixture-{generation}",
+        "no-recording",
+        "synthetic-worker",
+        f"epoch-{generation}",
+        generation,
+        "a" * 64,
+        "b" * 64,
+        hashlib.sha256(json.dumps(config, sort_keys=True).encode()).hexdigest(),
+        "d" * 64,
+        "unused-source-clock",
+        "live",
+    )
+
+    def observed(**changes):
+        return replace(
+            WorkerSnapshot(
+                start.worker_id,
+                "local-worker-process",
+                time.monotonic_ns(),
+                envelope.gpu_name,
+                envelope.driver_version,
+                start.image_sha256,
+                start.effective_config_sha256,
+                1000,
+                256,
+                2610,
+                10251,
+                start.run_id,
+                generation,
+                (),
+                True,
+            ),
+            **changes,
+        )
+
+    runtime = StreamingLifecycle(
+        start,
+        lease_root,
+        StreamMailbox(),
+        threading.Event(),
+        readiness=WorkerReadinessMonitor(
+            start,
+            envelope,
+            observed(),
+            mode=mode,
+            clock_domain_id="local-worker-process",
+            now_monotonic_ns=time.monotonic_ns(),
+        ),
+    )
+    heartbeat_stop = threading.Event()
+
+    def heartbeat():
+        while not heartbeat_stop.wait(0.1):
+            try:
+                runtime.renew(start)
+            except WorkerLeaseError:
+                return
+
+    renewer = threading.Thread(target=heartbeat, name="fixture-controller", daemon=True)
+    renewer.start()
+    child = None
+    packet = None
+    result = {"scenario": scenario, "config": config, "start_monotonic_ns": time.monotonic_ns()}
+    try:
+        child = runtime.spawn(
+            lambda: subprocess.Popen(
+                [sys.executable, "-c", "import time; print('ready',flush=True); time.sleep(15)"],
+                stdout=subprocess.PIPE,
+                start_new_session=True,
+            )
+        )
+        if not select.select([child.stdout], [], [], 2)[0] or child.stdout.readline() != b"ready\n":
+            raise RuntimeError("fixture child startup failed")
+        runtime.ready()
+        packet = {"sequence": 0, "payload_bytes": 32}
+        assert runtime.admit(start, packet) and runtime.mailbox.take() is packet
+        for _ in range(3):
+            assert not runtime.stop_event.wait(0.1)
+            runtime.observe_worker(start, observed())
+        result["running_after_refresh"] = runtime.state.value
+        result["fault_injected_monotonic_ns"] = time.monotonic_ns()
+        if scenario != "expired-inventory":
+            change = (
+                {"competing_gpu_clients": None}
+                if scenario == "unknown-inventory"
+                else {"memory_clock_mhz": 405}
+            )
+            try:
+                runtime.observe_worker(start, observed(**change))
+                assert scenario == "allowed-overload"
+            except WorkerLeaseError:
+                assert scenario != "allowed-overload"
+        if scenario == "allowed-overload":
+            runtime.validate_result_binding(start.to_dict())
+            result["overload_accepted"] = True
+            runtime.observe_worker(start, observed())
+            result["recovered_envelope_audit"] = runtime.snapshot()["worker_readiness"]
+            runtime.request_stop("completed")
+        else:
+            assert runtime.stop_event.wait(2)
+            assert runtime.reason == "worker-not-ready"
+            try:
+                runtime.observe_worker(start, observed())
+                raise AssertionError("late inventory revived stopped runtime")
+            except WorkerLeaseError:
+                result["late_refresh_rejected"] = True
+            try:
+                runtime.validate_result_binding(start.to_dict())
+                raise AssertionError("stopped runtime published a result")
+            except WorkerLeaseError:
+                result["late_result_rejected"] = True
+        child.wait(timeout=5)
+        result["child_exited_monotonic_ns"] = time.monotonic_ns()
+        assert not runtime.close() and not runtime.lease.released
+        result["active_input_prevents_retirement"] = True
+    finally:
+        heartbeat_stop.set()
+        renewer.join(timeout=1)
+        runtime.request_stop("failed")
+        runtime.stop_children()
+        if packet is not None:
+            runtime.mailbox.release(packet)
+        result["released"] = runtime.close()
+        result["runtime"] = runtime.snapshot()
+        result["input_bytes"] = runtime.mailbox.bytes
+        result["heartbeat_alive"] = renewer.is_alive()
+        result["finished_monotonic_ns"] = time.monotonic_ns()
+        if child is not None:
+            child.stdout.close()
+        (root / f"{scenario}.json").write_text(json.dumps(result, indent=2) + "\n")
+    assert result["released"] and not result["input_bytes"] and not result["heartbeat_alive"]
+    stop_ns = int(result["runtime"]["stop_requested_monotonic_ns"])
+    assert result["finished_monotonic_ns"] - stop_ns <= 5_000_000_000
+    return result
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output", required=True)
+    args = parser.parse_args()
+    root = Path(args.output)
+    root.mkdir()
+    report = {
+        "created_utc": datetime.now(UTC).isoformat(),
+        "started_monotonic_ns": time.monotonic_ns(),
+        "scope": "CPU-only real Linux process lifecycle; synthetic GPU telemetry",
+        "real_gpu_inventory_proved": False,
+        "full_profile_executed": False,
+        "gpu_devices_visible": sorted(str(p) for p in Path("/dev").glob("nvidia*")),
+        "cuda_visible_devices": os.environ.get("CUDA_VISIBLE_DEVICES"),
+    }
+    assert not report["gpu_devices_visible"] and report["cuda_visible_devices"] == ""
+    with tempfile.TemporaryDirectory(prefix="readiness-fixture-") as directory:
+        lease_root = Path(directory)
+        report["trials"] = [
+            trial(root, lease_root, i, case)
+            for i, case in enumerate(
+                ("strict-low-clocks", "allowed-overload", "unknown-inventory", "expired-inventory"),
+                1,
+            )
+        ]
+        report["terminal_fixture_owner"] = json.loads((lease_root / "owner.json").read_bytes())
+    report["finished_monotonic_ns"] = time.monotonic_ns()
+    report["passed"] = True
+    (root / "report.json").write_text(json.dumps(report, indent=2) + "\n")
+    print(json.dumps({"passed": True, "scenarios": len(report["trials"]), "gpu_models": 0}))
+
+
+if __name__ == "__main__":
+    main()

src/k1link/perception/streaming_lifecycle.py

@@ -23,6 +23,8 @@ from .realtime_contract import StreamStart
 from .realtime_scene import _wire_integer
 from .streaming_queue import IngressReservation, StreamBundle, StreamMailbox
 from .worker_lease import WorkerLease, WorkerLeaseError
+from .worker_operating_envelope import WorkerSnapshot
+from .worker_readiness import WorkerReadinessError, WorkerReadinessMonitor
 
 
 def _group_exists(pgid: int) -> bool:
@@ -45,8 +47,12 @@ class StreamingLifecycle:
         *,
         ttl_seconds: float = 2.0,
         clock_ns: Callable[[], int] = time.monotonic_ns,
+        readiness: WorkerReadinessMonitor | None = None,
     ) -> None:
         self.start, self.mailbox, self.stop_event = start, mailbox, stop
+        self._clock_ns, self._readiness = clock_ns, readiness
+        if readiness is not None:
+            readiness.check(start, now_monotonic_ns=clock_ns(), require_warmup=False)
         self._lock = threading.RLock()
         self._cleanup_lock = threading.Lock()
         self._children: list[subprocess.Popen[bytes]] = []
@@ -75,6 +81,40 @@ class StreamingLifecycle:
             if start == self.start:
                 self.request_stop("lease-lost")
             raise
+        self._check_readiness(require_warmup=self.state == GraphState.RUNNING)
+
+    def _check_readiness(self, *, require_warmup: bool) -> None:
+        if self._readiness is not None:
+            try:
+                self._readiness.check(
+                    self.start,
+                    now_monotonic_ns=self._clock_ns(),
+                    require_warmup=require_warmup,
+                )
+            except WorkerReadinessError:
+                self.request_stop("worker-not-ready")
+                raise
+
+    def observe_worker(self, start: StreamStart, observed: WorkerSnapshot) -> None:
+        """Trusted control plane only; sensor payloads never call this method.
+
+        This is not a lease renewal. A fresh inventory and an active controller
+        heartbeat are independent requirements. No host operations under lock.
+        """
+        with self._lock:
+            self._check(start, starting=True)
+            if self._readiness is None:
+                raise WorkerReadinessError("worker readiness monitoring is not configured")
+            try:
+                self._readiness.observe(
+                    start,
+                    observed,
+                    now_monotonic_ns=self._clock_ns(),
+                    require_warmup=self.state == GraphState.RUNNING,
+                )
+            except WorkerReadinessError:
+                self.request_stop("worker-not-ready")
+                raise
 
     def renew(self, start: StreamStart) -> None:
         with self._lock:
@@ -89,6 +129,7 @@ class StreamingLifecycle:
             if any(p.poll() is not None for p in self._children):
                 self.request_stop("child-exited")
                 raise WorkerLeaseError("owned child exited during warmup")
+            self._check_readiness(require_warmup=True)
             self.state = GraphState.RUNNING
 
     def spawn(self, factory: Callable[[], subprocess.Popen[bytes]]) -> subprocess.Popen[bytes]:
@@ -157,6 +198,7 @@ class StreamingLifecycle:
             "lease-lost",
             "child-exited",
             "invalid-process-group",
+            "worker-not-ready",
             "failed",
         ):
             raise ValueError("unknown runtime stop reason")
@@ -262,4 +304,9 @@ class StreamingLifecycle:
                     _wire_integer(self.retired_ns) if self.retired_ns is not None else None
                 ),
                 "event_clock": "worker-process-monotonic",
+                "worker_readiness": (
+                    self._readiness.snapshot()
+                    if self._readiness is not None
+                    else {"enabled": False, "realtime_qualified": False}
+                ),
             }

src/k1link/perception/worker_readiness.py

@@ -0,0 +1,140 @@
+"""Bounded controller-owned readiness state, independent of the sensor stream.
+
+The trusted Worker controller supplies observations in its local monotonic clock
+domain. It must include mode/envelope in the effective configuration identity.
+This does not collect Docker/NVML facts, authenticate a remote controller, set
+clocks, or certify real-time operation. The lifecycle serializes all access.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict
+from typing import Literal
+
+from .realtime_contract import RealtimeContractError, StreamStart, _identifier, _integer
+from .worker_lease import WorkerLeaseError
+from .worker_operating_envelope import (
+    WorkerOperatingEnvelope,
+    WorkerSnapshot,
+    operating_envelope_failures,
+)
+
+ReadinessMode = Literal["strict-envelope", "labelled-experiment"]
+# Only measured performance conditions are waivable. Missing ownership, input
+# identity, warmup or inventory NEVER becomes an allowed overload experiment.
+_PERFORMANCE_FAILURES = frozenset(
+    f"{field}-outside-envelope-or-unknown"
+    for field in ("gpu_name", "driver_version", "cpu_limit_millicores", "memory_limit_mib")
+) | frozenset(
+    f"{field}-below-envelope-or-unknown" for field in ("sm_clock_mhz", "memory_clock_mhz")
+)
+
+
+class WorkerReadinessError(WorkerLeaseError):
+    """A current controller lost the prerequisites for further execution."""
+
+
+class WorkerReadinessMonitor:
+    def __init__(
+        self,
+        start: StreamStart,
+        envelope: WorkerOperatingEnvelope,
+        initial: WorkerSnapshot,
+        *,
+        mode: ReadinessMode,
+        clock_domain_id: str,
+        now_monotonic_ns: int,
+    ) -> None:
+        if mode not in ("strict-envelope", "labelled-experiment"):
+            raise RealtimeContractError("unknown Worker readiness mode")
+        _identifier(clock_domain_id, "worker clock domain")
+        self.start, self.envelope = start, envelope
+        self.mode, self.clock_domain_id = mode, clock_domain_id
+        self._observed = initial
+        self._last_check_ns = now_monotonic_ns
+        self._failure: tuple[str, ...] = ()
+        self._current: tuple[str, ...] = ()
+        self._post_warmup_violations: set[str] = set()
+        self._warmup_checked = False
+        self.observations = 1
+        self.check(start, now_monotonic_ns=now_monotonic_ns, require_warmup=False)
+
+    def _bound(self, start: StreamStart) -> None:
+        # A stale client may not poison the current owner's state.
+        if start != self.start:
+            raise WorkerReadinessError("readiness stream identity mismatch")
+
+    def _fail(self, reasons: tuple[str, ...]) -> None:
+        self._failure = self._failure or reasons
+        raise WorkerReadinessError("worker readiness lost: " + ",".join(self._failure))
+
+    def check(self, start: StreamStart, *, now_monotonic_ns: int, require_warmup: bool) -> None:
+        self._bound(start)
+        if self._failure:
+            self._fail(self._failure)
+        try:
+            _integer(now_monotonic_ns, "worker monotonic time")
+            if now_monotonic_ns < self._last_check_ns:
+                raise RealtimeContractError("worker clock moved backwards")
+            failures = operating_envelope_failures(
+                self.envelope,
+                start,
+                self._observed,
+                now_monotonic_ns=now_monotonic_ns,
+                clock_domain_id=self.clock_domain_id,
+            )
+        except RealtimeContractError:
+            self._fail(("worker-snapshot-clock-invalid",))
+            return  # Unreachable, keeps static narrowing explicit.
+        self._last_check_ns = now_monotonic_ns
+        self._current = failures
+        if require_warmup:
+            self._warmup_checked = True
+            self._post_warmup_violations.update(set(failures) & _PERFORMANCE_FAILURES)
+        fatal = tuple(
+            failure
+            for failure in failures
+            if failure not in _PERFORMANCE_FAILURES
+            and (require_warmup or failure != "warmup-not-complete")
+        )
+        if fatal:
+            self._fail(fatal)
+        if require_warmup and self.mode == "strict-envelope" and failures:
+            self._fail(failures)
+
+    def observe(
+        self,
+        start: StreamStart,
+        observed: WorkerSnapshot,
+        *,
+        now_monotonic_ns: int,
+        require_warmup: bool,
+    ) -> None:
+        # Recheck the OLD observation before replacement: late telemetry cannot
+        # resurrect expired authority, even if the watchdog has not run yet.
+        self.check(start, now_monotonic_ns=now_monotonic_ns, require_warmup=require_warmup)
+        if observed.observed_monotonic_ns <= self._observed.observed_monotonic_ns:
+            self._fail(("worker-snapshot-not-increasing",))
+        self._observed = observed
+        self.observations += 1
+        self.check(start, now_monotonic_ns=now_monotonic_ns, require_warmup=require_warmup)
+
+    def snapshot(self) -> dict[str, object]:
+        # No history of individual telemetry samples, only fixed-size state.
+        observed = asdict(self._observed)
+        observed["observed_monotonic_ns"] = str(self._observed.observed_monotonic_ns)
+        return {
+            "enabled": True,
+            "mode": self.mode,
+            "envelope": asdict(self.envelope),
+            "worker_clock_domain_id": self.clock_domain_id,
+            "last_check_monotonic_ns": str(self._last_check_ns),
+            "observations": self.observations,
+            "latest_observation": observed,
+            "current_failures": list(self._current),
+            "terminal_failures": list(self._failure),
+            "post_warmup_envelope_violations": sorted(self._post_warmup_violations),
+            "warmup_readiness_checked": self._warmup_checked,
+            "realtime_qualified": False,
+            "actuation_allowed": False,
+        }

tests/test_perception_worker_readiness.py

@@ -0,0 +1,380 @@
+"""Small synthetic control-plane checks; no models, GPU, network or host setters."""
+
+import subprocess
+import sys
+import threading
+from dataclasses import replace
+
+import pytest
+
+from k1link.perception.graph_contracts import GraphState
+from k1link.perception.realtime_contract import RealtimeContractError, StreamStart
+from k1link.perception.streaming_lifecycle import StreamingLifecycle
+from k1link.perception.streaming_queue import StreamMailbox
+from k1link.perception.worker_lease import WorkerLeaseError
+from k1link.perception.worker_operating_envelope import WorkerOperatingEnvelope, WorkerSnapshot
+from k1link.perception.worker_readiness import WorkerReadinessError, WorkerReadinessMonitor
+
+
+def binding(generation=1):
+    return StreamStart(
+        f"run-{generation}",
+        "fixture",
+        "worker-006",
+        f"epoch-{generation}",
+        generation,
+        "a" * 64,
+        "b" * 64,
+        "c" * 64,
+        "d" * 64,
+        "source-clock",
+        "recorded-source-paced",
+    )
+
+
+def observation(now=1_000_000_000, **changes):
+    return replace(
+        WorkerSnapshot(
+            "worker-006",
+            "worker-clock",
+            now,
+            "RTX 4090",
+            "610.47",
+            "b" * 64,
+            "c" * 64,
+            8000,
+            8192,
+            2610,
+            10251,
+            "run-1",
+            1,
+            (),
+            True,
+        ),
+        **changes,
+    )
+
+
+def monitor(*, initial=None, mode="strict-envelope", start=None):
+    return WorkerReadinessMonitor(
+        start or binding(),
+        WorkerOperatingEnvelope("fixture/v1", "RTX 4090", "610.47", 8000, 8192, 2610, 10251),
+        initial or observation(),
+        mode=mode,
+        clock_domain_id="worker-clock",
+        now_monotonic_ns=1_000_000_000,
+    )
+
+
+@pytest.fixture
+def controlled(tmp_path):
+    runs = []
+
+    def create(**options):
+        clock = [1_000_000_000]
+        readiness = monitor(**options)
+        run = StreamingLifecycle(
+            binding(),
+            tmp_path / "worker",
+            StreamMailbox(),
+            threading.Event(),
+            readiness=readiness,
+            clock_ns=lambda: clock[0],
+        )
+        runs.append(run)
+        return run, clock
+
+    yield create
+    for run in runs:
+        assert run.close()
+
+
+def test_admission_requires_fresh_exclusive_inventory_before_any_lease_or_spawn(tmp_path):
+    with pytest.raises(WorkerReadinessError, match="competing-gpu"):
+        StreamingLifecycle(
+            binding(),
+            tmp_path / "worker",
+            StreamMailbox(),
+            threading.Event(),
+            readiness=monitor(initial=observation(competing_gpu_clients=("other-model",))),
+        )
+    assert not (tmp_path / "worker").exists()
+
+
+@pytest.mark.parametrize("warmup", [False, None])
+def test_warmup_may_start_without_ready_but_cannot_admit_frames(controlled, warmup):
+    run, _ = controlled(initial=observation(warmup_complete=warmup, sm_clock_mhz=210))
+    assert run.state == GraphState.STARTING
+    with pytest.raises(WorkerLeaseError, match="accepting"):
+        run.admit(run.start, {"sequence": 0, "payload_bytes": 1})
+    with pytest.raises(WorkerReadinessError, match="warmup-not-complete"):
+        run.ready()
+    assert run.reason == "worker-not-ready"
+
+
+def test_warmup_transition_then_frames_and_results_use_same_monitor(controlled):
+    run, clock = controlled(initial=observation(warmup_complete=False, sm_clock_mhz=210))
+    clock[0] += 100_000_000
+    run.observe_worker(run.start, observation(clock[0]))
+    run.ready()
+    with run.work(run.start, "gpu"):
+        pass
+    run.validate_result_binding(run.start.to_dict())
+    audit = run.snapshot()["worker_readiness"]
+    assert audit["observations"] == 2
+    assert audit["warmup_readiness_checked"]
+    assert audit["post_warmup_envelope_violations"] == []
+    assert audit["realtime_qualified"] is False
+
+
+@pytest.mark.parametrize("mode", ["strict-envelope", "labelled-experiment"])
+@pytest.mark.parametrize(
+    "change,reason",
+    [
+        ({"worker_id": "worker-007"}, "worker_id-mismatch"),
+        ({"image_sha256": None}, "image_sha256-mismatch"),
+        ({"effective_config_sha256": "f" * 64}, "effective_config_sha256-mismatch"),
+        ({"gpu_owner_run_id": "other"}, "exclusive-worker-lease"),
+        ({"lease_generation": 2}, "exclusive-worker-lease"),
+        ({"competing_gpu_clients": ("other-model",)}, "competing-gpu"),
+        ({"competing_gpu_clients": None}, "competing-gpu"),
+        ({"warmup_complete": None}, "warmup-not-complete"),
+    ],
+)
+def test_hard_failures_stop_even_labelled_experiments(controlled, mode, change, reason):
+    run, clock = controlled(mode=mode)
+    run.ready()
+    clock[0] += 100_000_000
+    with pytest.raises(WorkerReadinessError, match=reason):
+        run.observe_worker(run.start, observation(clock[0], **change))
+    assert run.stop_event.is_set() and run.state == GraphState.STOPPING
+    assert run.reason == "worker-not-ready" and not run.lease.released
+    with pytest.raises(WorkerLeaseError):
+        run.validate_result_binding(run.start.to_dict())
+
+
+def test_performance_failure_is_allowed_only_in_explicit_experiment_and_is_latched(controlled):
+    run, clock = controlled(mode="labelled-experiment")
+    run.ready()
+    clock[0] += 100_000_000
+    run.observe_worker(run.start, observation(clock[0], memory_clock_mhz=405))
+    run.renew(run.start)
+    assert not run.stop_event.is_set()
+    clock[0] += 100_000_000
+    run.observe_worker(run.start, observation(clock[0]))
+    audit = run.snapshot()["worker_readiness"]
+    assert audit["current_failures"] == []
+    assert audit["post_warmup_envelope_violations"] == [
+        "memory_clock_mhz-below-envelope-or-unknown"
+    ]
+    assert audit["realtime_qualified"] is False
+
+
+@pytest.mark.parametrize("initial", [True, False])
+def test_strict_envelope_rejects_low_clocks_at_ready_or_during_run(controlled, initial):
+    run, clock = controlled(initial=observation(memory_clock_mhz=405) if initial else None)
+    if initial:
+        with pytest.raises(WorkerReadinessError, match="memory_clock_mhz-below"):
+            run.ready()
+    else:
+        run.ready()
+        clock[0] += 100_000_000
+        with pytest.raises(WorkerReadinessError, match="memory_clock_mhz-below"):
+            run.observe_worker(run.start, observation(clock[0], memory_clock_mhz=405))
+    assert run.reason == "worker-not-ready"
+
+
+@pytest.mark.parametrize("operation", ["renew", "admit", "result", "refresh"])
+def test_heartbeats_cannot_extend_inventory_or_late_refresh_revive_it(controlled, operation):
+    run, clock = controlled(mode="labelled-experiment")
+    run.ready()
+    clock[0] += 900_000_000
+    run.renew(run.start)  # Lease now lasts beyond the inventory deadline.
+    clock[0] += 100_000_001
+    with pytest.raises(WorkerReadinessError, match="worker-snapshot-expired"):
+        if operation == "renew":
+            run.renew(run.start)
+        elif operation == "admit":
+            run.admit(run.start, {"sequence": 0, "payload_bytes": 1})
+        elif operation == "result":
+            run.validate_result_binding(run.start.to_dict())
+        else:
+            run.observe_worker(run.start, observation(clock[0]))
+    assert run.reason == "worker-not-ready"
+    assert run.lease.renewals == 1
+
+
+def test_fresh_inventory_does_not_renew_lease(controlled):
+    run, clock = controlled()
+    run.ready()
+    for _ in range(2):
+        clock[0] += 900_000_000
+        run.observe_worker(run.start, observation(clock[0]))
+    assert run.lease.renewals == 0
+    clock[0] += 200_000_000
+    with pytest.raises(WorkerLeaseError):
+        run.observe_worker(run.start, observation(clock[0]))
+    assert run.reason == "lease-lost"
+
+
+@pytest.mark.parametrize(
+    "field,value", [("epoch_id", "old"), ("lease_generation", 2), ("source_id", "old")]
+)
+def test_old_controller_update_cannot_change_or_stop_current_owner(controlled, field, value):
+    run, clock = controlled()
+    run.ready()
+    before = run.snapshot()["worker_readiness"]
+    clock[0] += 100_000_000
+    with pytest.raises(WorkerLeaseError, match="mismatch"):
+        run.observe_worker(
+            replace(run.start, **{field: value}), observation(clock[0], competing_gpu_clients=None)
+        )
+    assert run.snapshot()["worker_readiness"] == before
+    assert run.state == GraphState.RUNNING and not run.stop_event.is_set()
+
+
+@pytest.mark.parametrize(
+    "change",
+    [
+        {"observed_monotonic_ns": 1_000_000_000},
+        {"observed_monotonic_ns": 1_200_000_000},
+        {"clock_domain_id": "mac-clock"},
+    ],
+)
+def test_bad_telemetry_order_and_clock_fail_closed(controlled, change):
+    run, clock = controlled()
+    run.ready()
+    clock[0] += 100_000_000
+    with pytest.raises(WorkerReadinessError):
+        run.observe_worker(run.start, replace(observation(clock[0]), **change))
+    assert run.reason == "worker-not-ready"
+
+
+def test_idle_watchdog_stops_owned_child_on_inventory_expiry_without_another_frame(controlled):
+    run, clock = controlled()
+    child = run.spawn(
+        lambda: subprocess.Popen(
+            [sys.executable, "-c", "import time; print('ready', flush=True); time.sleep(10)"],
+            stdout=subprocess.PIPE,
+            start_new_session=True,
+        )
+    )
+    try:
+        assert child.stdout.readline() == b"ready\n"
+        run.ready()
+        clock[0] += 1_000_000_001  # Lease still live; no frame or heartbeat to trigger a check.
+        assert run.stop_event.wait(1)
+        child.wait(timeout=2)
+        assert run.reason == "worker-not-ready" and not run.lease.released
+    finally:
+        assert run.close()
+        child.stdout.close()
+
+
+def test_loss_during_compute_rejects_return_and_keeps_lease_until_buffers_released(controlled):
+    run, clock = controlled()
+    run.ready()
+    packet = {"sequence": 0, "payload_bytes": 10}
+    run.admit(run.start, packet)
+    assert run.mailbox.take() is packet
+    try:
+        with pytest.raises(WorkerLeaseError), run.work(run.start, "gpu"):
+            clock[0] += 1_000_000_001
+            with pytest.raises(WorkerLeaseError):
+                run.check_current(run.start)
+            assert run.reason == "worker-not-ready"
+            assert not run.close()  # Callback and input still owned.
+    finally:
+        run.mailbox.release(packet)
+    assert run.close()
+    assert run.state == GraphState.FAILED
+
+
+def test_observation_loss_during_compute_blocks_return_and_preserves_owned_input(controlled):
+    run, clock = controlled()
+    run.ready()
+    with pytest.raises(WorkerLeaseError), run.work(run.start, "cpu"):
+        clock[0] += 100_000_000
+        with pytest.raises(WorkerReadinessError):
+            run.observe_worker(run.start, observation(clock[0], competing_gpu_clients=None))
+        assert not run.close()
+    assert run.close()
+
+
+def test_unmonitored_legacy_runtime_cannot_report_monitored_readiness(tmp_path):
+    run = StreamingLifecycle(binding(), tmp_path, StreamMailbox(), threading.Event())
+    try:
+        assert run.snapshot()["worker_readiness"] == {"enabled": False, "realtime_qualified": False}
+        with pytest.raises(WorkerReadinessError, match="not configured"):
+            run.observe_worker(run.start, observation())
+    finally:
+        assert run.close()
+
+
+def test_invalid_mode_and_binding_rejected_without_mutating_monitor():
+    with pytest.raises(RealtimeContractError):
+        monitor(mode="automatic-pass")
+    value = monitor()
+    before = value.snapshot()
+    with pytest.raises(WorkerReadinessError, match="mismatch"):
+        value.check(binding(2), now_monotonic_ns=1_000_000_000, require_warmup=True)
+    assert value.snapshot() == before
+
+
+def test_failed_monitor_is_terminal_even_when_newer_healthy_snapshot_arrives():
+    value = monitor()
+    with pytest.raises(WorkerReadinessError):
+        value.check(binding(), now_monotonic_ns=2_000_000_001, require_warmup=True)
+    with pytest.raises(WorkerReadinessError, match="expired"):
+        value.observe(
+            binding(),
+            observation(2_000_000_002),
+            now_monotonic_ns=2_000_000_002,
+            require_warmup=True,
+        )
+    assert value.snapshot()["observations"] == 1
+
+
+def test_only_verified_cleanup_allows_next_generation_with_new_monitor(controlled, tmp_path):
+    run, clock = controlled()
+    run.ready()
+    packet = {"sequence": 0, "payload_bytes": 10}
+    run.admit(run.start, packet)
+    assert run.mailbox.take() is packet
+    clock[0] += 100_000_000
+    with pytest.raises(WorkerReadinessError):
+        run.observe_worker(run.start, observation(clock[0], competing_gpu_clients=None))
+    next_start = binding(2)
+    next_monitor = monitor(
+        start=next_start,
+        initial=observation(gpu_owner_run_id=next_start.run_id, lease_generation=2),
+    )
+    try:
+        assert not run.close()
+        with pytest.raises(WorkerLeaseError, match="already owned"):
+            StreamingLifecycle(
+                next_start,
+                tmp_path / "worker",
+                StreamMailbox(),
+                threading.Event(),
+                readiness=next_monitor,
+                clock_ns=lambda: clock[0],
+            )
+    finally:
+        run.mailbox.release(packet)
+        assert run.close()
+    next_run = StreamingLifecycle(
+        next_start,
+        tmp_path / "worker",
+        StreamMailbox(),
+        threading.Event(),
+        readiness=next_monitor,
+        clock_ns=lambda: clock[0],
+    )
+    try:
+        next_run.ready()
+        with pytest.raises(WorkerLeaseError, match="mismatch"):
+            next_run.validate_result_binding(run.start.to_dict())
+        assert next_run.state == GraphState.RUNNING
+    finally:
+        assert next_run.close()