feat(perception): preserve resident runtime across recoverable input gaps

src/k1link/perception/streaming_continuity.py

@@ -0,0 +1,92 @@
+"""Recoverable input epochs; resident model ownership is a separate lifetime.
+
+Used only by the explicitly enabled resumable adapter. The controller supplies
+the source-clock mapping and the profile adapter supplies verified decoder and
+sensor evidence. No socket, GPU, vehicle command or unbounded history lives here.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from hashlib import sha256
+
+from .realtime_contract import StreamStart, _integer
+
+
+class StreamSuspended(RuntimeError):
+    """Discard obsolete work, keep the resident model processes and lease."""
+
+
+@dataclass(frozen=True, slots=True)
+class ResumeEvidence:
+    camera_ns: int
+    pose_ns: int
+    oldest_points_ns: int
+    newest_points_ns: int
+    decoded_keyframe: bool
+
+    def validate(self, *, cutoff_ns: int, source_now_ns: int) -> None:
+        for name in ("camera_ns", "pose_ns", "oldest_points_ns", "newest_points_ns"):
+            _integer(getattr(self, name), name)
+        _integer(source_now_ns, "source clock")
+        if (
+            self.decoded_keyframe is not True
+            or not cutoff_ns <= self.pose_ns <= self.camera_ns <= source_now_ns
+            or not cutoff_ns <= self.oldest_points_ns <= self.newest_points_ns <= self.camera_ns
+            or source_now_ns - self.camera_ns > 250_000_000
+            or self.camera_ns - self.pose_ns > 100_000_000
+            or self.camera_ns - self.newest_points_ns > 100_000_000
+            or self.camera_ns - self.oldest_points_ns > 250_000_000
+            or max(
+                abs(self.pose_ns - self.oldest_points_ns), abs(self.pose_ns - self.newest_points_ns)
+            )
+            > 100_000_000
+        ):
+            raise StreamSuspended("fresh keyframe and causal camera/pose/points required")
+
+
+class InputContinuity:
+    """Access is serialized by StreamingLifecycle's lock."""
+
+    def __init__(self, activation: StreamStart) -> None:
+        self.activation = self.epoch = activation
+        self.phase = "active"
+        self.generation = self.pauses = 0
+        self.reason: str | None = None
+        self.cutoff_ns = 0
+
+    def check(self, epoch: StreamStart, *, synchronizing: bool = False) -> None:
+        if epoch != self.epoch:
+            raise StreamSuspended("obsolete input epoch")
+        if self.phase != "active" and not (synchronizing and self.phase == "synchronizing"):
+            raise StreamSuspended("input is waiting for resynchronization")
+
+    def pause(self, reason: str) -> None:
+        if reason not in ("input-disconnected", "input-timeout", "source-gap", "worker-telemetry"):
+            raise ValueError("unknown recoverable pause reason")
+        if self.phase != "waiting":
+            self.pauses += 1
+        self.phase, self.reason = "waiting", reason
+
+    def begin(self, source_now_ns: int) -> StreamStart:
+        _integer(source_now_ns, "source clock")
+        if self.phase != "waiting" or source_now_ns < self.cutoff_ns:
+            raise StreamSuspended("new input needs a pause and nonregressing source clock")
+        self.generation += 1
+        identity = sha256(f"{self.activation.epoch_id}:{self.generation}".encode()).hexdigest()
+        # Wire v2 already binds every packet to the complete StreamStart. Change
+        # the input epoch, NOT the resident lease generation/model activation.
+        self.epoch = replace(self.activation, epoch_id="input-" + identity)
+        self.cutoff_ns, self.phase = source_now_ns, "synchronizing"
+        return self.epoch
+
+    def snapshot(self) -> dict[str, object]:
+        return {
+            "phase": self.phase,
+            "reason": self.reason,
+            "input_generation": self.generation,
+            "pauses": self.pauses,
+            "input_start": self.epoch.to_dict(),
+            "minimum_source_ns": str(self.cutoff_ns),
+            "actuation_allowed": False,
+        }

src/k1link/perception/streaming_decoder.py

@@ -105,6 +105,18 @@ class FragmentDecoder:
             self.failed = True
             raise
 
