diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_queue.py b/experiments/perception/worker/streaming_profile_stage1/pilot_queue.py
index 55820df..1224c63 100644
--- a/experiments/perception/worker/streaming_profile_stage1/pilot_queue.py
+++ b/experiments/perception/worker/streaming_profile_stage1/pilot_queue.py
@@ -1,74 +1,5 @@
-"""One producer, one graph, bounded pending queue and active payload accounting."""
+"""Compatibility import; the bounded mailbox belongs to the shared runtime."""
 
-from collections import deque
-from threading import Condition
+from k1link.perception.streaming_queue import StreamMailbox as Mailbox
 
-
-class Mailbox:
-    def __init__(self, capacity=2, byte_limit=16 * 1024 * 1024):
-        self.capacity = capacity
-        self.byte_limit = byte_limit
-        self.condition = Condition()
-        self.pending = deque()
-        self.bytes = 0
-        self.peak_bytes = 0
-        self.peak_pending = 0
-        self.external_pending = 0
-        self.dropped = []
-        self.done = False
-        self.error = None
-
-    def put(self, bundle):
-        size = bundle["payload_bytes"]
-        with self.condition:
-            while self.pending and (
-                len(self.pending) + self.external_pending >= self.capacity
-                or self.bytes + size > self.byte_limit
-            ):
-                old = self.pending.popleft()
-                self.bytes -= old["payload_bytes"]
-                self.dropped.append({"sequence": old["sequence"], "reason": "pending-overflow"})
-            if len(self.pending) + self.external_pending >= self.capacity:
-                self.dropped.append({"sequence": bundle["sequence"], "reason": "pending-overflow"})
-                return
-            if self.bytes + size > self.byte_limit:
-                self.dropped.append({"sequence": bundle["sequence"], "reason": "byte-budget"})
-                return
-            self.pending.append(bundle)
-            self.bytes += size
-            self.peak_bytes = max(self.peak_bytes, self.bytes)
-            self.peak_pending = max(self.peak_pending, len(self.pending) + self.external_pending)
-            self.condition.notify_all()
-
-    def reserve_completed(self):
-        """A finished GPU result shares the SAME pending budget as ingress."""
-        with self.condition:
-            if self.external_pending:
-                raise ValueError("only one completed GPU slot is permitted")
-            while len(self.pending) + 1 > self.capacity:
-                old = self.pending.popleft()
-                self.bytes -= old["payload_bytes"]
-                self.dropped.append({"sequence": old["sequence"], "reason": "handoff-overflow"})
-            self.external_pending = 1
-            self.peak_pending = max(self.peak_pending, len(self.pending) + 1)
-
-    def take_completed(self):
-        with self.condition:
-            if self.external_pending != 1:
-                raise ValueError("completed GPU slot accounting mismatch")
-            self.external_pending = 0
-
-    def take(self):
-        with self.condition:
-            self.condition.wait_for(lambda: self.pending or self.done)
-            return self.pending.popleft() if self.pending else None
-
-    def release(self, bundle):
-        with self.condition:
-            self.bytes -= bundle["payload_bytes"]
-
-    def finish(self, error=None):
-        with self.condition:
-            self.done = True
-            self.error = error
-            self.condition.notify_all()
+__all__ = ["Mailbox"]
diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_scheduler.py b/experiments/perception/worker/streaming_profile_stage1/pilot_scheduler.py
index 2e15594..52efd7a 100644
--- a/experiments/perception/worker/streaming_profile_stage1/pilot_scheduler.py
+++ b/experiments/perception/worker/streaming_profile_stage1/pilot_scheduler.py
@@ -1,91 +1,5 @@
-"""One serial GPU stage can overlap one chronological CPU fusion stage.
+"""Compatibility import; no separate per-LAB GPU scheduling implementation."""
 
-No GPU model concurrency. Ingress and completed-GPU results share two pending
-slots. Active payloads remain in the shared mailbox byte budget.
-"""
+from k1link.perception.streaming_scheduler import SerialGpuStage as GpuStage
 
