diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_readiness_probe.py b/experiments/perception/worker/streaming_profile_stage1/pilot_readiness_probe.py new file mode 100644 index 0000000..c4f3c8b --- /dev/null +++ b/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() diff --git a/src/k1link/perception/streaming_lifecycle.py b/src/k1link/perception/streaming_lifecycle.py index 6d9c488..163b704 100644 --- a/src/k1link/perception/streaming_lifecycle.py +++ b/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} + ), } diff --git a/src/k1link/perception/worker_readiness.py b/src/k1link/perception/worker_readiness.py new file mode 100644 index 0000000..f1bff8b --- /dev/null +++ b/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, + } diff --git a/tests/test_perception_worker_readiness.py b/tests/test_perception_worker_readiness.py new file mode 100644 index 0000000..0826903 --- /dev/null +++ b/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()