+    def reset(self, payload: bytes) -> None:
+        """New input epoch only: release codec references, retain the child.
+
+        A genuine decoder failure is not revived by reset. The first new frame
+        still has to independently pass random-access/keyframe validation.
+        """
+        if self.failed:
+            raise StreamingDecodeError("failed decoder cannot be resumed")
+        self._codec = self._init = self._timing = None
+        self._next_dts, self.frames = None, 0
+        self.configure(payload)
+
     def close(self) -> None:
         self.failed = True
         self._codec = self._init = self._timing = None

src/k1link/perception/streaming_decoder_client.py

@@ -6,6 +6,7 @@ import math
 from typing import Any
 
 from .graph_contracts import GraphState
+from .realtime_contract import StreamStart
 from .streaming_lifecycle import StreamingLifecycle
 from .streaming_pipe_rpc import HEADER_SCRATCH, MAX_OUTPUT, BoundedPipeRpc, PipeRpcError
 
@@ -57,6 +58,19 @@ class StreamingDecoderClient:
         self.frames += 1
         return result
 
+    def reset(self, epoch: StreamStart, raw: bytes) -> None:
+        self.runtime.check_input(epoch, synchronizing=True)
+        if (
+            self.closed
+            or self.runtime.continuity is None
+            or self.runtime.continuity.phase != "synchronizing"
+            or not 0 < len(raw) <= 65536
+        ):
+            raise PipeRpcError("decoder reset requires a new input epoch")
+        if self.rpc.exchange({"op": "reset"}, raw, bytearray()) != {"reset": True}:
+            raise PipeRpcError("decoder reset response changed")
+        self.frames = 0
+
     def close(self) -> None:
         # Caller joins its callback and stops the owned child before closing.
         if not self.closed:

src/k1link/perception/streaming_ingress.py

