"""Lifecycle/fencing for one subprocess-backed full perception profile. Uses existing GraphState, StreamStart and bounded scheduler primitives. The trusted controller supplies child commands and one Worker-wide lease directory; neither is accepted from stream payloads. No backend queue or model is created here. External-client inventory and real network authentication remain separate. """ from __future__ import annotations import os import signal import subprocess import threading import time from collections.abc import Callable, Iterator from contextlib import contextmanager, suppress from pathlib import Path from typing import Any, Literal from .graph_contracts import GraphState from .realtime_contract import StreamStart from .realtime_scene import _wire_integer from .streaming_continuity import InputContinuity, ResumeEvidence, StreamSuspended 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, WorkerTelemetryUnavailable, ) def _group_exists(pgid: int) -> bool: try: os.killpg(pgid, 0) return True except ProcessLookupError: return False except PermissionError: return True # Unknown is not a release proof. class StreamingLifecycle: def __init__( self, start: StreamStart, lease_root: Path, mailbox: StreamMailbox, stop: threading.Event, *, ttl_seconds: float = 2.0, clock_ns: Callable[[], int] = time.monotonic_ns, readiness: WorkerReadinessMonitor | None = None, recover_input: bool = False, source_clock_ns: Callable[[], int] | None = None, ) -> None: self.start, self.mailbox, self.stop_event = start, mailbox, stop self._clock_ns, self._readiness = clock_ns, readiness if recover_input and ( source_clock_ns is None or (readiness is not None and not readiness.recoverable) ): raise ValueError("resumable input needs mapped source clock and recoverable telemetry") self.continuity = InputContinuity(start) if recover_input else None self._source_clock_ns = source_clock_ns self._ingress_thread: threading.Thread | None = None 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]] = [] self._threads: list[threading.Thread] = [] self._active = {"gpu": 0, "cpu": 0, "resync": 0} self.state = GraphState.CREATED self.reason: str | None = None self.stop_requested_ns: int | None = None self.retired_ns: int | None = None self.lease = WorkerLease(lease_root, start, ttl_seconds=ttl_seconds, clock_ns=clock_ns) self.state = GraphState.STARTING self._watchdog_stop = threading.Event() self._watchdog = threading.Thread( target=self._watch, name="perception-lease-watchdog", daemon=True ) self._watchdog.start() def _check( self, start: StreamStart, *, starting: bool = False, allow_unavailable: bool = False ) -> None: allowed = (GraphState.STARTING, GraphState.RUNNING) if starting else (GraphState.RUNNING,) if self.state not in allowed or self.stop_event.is_set(): raise WorkerLeaseError("runtime is not accepting work") try: self.lease.check(start) except WorkerLeaseError: # A stale client is rejected but cannot cancel the current owner. if start == self.start: self.request_stop("lease-lost") raise self._check_readiness( require_warmup=self.state == GraphState.RUNNING, allow_unavailable=allow_unavailable ) def _check_readiness(self, *, require_warmup: bool, allow_unavailable: bool = False) -> None: if self._readiness is not None: try: self._readiness.check( self.start, now_monotonic_ns=self._clock_ns(), require_warmup=require_warmup, ) except WorkerTelemetryUnavailable as exc: if self.continuity is None: self.request_stop("worker-not-ready") raise self.continuity.pause("worker-telemetry") self.mailbox.pause() if not allow_unavailable: raise StreamSuspended(str(exc)) from exc 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, allow_unavailable=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 WorkerTelemetryUnavailable: if self.continuity is None: self.request_stop("worker-not-ready") raise self.continuity.pause("worker-telemetry") self.mailbox.pause() except WorkerReadinessError: self.request_stop("worker-not-ready") raise def renew(self, start: StreamStart) -> None: with self._lock: self._check(start, starting=True, allow_unavailable=True) self.lease.renew(start) def ready(self) -> None: with self._lock: self._check(self.start, starting=True) if self.state != GraphState.STARTING: raise WorkerLeaseError("warmup completion already consumed") 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]: """Trusted, bounded process creation + registration, atomic with stop.""" with self._lock: self._check(self.start, starting=True) if self.state != GraphState.STARTING or len(self._children) >= 8: raise WorkerLeaseError("model processes may start only once during warmup") process = factory() self._children.append(process) if process.poll() is None and os.getpgid(process.pid) != process.pid: self.request_stop("invalid-process-group") raise WorkerLeaseError("runtime children need dedicated process groups") return process def track_thread(self, thread: threading.Thread) -> None: with self._lock: self._check(self.start, starting=True) self._threads = [t for t in self._threads if t.ident is None or t.is_alive()] if len(self._threads) >= 8 or thread in self._threads: raise WorkerLeaseError("runtime thread registration outside bound") self._threads.append(thread) def register_ingress(self, thread: threading.Thread, epoch: StreamStart) -> None: with self._lock: if self.continuity is not None: self.check_input(epoch, synchronizing=True) if self._ingress_thread is not None and ( self._ingress_thread.ident is None or self._ingress_thread.is_alive() ): raise StreamSuspended("one input connection per resident profile") self.track_thread(thread) self._ingress_thread = thread def check_input( self, epoch: StreamStart, *, synchronizing: bool = False, event_ns: int | None = None ) -> None: with self._lock: if self.continuity is None: self._check(epoch) return self._check(self.start) self.continuity.check(epoch, synchronizing=synchronizing) if event_ns is not None: assert self._source_clock_ns is not None now = self._source_clock_ns() if event_ns < self.continuity.cutoff_ns or not 0 <= now - event_ns <= 250_000_000: raise StreamSuspended("obsolete or future input, no backlog replay") def pause_input(self, epoch: StreamStart, reason: str) -> None: with self._lock: if self.continuity is None: raise ValueError("resumable input was not enabled") # A disconnected old socket must not pause the replacement epoch. if epoch != self.continuity.epoch: raise StreamSuspended("obsolete input epoch") self._check(self.start, allow_unavailable=True) self.continuity.pause(reason) self.mailbox.pause() def begin_input(self, start: StreamStart) -> StreamStart: with self._lock: self._check(start) if self.continuity is None or self._source_clock_ns is None: raise ValueError("resumable input was not enabled") if ( not self.mailbox.epoch_drained or any(self._active.values()) or (self._ingress_thread is not None and self._ingress_thread.is_alive()) ): raise StreamSuspended("old epoch callbacks or connection still active") if self.continuity.phase != "waiting": raise StreamSuspended("pause required before a new input epoch") self.mailbox.begin_epoch() if self._ingress_thread is not None: self._threads = [t for t in self._threads if t is not self._ingress_thread] self._ingress_thread = None return self.continuity.begin(self._source_clock_ns()) def resume_input( self, epoch: StreamStart, evidence: ResumeEvidence, reset_temporal: Callable[[], None] ) -> None: """Trusted adapter proves a new decoded keyframe + current sensor pair. Reset only temporal/rolling outputs, outside the lock, after old compute has drained. No model restart. A failed reset is an actual runtime fault. """ with self._lock: self.check_input(epoch, synchronizing=True) if ( self.continuity is None or self.continuity.phase != "synchronizing" or any(self._active.values()) ): raise StreamSuspended("resynchronization is not quiescent") assert self._source_clock_ns is not None evidence.validate( cutoff_ns=self.continuity.cutoff_ns, source_now_ns=self._source_clock_ns() ) self._active["resync"] = 1 try: reset_temporal() with self._lock: self.check_input(epoch, synchronizing=True) evidence.validate( cutoff_ns=self.continuity.cutoff_ns, source_now_ns=self._source_clock_ns() ) self.continuity.phase, self.continuity.reason = "active", None except StreamSuspended: raise except BaseException: self.request_stop("failed") raise finally: with self._lock: self._active["resync"] = 0 def admit(self, start: StreamStart, bundle: StreamBundle) -> bool: with self._lock: self.check_input(start) return self.mailbox.put(bundle) def admit_reserved( self, start: StreamStart, bundle: StreamBundle, reservation: IngressReservation ) -> bool: with self._lock: self.check_input(start) return self.mailbox.put_reserved(bundle, reservation) @contextmanager def work(self, start: StreamStart, lane: Literal["gpu", "cpu"]) -> Iterator[None]: with self._lock: self.check_input(start) if lane not in self._active or self._active[lane]: raise WorkerLeaseError("only one active call per compute lane") self._active[lane] += 1 try: yield self.check_input(start) except StreamSuspended: raise # The caller discards this epoch's result, not its models. except BaseException: self.request_stop("failed") raise finally: with self._lock: self._active[lane] -= 1 def check_current(self, start: StreamStart, *, starting: bool = False) -> None: """Resource/IPC guard: drain active RPC on pause; work/result gates fence it. Only child startup handshakes may opt into the warmup state. """ with self._lock: self._check(start, starting=starting, allow_unavailable=self.continuity is not None) def validate_result_binding(self, value: object) -> None: """Recheck at receipt/use; accepting a hash alone cannot renew authority.""" start = StreamStart.from_dict(value) self.check_input(start) def request_stop(self, reason: str = "cancelled") -> None: if reason not in ( "completed", "cancelled", "lease-lost", "child-exited", "invalid-process-group", "worker-not-ready", "failed", ): raise ValueError("unknown runtime stop reason") with self._lock: if self.state in (GraphState.STOPPED, GraphState.CANCELLED, GraphState.FAILED): return self.reason = self.reason or reason if self.stop_requested_ns is None: self.stop_requested_ns = time.monotonic_ns() self.state = GraphState.STOPPING self.stop_event.set() self.mailbox.cancel() def _watch(self) -> None: while not self._watchdog_stop.wait(0.05): try: with self._lock: if self.state in (GraphState.STARTING, GraphState.RUNNING): self._check(self.start, starting=True, allow_unavailable=True) if any(p.poll() is not None for p in self._children): self.request_stop("child-exited") stopping = self.state == GraphState.STOPPING if stopping: self.stop_children() return except (WorkerLeaseError, OSError): self.request_stop("lease-lost") self.stop_children() return def stop_children(self) -> bool: """Stop only owned groups. No Boolean resource-release attestation input.""" with self._cleanup_lock: with self._lock: children = tuple(self._children) for sent_signal, grace in ((signal.SIGTERM, 4.0), (signal.SIGKILL, 1.0)): for process in reversed(children): if _group_exists(process.pid): with suppress(ProcessLookupError): os.killpg(process.pid, sent_signal) deadline = time.monotonic() + grace while True: complete = all( p.poll() is not None and not _group_exists(p.pid) for p in children ) if complete or time.monotonic() >= deadline: break time.sleep(0.01) if complete: return True return False def close(self, reason: str = "completed") -> bool: self.request_stop(reason) self._watchdog_stop.set() children_stopped = self.stop_children() if self._watchdog is not threading.current_thread(): self._watchdog.join(timeout=0.1) with self._lock: if self.lease.released: return True # Caller joins/finishes its adapters; active callbacks and payloads # continue to fence even if all GPU subprocesses have died already. if ( not children_stopped or any(self._active.values()) or any(t.is_alive() for t in self._threads) or self._watchdog.is_alive() or not self.mailbox.quiescent ): return False self.lease._retire_after_verified_stop() self.retired_ns = time.monotonic_ns() self.state = ( GraphState.STOPPED if self.reason == "completed" else GraphState.CANCELLED if self.reason == "cancelled" else GraphState.FAILED ) return True def snapshot(self) -> dict[str, Any]: with self._lock: return { "state": self.state.value, "reason": self.reason, "start": self.start.to_dict(), "active_calls": dict(self._active), "lease_released": self.lease.released, "input_payloads_released": self.mailbox.quiescent, "owned_children": len(self._children), "live_children": sum(p.poll() is None for p in self._children), "actuation_allowed": False, "lease_renewals": self.lease.renewals, "lease_deadline_monotonic_ns": _wire_integer(self.lease.deadline_ns), "stop_requested_monotonic_ns": ( _wire_integer(self.stop_requested_ns) if self.stop_requested_ns is not None else None ), "retired_monotonic_ns": ( _wire_integer(self.retired_ns) if self.retired_ns is not None else None ), "event_clock": "worker-process-monotonic", "input_continuity": self.continuity.snapshot() if self.continuity is not None else {"enabled": False}, "worker_readiness": ( self._readiness.snapshot() if self._readiness is not None else {"enabled": False, "realtime_qualified": False} ), }