-import threading
-import traceback
-
-
-class GpuStage:
-    def __init__(self, mailbox, compute, stop):
-        self.mailbox = mailbox
-        self.compute = compute
-        self.stop = stop
-        self.output_condition = threading.Condition()
-        self.completed = None
-        self.completed_reserved = False
-        self.consumer_waiting = False
-        # Reserve the output slot BEFORE starting another GPU call. Otherwise a
-        # blocked put would hide a third pending frame outside the two queues.
-        self.output_slot = threading.Semaphore(1)
-        self.finished = threading.Event()
-        self.error = None
-        self.peak_pending = 0
-        self.direct_handoffs = 0
-        self.buffered_handoffs = 0
-        self.thread = threading.Thread(target=self._run, daemon=True)
-        self.thread.start()
-
-    def _run(self):
-        try:
-            while not self.stop.is_set():
-                if not self.output_slot.acquire(timeout=0.05):
-                    continue
-                bundle = self.mailbox.take()
-                if bundle is None:
-                    break
-                result = self.compute(bundle)
-                with self.output_condition:
-                    if self.completed is not None:
-                        raise ValueError("completed GPU slot is already occupied")
-                    # A receiver already blocked in take() accepts the result as
-                    # a synchronous rendezvous. It never becomes pending and
-                    # therefore must not evict a newer ingress frame merely for
-                    # the few instructions between publish and receive.
-                    reserved = not self.consumer_waiting
-                    if reserved:
-                        self.mailbox.reserve_completed()
-                        self.buffered_handoffs += 1
-                    else:
-                        self.direct_handoffs += 1
-                    self.completed = (bundle, result)
-                    self.completed_reserved = reserved
-                    self.peak_pending = max(self.peak_pending, int(reserved))
-                    self.output_condition.notify_all()
-        except Exception:
-            self.error = traceback.format_exc()
-        finally:
-            self.finished.set()
-            with self.output_condition:
-                self.output_condition.notify_all()
-
-    def take(self):
-        while True:
-            with self.output_condition:
-                if self.completed is not None:
-                    result = self.completed
-                    reserved = self.completed_reserved
-                    self.completed = None
-                    self.completed_reserved = False
-                    self.consumer_waiting = False
-                elif self.finished.is_set():
-                    self.consumer_waiting = False
-                    if self.error:
-                        raise RuntimeError(self.error) from None
-                    return None
-                else:
-                    self.consumer_waiting = True
-                    self.output_condition.wait(timeout=0.05)
-                    continue
-            if reserved:
-                self.mailbox.take_completed()
-            self.output_slot.release()
-            return result
-
-    def close(self):
-        self.stop.set()
-        self.mailbox.finish(self.mailbox.error)
-        self.thread.join(timeout=2)
-        return not self.thread.is_alive()
+__all__ = ["GpuStage"]
diff --git a/src/k1link/perception/streaming_queue.py b/src/k1link/perception/streaming_queue.py
new file mode 100644
index 0000000..1f72f83
--- /dev/null
+++ b/src/k1link/perception/streaming_queue.py
@@ -0,0 +1,147 @@
+"""Bounded stream ingress shared by serial GPU and chronological CPU stages.
+
+No source duration, recording path, model, lease acquisition, or output history.
+Active inputs remain byte-accounted until their owner releases them. A completed
+GPU result shares the pending budget with ingress; it is not a third queue.
+These are scheduler primitives, not a replacement for graph terminal outcomes.
+"""
+
+from __future__ import annotations
+
+from collections import Counter, deque
+from collections.abc import Mapping
+from threading import Condition
+from typing import Any
+
+StreamBundle = Mapping[str, Any]
+DISCARD_REASONS = frozenset(("cancelled", "gpu-failed", "processing-failed"))
+
+
+class StreamMailbox:
+    def __init__(self, capacity: int = 2, byte_limit: int = 16 * 1024 * 1024) -> None:
+        if type(capacity) is not int or not 1 <= capacity <= 2:
+            raise ValueError("pending capacity must be 1..2")
+        if type(byte_limit) is not int or not 1 <= byte_limit <= 16 * 1024 * 1024:
+            raise ValueError("inflight byte budget must be 1..16 MiB")
+        self.capacity, self.byte_limit = capacity, byte_limit
+        self.condition = Condition()
+        self.pending: deque[StreamBundle] = deque()
+        self.bytes = self.peak_bytes = self.peak_pending = self.external_pending = 0
+        self.drop_counts: Counter[str] = Counter()
+        self._recent_drops: deque[dict[str, Any]] = deque(maxlen=256)
+        self._owned: dict[int, tuple[StreamBundle, int]] = {}
+        self._active: set[int] = set()
+        self._last_sequence = -1
+        self.done = False
+        self.error: str | None = None
+
+    @property
+    def dropped(self) -> list[dict[str, Any]]:
+        """Bounded diagnostic tail, NOT an all-session terminal journal."""
+        with self.condition:
+            return list(self._recent_drops)
+
+    @property
+    def dropped_count(self) -> int:
+        with self.condition:
+            return sum(self.drop_counts.values())
+
+    @property
+    def quiescent(self) -> bool:
+        with self.condition:
+            return self.done and not self._owned and not self.external_pending
+
+    def _drop(self, bundle: StreamBundle, reason: str) -> None:
+        self.drop_counts[reason] += 1
+        self._recent_drops.append({"sequence": bundle["sequence"], "reason": reason})
+
+    def _discard_pending(self, reason: str) -> None:
+        old = self.pending.popleft()
+        _, size = self._owned.pop(id(old))
+        self.bytes -= size
+        self._drop(old, reason)
+
+    def put(self, bundle: StreamBundle) -> bool:
+        size, sequence = bundle["payload_bytes"], bundle["sequence"]
+        if type(size) is not int or size <= 0 or type(sequence) is not int or sequence < 0:
+            raise ValueError("invalid source sequence or payload size")
+        with self.condition:
+            if id(bundle) in self._owned:
+                raise ValueError("input is already owned")
+            if sequence <= self._last_sequence:
+                raise ValueError("stream sequence must strictly increase")
+            self._last_sequence = sequence
+            if self.done or size > self.byte_limit:
+                self._drop(bundle, "ingress-closed" if self.done else "byte-budget")
+                return False
+            while self.pending and (
+                len(self.pending) + self.external_pending >= self.capacity
+                or self.bytes + size > self.byte_limit
+            ):
+                self._discard_pending("pending-overflow")
+            if len(self.pending) + self.external_pending >= self.capacity:
+                self._drop(bundle, "pending-overflow")
+                return False
+            if self.bytes + size > self.byte_limit:
+                self._drop(bundle, "byte-budget")
+                return False
+            self.pending.append(bundle)
+            self._owned[id(bundle)] = (bundle, size)
+            self.bytes += size
+            self.peak_bytes = max(self.peak_bytes, self.bytes)
+            self.peak_pending = max(self.peak_pending, len(self.pending) + self.external_pending)
+            self.condition.notify_all()
+            return True
+
+    def reserve_completed(self) -> None:
+        with self.condition:
+            if self.external_pending:
+                raise ValueError("only one completed GPU slot is permitted")
+            while len(self.pending) + 1 > self.capacity:
+                self._discard_pending("handoff-overflow")
+            self.external_pending = 1
+            self.peak_pending = max(self.peak_pending, len(self.pending) + 1)
+
+    def take_completed(self) -> None:
+        with self.condition:
+            if self.external_pending != 1:
+                raise ValueError("completed GPU slot accounting mismatch")
+            self.external_pending = 0
+
+    def take(self) -> StreamBundle | None:
+        with self.condition:
+            self.condition.wait_for(lambda: self.pending or self.done)
+            if not self.pending:
+                return None
+            bundle = self.pending.popleft()
+            self._active.add(id(bundle))
+            return bundle
+
+    def release(self, bundle: StreamBundle, *, discard_reason: str | None = None) -> None:
+        if discard_reason is not None and discard_reason not in DISCARD_REASONS:
+            raise ValueError("unknown discard reason")
+        with self.condition:
+            if id(bundle) not in self._active or self._owned[id(bundle)][0] is not bundle:
+                raise ValueError("input release ownership mismatch")
+            self._active.remove(id(bundle))
+            _, size = self._owned.pop(id(bundle))
+            self.bytes -= size
+            if discard_reason is not None:
+                self._drop(bundle, discard_reason)
+
+    def finish(self, error: str | None = None) -> None:
+        """Close ingress, allowing accepted pending work to drain."""
+        with self.condition:
+            self.done = True
+            self.error = self.error or error
+            self.condition.notify_all()
+
+    def cancel(self, reason: str = "cancelled") -> None:
+        """Discard pending work, but never free another stage's active input."""
+        if reason not in DISCARD_REASONS:
+            raise ValueError("unknown cancellation reason")
+        with self.condition:
+            self.done = True
+            while self.pending:
+                self._discard_pending(reason)
+            self.condition.notify_all()
diff --git a/src/k1link/perception/streaming_scheduler.py b/src/k1link/perception/streaming_scheduler.py
new file mode 100644
index 0000000..e996bbe
--- /dev/null
+++ b/src/k1link/perception/streaming_scheduler.py
@@ -0,0 +1,116 @@
+"""One serial GPU lane overlapped with one chronological CPU consumer.
+
+The callback owns all GPU model calls for a frame (e.g. DDRNet then RF-DETR).
+This class does not acquire the Worker lease or prove external GPU exclusivity.
+Close is cooperative: False means an active callback still owns resources and
+the supervisor must not admit another profile. No thread is forcefully killed.
+"""
+
+from __future__ import annotations
+
+import threading
+import traceback
+from collections.abc import Callable
+from typing import Any
+
+from .streaming_queue import StreamBundle, StreamMailbox
+
+
+class SerialGpuStage:
+    def __init__(
+        self, mailbox: StreamMailbox, compute: Callable[[StreamBundle], Any], stop: threading.Event
+    ) -> None:
+        self.mailbox, self.compute, self.stop = mailbox, compute, stop
+        self.output_condition = threading.Condition()
+        self.completed: tuple[StreamBundle, Any] | None = None
+        self.completed_reserved = self.consumer_waiting = False
+        self.output_slot = threading.Semaphore(1)
+        self.finished = threading.Event()
+        self.error: str | None = None
+        self.peak_pending = self.direct_handoffs = self.buffered_handoffs = 0
+        self.thread = threading.Thread(target=self._run, name="perception-serial-gpu", daemon=True)
+        self.thread.start()
+
+    def _run(self) -> None:
+        active = None
+        try:
+            while not self.stop.is_set():
+                # Reserve output BEFORE inference: no hidden third pending result.
+                if not self.output_slot.acquire(timeout=0.05):
+                    continue
+                active = self.mailbox.take()
+                if active is None:
+                    break
+                if self.stop.is_set():
+                    break
+                result = self.compute(active)
+                with self.output_condition:
+                    if self.stop.is_set():
+                        break
+                    if self.completed is not None:
+                        raise ValueError("completed GPU slot is already occupied")
+                    # A blocked consumer accepts a synchronous rendezvous, not
+                    # a pending slot that would spuriously evict newer ingress.
+                    reserved = not self.consumer_waiting
+                    if reserved:
+                        self.mailbox.reserve_completed()
+                        self.buffered_handoffs += 1
+                    else:
+                        self.direct_handoffs += 1
+                    self.completed = (active, result)
+                    active = None
+                    self.completed_reserved = reserved
+                    self.peak_pending = max(self.peak_pending, int(reserved))
+                    self.output_condition.notify_all()
+        except Exception:
+            self.error = traceback.format_exc()
+            self.mailbox.finish(self.error)
+            self.mailbox.cancel("gpu-failed")
+        finally:
+            if active is not None:
+                self.mailbox.release(
+                    active, discard_reason="gpu-failed" if self.error else "cancelled"
+                )
+            self.finished.set()
+            with self.output_condition:
+                self.output_condition.notify_all()
+
+    def take(self) -> tuple[StreamBundle, Any] | None:
+        while True:
+            with self.output_condition:
+                if self.stop.is_set():
+                    self.consumer_waiting = False
+                    return None
+                if self.completed is not None:
+                    result = self.completed
+                    reserved = self.completed_reserved
+                    self.completed = None
+                    self.completed_reserved = self.consumer_waiting = False
+                elif self.finished.is_set():
+                    self.consumer_waiting = False
+                    if self.error:
+                        raise RuntimeError(self.error) from None
+                    return None
+                else:
+                    self.consumer_waiting = True
+                    self.output_condition.wait(timeout=0.05)
+                    continue
+            if reserved:
+                self.mailbox.take_completed()
+            self.output_slot.release()
+            return result
+
+    def close(self, *, timeout: float = 2.0) -> bool:
+        if not 0 <= timeout <= 2:
+            raise ValueError("stop timeout must be 0..2 seconds")
+        self.stop.set()
+        self.mailbox.cancel()
+        self.thread.join(timeout=timeout)
+        with self.output_condition:
+            if self.completed is not None:
+                if self.completed_reserved:
+                    self.mailbox.take_completed()
+                self.mailbox.release(self.completed[0], discard_reason="cancelled")
+                self.completed = None
+                self.completed_reserved = False
+        return not self.thread.is_alive()
diff --git a/tests/test_perception_streaming_runtime.py b/tests/test_perception_streaming_runtime.py
new file mode 100644
index 0000000..6e639cb
--- /dev/null
+++ b/tests/test_perception_streaming_runtime.py
@@ -0,0 +1,141 @@
+"""Small synthetic scheduler lifecycle tests; no models, GPU or source IO."""
+
+import threading
+
+import pytest
+
+from k1link.perception.streaming_queue import StreamMailbox
+from k1link.perception.streaming_scheduler import SerialGpuStage
+
+
+def bundle(sequence, size=10):
+    return {"sequence": sequence, "payload_bytes": size}
+
+
+def test_finish_drains_but_cancel_drops_pending_without_freeing_active_owner():
+    queue = StreamMailbox()
+    first = bundle(0)
+    queue.put(first)
+    assert queue.take() is first
+    queue.put(bundle(1))
+    queue.finish()
+    assert not queue.put(bundle(2))
+    assert queue.bytes == 20 and not queue.quiescent
+    queue.cancel()
+    assert queue.take() is None and queue.bytes == 10
+    assert queue.drop_counts == {"ingress-closed": 1, "cancelled": 1}
+    queue.release(first)
+    assert queue.quiescent
+    with pytest.raises(ValueError, match="ownership"):
+        queue.release(first)
+
+
+def test_oversized_message_does_not_evict_accepted_small_input():
+    queue = StreamMailbox(byte_limit=100)
+    first = bundle(0)
+    queue.put(first)
+    assert not queue.put(bundle(1, 101))
+    assert queue.take() is first and queue.bytes == 10
+    queue.release(first)
+
+
+def test_diagnostic_history_is_bounded_and_total_drops_remain_exact():
+    queue = StreamMailbox()
+    # Boundary-sized synthetic fixture, not a local sustained-load benchmark.
+    for sequence in range(260):
+        queue.put(bundle(sequence))
+    assert queue.dropped_count == 258 and len(queue.dropped) == 256
+    assert queue.peak_pending == 2 and queue.bytes == 20
+    queue.cancel()
+    assert queue.dropped_count == 260 and queue.quiescent
+
+
+@pytest.mark.parametrize("sequence,size", [(True, 1), (-1, 1), (0, 0), (0, True)])
+def test_invalid_admission_is_rejected(sequence, size):
+    queue = StreamMailbox()
+    with pytest.raises(ValueError):
+        queue.put(bundle(sequence, size))
+    assert queue.bytes == 0
+
+
+def test_release_uses_admitted_size_and_does_not_allow_reusing_owned_input():
+    queue = StreamMailbox()
+    first = bundle(0)
+    queue.put(first)
+    assert queue.take() is first
+    first.update(sequence=1, payload_bytes=1000)
+    with pytest.raises(ValueError, match="owned"):
+        queue.put(first)
+    with pytest.raises(ValueError, match="ownership"):
+        queue.release(dict(first))
+    queue.release(first)
+    assert queue.bytes == 0
+    with pytest.raises(ValueError, match="increase"):
+        queue.put(bundle(0))
+
+
+def test_failed_gpu_releases_its_input_and_discards_pending():
+    started, proceed = threading.Event(), threading.Event()
+    queue = StreamMailbox()
+
+    def compute(_bundle):
+        started.set()
+        assert proceed.wait(1)
+        raise ValueError("synthetic failure")
+
+    stage = SerialGpuStage(queue, compute, threading.Event())
+    try:
+        queue.put(bundle(0))
+        assert started.wait(1)
+        queue.put(bundle(1))
+        proceed.set()
+        with pytest.raises(RuntimeError, match="synthetic failure"):
+            stage.take()
+        assert queue.quiescent and queue.bytes == 0
+        assert queue.drop_counts == {"gpu-failed": 2}
+    finally:
+        proceed.set()
+        assert stage.close()
+
+
+def test_timed_out_stop_retains_ownership_until_callback_actually_exits():
+    started, proceed = threading.Event(), threading.Event()
+    queue = StreamMailbox()
+
+    def compute(_bundle):
+        started.set()
+        assert proceed.wait(1)
+        return "must not be published after cancellation"
+
+    stage = SerialGpuStage(queue, compute, threading.Event())
+    try:
+        queue.put(bundle(0))
+        assert started.wait(1)
+        assert not stage.close(timeout=0)
+        assert queue.bytes == 10 and not queue.quiescent
+        assert stage.take() is None
+        proceed.set()
+        assert stage.finished.wait(1)
+        assert stage.close() and queue.quiescent
+        assert queue.drop_counts == {"cancelled": 1}
+    finally:
+        proceed.set()
+        assert stage.close()
+
+
+def test_stop_discards_completed_result_but_cannot_free_cpu_owned_input():
+    queue = StreamMailbox()
+    stage = SerialGpuStage(queue, lambda item: item["sequence"], threading.Event())
+    try:
+        queue.put(bundle(0))
+        first, _ = stage.take()
+        queue.put(bundle(1))
+        with stage.output_condition:
+            assert stage.output_condition.wait_for(lambda: stage.completed is not None, timeout=1)
+        assert queue.bytes == 20
+        assert stage.close()
+        assert queue.bytes == 10 and not queue.quiescent
+        queue.release(first)
+        assert queue.quiescent
+    finally:
+        assert stage.close()