@@ -1,7 +1,9 @@
 """One controller-supplied IPC socket feeding the common streaming lifecycle.
 
-No listener, acquisition, archive reader, model RPC, network authentication or
-automatic reconnect. Reconnect needs a new StreamStart/lease and decoder state.
+No listener, acquisition, archive reader, model RPC or network authentication.
+An explicitly resumable runtime preserves its resident lease on disconnect;
+the controller supplies a new input epoch/socket and the adapter resets decoder
+and temporal state. The historical non-resumable diagnostic remains supported.
 The synchronous consumer is a trusted adapter: raw events are borrowed for that
 call only. It must reserve decoder scratch before allocation and retain decoded
 work only through the common mailbox. Slow/failed consumers fail this stream;
@@ -16,15 +18,22 @@ import threading
 import time
 from collections import Counter
 from collections.abc import Callable
+from contextlib import suppress
 from hashlib import sha256
 from typing import Any
 
 from k1link.compute.live_perception import LiveIngressEvent
 
 from . import streaming_wire as wire
+from .realtime_contract import StreamStart
+from .streaming_continuity import StreamSuspended
 from .streaming_lifecycle import StreamingLifecycle
 
 
+class InputInterrupted(wire.StreamWireError):
+    """Transport unavailable, not malformed model data or an ownership loss."""
+
+
 class StreamingIngress:
     def __init__(
         self,
@@ -37,6 +46,7 @@ class StreamingIngress:
         *,
         fragment_timeout: float = 0.25,
         idle_timeout: float = 2.0,
+        input_epoch: StreamStart | None = None,
     ) -> None:
         if not 0 < fragment_timeout <= 2 or not 0 < idle_timeout <= 30:
             raise ValueError("invalid bounded ingress timeouts")
@@ -49,7 +59,8 @@ class StreamingIngress:
         self.session_id, self.session_generation = session_id, session_generation
         self.consume, self.notice = consume, notice
         self.fragment_timeout, self.idle_timeout = fragment_timeout, idle_timeout
-        self.binding = wire.binding(runtime.start)
+        self.input_epoch = input_epoch or runtime.start
+        self.binding = wire.binding(self.input_epoch)
         self.opened = False
         self.error: str | None = None
         self.terminal: str | None = None
@@ -60,10 +71,10 @@ class StreamingIngress:
         self.thread = threading.Thread(
             target=self._serve, name="perception-binary-ingress", daemon=True
         )
-        runtime.track_thread(self.thread)
+        runtime.register_ingress(self.thread, self.input_epoch)
 
     def start(self) -> None:
-        self.runtime.check_current(self.runtime.start)
+        self.runtime.check_input(self.input_epoch, synchronizing=True)
         self.thread.start()
 
     def join(self, timeout: float = 1.0) -> bool:
@@ -71,9 +82,9 @@ class StreamingIngress:
         return not self.thread.is_alive()
 
     def _check(self, deadline: float) -> None:
-        self.runtime.check_current(self.runtime.start)
+        self.runtime.check_input(self.input_epoch, synchronizing=True)
         if time.monotonic() >= deadline:
-            raise wire.StreamWireError("ingress deadline exceeded")
+            raise InputInterrupted("ingress deadline exceeded")
 
     def _read_into(self, target: memoryview, deadline: float) -> None:
         offset = 0
@@ -87,8 +98,10 @@ class StreamingIngress:
                 read = self.connection.recv_into(target[offset:])
             except BlockingIOError:
                 continue
+            except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError) as exc:
+                raise InputInterrupted("input connection lost") from exc
             if read == 0:
-                raise wire.StreamWireError("unexpected EOF without explicit End")
+                raise InputInterrupted("unexpected EOF without explicit End")
             self.counts["wire_bytes"] += read
             offset += read
         self._check(deadline)
@@ -109,8 +122,10 @@ class StreamingIngress:
 
     def _bound(self, header: dict[str, Any]) -> None:
         if header.get("binding") != self.binding:
+            if self.runtime.continuity is not None:
+                raise StreamSuspended("obsolete packet from another input epoch")
             raise wire.StreamWireError("frame belongs to another StreamStart")
-        self.runtime.check_current(self.runtime.start)
+        self.runtime.check_input(self.input_epoch, synchronizing=True)
 
     def _fragment(self, header: dict[str, Any]) -> tuple[dict[str, Any], int, int, int]:
         if set(header) != {
@@ -157,6 +172,8 @@ class StreamingIngress:
         if modality == "camera-frame" and not self.camera_initialized:
             raise wire.StreamWireError("camera frame requires new codec initialization")
         if modality == "camera-frame" and previous and event["source_sequence"] != previous[1] + 1:
+            if self.runtime.continuity is not None:
+                raise StreamSuspended("camera sequence gap requires resynchronization")
             raise wire.StreamWireError("camera sequence gap requires new codec initialization")
         self.counts["observations_started"] += 1
         reservation = self.runtime.mailbox.reserve_ingress(2 * total)
@@ -191,6 +208,10 @@ class StreamingIngress:
             if sha256(raw).hexdigest() != first["payload_sha256"]:
                 raise wire.StreamWireError("observation integrity mismatch")
             self._check(deadline)
+            if event["modality"] != "camera-init":
+                self.runtime.check_input(
+                    self.input_epoch, synchronizing=True, event_ns=event["received_monotonic_ns"]
+                )
             admitted = LiveIngressEvent(**event, payload=bytes(raw))
             raw = None  # Immutable callback bytes remain fully reserved.
             self.consume(admitted)
@@ -224,8 +245,10 @@ class StreamingIngress:
         self.counts["declared_gap_observations"] += wire.uint64(header["count"])
         if header["modality"].startswith("camera"):
             self.camera_initialized = False
+        if self.runtime.continuity is not None:
+            raise StreamSuspended("declared source gap requires resynchronization")
         self.notice(dict(header))
-        self.runtime.check_current(self.runtime.start)
+        self.runtime.check_input(self.input_epoch, synchronizing=True)
         self.counts["gap_notices"] += 1
 
     def _serve(self) -> None:
@@ -240,7 +263,7 @@ class StreamingIngress:
             expected = wire.parse_header(
                 bytearray(
                     wire.open_packet(
-                        self.runtime.start,
+                        self.input_epoch,
                         self.session_id,
                         self.session_generation,
                     )[wire.PREFIX.size :]
@@ -269,8 +292,21 @@ class StreamingIngress:
             self.error = str(exc)
             self.terminal = "failed"
             if self.opened:
-                self.runtime.mailbox.finish(self.error)
-                self.runtime.request_stop("failed")
+                if self.runtime.continuity is not None and isinstance(
+                    exc, (InputInterrupted, StreamSuspended)
+                ):
+                    self.terminal = "paused"
+                    # A late old socket cannot pause a replacement connection.
+                    with suppress(RuntimeError):  # Already replaced, stopping, or fenced.
+                        self.runtime.pause_input(
+                            self.input_epoch,
+                            "input-disconnected"
+                            if isinstance(exc, InputInterrupted)
+                            else "source-gap",
+                        )
+                else:
+                    self.runtime.mailbox.finish(self.error)
+                    self.runtime.request_stop("failed")
         finally:
             self.connection.close()
             if reservation is not None:

src/k1link/perception/streaming_lifecycle.py

@@ -21,10 +21,15 @@ 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
+from .worker_readiness import (
+    WorkerReadinessError,
+    WorkerReadinessMonitor,
+    WorkerTelemetryUnavailable,
+)
 
 
 def _group_exists(pgid: int) -> bool:
@@ -48,16 +53,25 @@ class StreamingLifecycle:
         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}
+        self._active = {"gpu": 0, "cpu": 0, "resync": 0}
         self.state = GraphState.CREATED
         self.reason: str | None = None
         self.stop_requested_ns: int | None = None
@@ -70,7 +84,9 @@ class StreamingLifecycle:
         )
         self._watchdog.start()
 
-    def _check(self, start: StreamStart, *, starting: bool = False) -> None:
+    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")
@@ -81,9 +97,11 @@ class StreamingLifecycle:
             if start == self.start:
                 self.request_stop("lease-lost")
             raise
-        self._check_readiness(require_warmup=self.state == GraphState.RUNNING)
+        self._check_readiness(
+            require_warmup=self.state == GraphState.RUNNING, allow_unavailable=allow_unavailable
+        )
 
-    def _check_readiness(self, *, require_warmup: bool) -> None:
+    def _check_readiness(self, *, require_warmup: bool, allow_unavailable: bool = False) -> None:
         if self._readiness is not None:
             try:
                 self._readiness.check(
@@ -91,6 +109,14 @@ class StreamingLifecycle:
                     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
@@ -102,7 +128,7 @@ class StreamingLifecycle:
         heartbeat are independent requirements. No host operations under lock.
         """
         with self._lock:
-            self._check(start, starting=True)
+            self._check(start, starting=True, allow_unavailable=True)
             if self._readiness is None:
                 raise WorkerReadinessError("worker readiness monitoring is not configured")
             try:
@@ -112,13 +138,19 @@ class StreamingLifecycle:
                     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)
+            self._check(start, starting=True, allow_unavailable=True)
             self.lease.renew(start)
 
     def ready(self) -> None:
@@ -148,32 +180,127 @@ class StreamingLifecycle:
     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 admit(self, start: StreamStart, bundle: StreamBundle) -> bool:
+    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 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(start)
+            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(start)
+            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_current(start)
+            self.check_input(start)
+        except StreamSuspended:
+            raise  # The caller discards this epoch's result, not its models.
         except BaseException:
             self.request_stop("failed")
             raise
@@ -182,14 +309,17 @@ class StreamingLifecycle:
                 self._active[lane] -= 1
 
     def check_current(self, start: StreamStart, *, starting: bool = False) -> None:
-        """Only child startup handshakes may opt into the warmup state."""
+        """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)
+            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_current(start)
+        self.check_input(start)
 
     def request_stop(self, reason: str = "cancelled") -> None:
         if reason not in (
@@ -217,7 +347,7 @@ class StreamingLifecycle:
             try:
                 with self._lock:
                     if self.state in (GraphState.STARTING, GraphState.RUNNING):
-                        self._check(self.start, starting=True)
+                        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
@@ -304,6 +434,9 @@ class StreamingLifecycle:
                     _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

src/k1link/perception/streaming_queue.py

@@ -14,7 +14,7 @@ from threading import Condition
 from typing import Any
 
 StreamBundle = Mapping[str, Any]
-DISCARD_REASONS = frozenset(("cancelled", "gpu-failed", "processing-failed"))
+DISCARD_REASONS = frozenset(("cancelled", "gpu-failed", "processing-failed", "input-gap"))
 
 
 class IngressReservation:
@@ -210,3 +210,21 @@ class StreamMailbox:
             while self.pending:
                 self._discard_pending(reason)
             self.condition.notify_all()
+
+    def pause(self) -> None:
+        """Drop queued work without EOF, freeing only queue-owned payloads."""
+        with self.condition:
+            while self.pending:
+                self._discard_pending("input-gap")
+            self.condition.notify_all()
+
+    def begin_epoch(self) -> None:
+        """Only after all prior queued/active/completed bundles are released.
+
+        Long-lived bounded decoder/sensor scratch reservations may remain.
+        The lifecycle separately verifies that the old ingress thread is gone.
+        """
+        with self.condition:
+            if self.done or self._owned or self.external_pending:
+                raise ValueError("old epoch still owns work or mailbox is closed")
+            self._last_sequence = -1

src/k1link/perception/streaming_scheduler.py

@@ -13,6 +13,7 @@ import traceback
 from collections.abc import Callable
 from typing import Any
 
+from .streaming_continuity import StreamSuspended
 from .streaming_queue import StreamBundle, StreamMailbox
 
 
@@ -43,7 +44,13 @@ class SerialGpuStage:
                     break
                 if self.stop.is_set():
                     break
-                result = self.compute(active)
+                try:
+                    result = self.compute(active)
+                except StreamSuspended:
+                    self.mailbox.release(active, discard_reason="input-gap")
+                    active = None
+                    self.output_slot.release()
+                    continue
                 with self.output_condition:
                     if self.stop.is_set():
                         break

src/k1link/perception/streaming_sensors.py

@@ -246,3 +246,12 @@ class CausalSensorWindow:
             self.pose = None
             self.closed = True
             self._reservation.release()
+
+    def reset(self) -> None:
+        """Called by the single adapter after old-epoch callbacks have drained."""
+        if self.closed:
+            raise ValueError("closed sensor window cannot resume")
+        self.rolling.clear()
+        self.fresh.clear()
+        self.pose = self.previous_camera_time = None
+        self._last_time = -1

src/k1link/perception/worker_readiness.py

@@ -8,6 +8,7 @@ clocks, or certify real-time operation. The lifecycle serializes all access.
 
 from __future__ import annotations
 
+from contextlib import suppress
 from dataclasses import asdict
 from typing import Literal
 
@@ -34,6 +35,10 @@ class WorkerReadinessError(WorkerLeaseError):
     """A current controller lost the prerequisites for further execution."""
 
 
+class WorkerTelemetryUnavailable(WorkerReadinessError):
+    """Readiness is unproved, but no confirmed ownership/runtime failure exists."""
+
+
 class WorkerReadinessMonitor:
     def __init__(
         self,
@@ -44,12 +49,14 @@ class WorkerReadinessMonitor:
         mode: ReadinessMode,
         clock_domain_id: str,
         now_monotonic_ns: int,
+        recoverable: bool = False,
     ) -> 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.recoverable = recoverable
         self._observed = initial
         self._last_check_ns = now_monotonic_ns
         self._failure: tuple[str, ...] = ()
@@ -97,7 +104,26 @@ class WorkerReadinessMonitor:
             if failure not in _PERFORMANCE_FAILURES
             and (require_warmup or failure != "warmup-not-complete")
         )
-        if fatal:
+        if self.recoverable:
+            # Missing metrics are not a proof that another process owns GPU.
+            # Known identity/conflict/clock failures still fence immediately.
+            uncertain = {"worker-snapshot-expired", "warmup-not-complete"}
+            for field in ("image_sha256", "effective_config_sha256"):
+                if getattr(self._observed, field) is None:
+                    uncertain.add(f"{field}-mismatch-or-unknown")
+            if self._observed.competing_gpu_clients is None:
+                uncertain.add("competing-gpu-clients-or-inventory-unknown")
+            if self._observed.gpu_owner_run_id in (
+                None,
+                start.run_id,
+            ) and self._observed.lease_generation in (None, start.lease_generation):
+                uncertain.add("exclusive-worker-lease-unproved")
+            confirmed = tuple(reason for reason in fatal if reason not in uncertain)
+            if confirmed:
+                self._fail(confirmed)
+            if fatal or (require_warmup and self.mode == "strict-envelope" and failures):
+                raise WorkerTelemetryUnavailable("worker readiness unproved: " + ",".join(failures))
+        elif fatal:
             self._fail(fatal)
         if require_warmup and self.mode == "strict-envelope" and failures:
             self._fail(failures)
@@ -112,7 +138,9 @@ class WorkerReadinessMonitor:
     ) -> 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)
+        # Fresh facts may restore readiness, not the separate lease/input epoch.
+        with suppress(WorkerTelemetryUnavailable):
+            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
@@ -126,6 +154,7 @@ class WorkerReadinessMonitor:
         return {
             "enabled": True,
             "mode": self.mode,
+            "recoverable": self.recoverable,
             "envelope": asdict(self.envelope),
             "worker_clock_domain_id": self.clock_domain_id,
             "last_check_monotonic_ns": str(self._last_check_ns),