diff --git a/omotion/ScanWorkflow.py b/omotion/ScanWorkflow.py
index b995475e..421abbd4 100644
--- a/omotion/ScanWorkflow.py
+++ b/omotion/ScanWorkflow.py
@@ -378,6 +378,7 @@ def start_scan(
         from omotion.pipeline.sinks import (
             CsvSink as PipelineCsvSink, DiagnosticsLogSink, ScanDBSink, ScanMetadata,
         )
+        from omotion.pipeline.async_sink import AsyncSink
         from omotion.pipeline.pedestal import SensorPedestals, pedestal_for_fw
 
         # Clear the prior scan's outcome up front so a refusal below (busy, or
@@ -479,8 +480,12 @@ def _safe_pedestal(sensor) -> float:
                     True if scan_db_path is None
                     else bool(request.write_corrected_csv)
                 )
+                # AsyncSink: run the CSV writes on a worker thread so a disk
+                # flush can't stall the runner's USB-drain loop and overflow
+                # the firmware histo queue (2026-06 soak root-cause).
                 default_sinks.append(
-                    PipelineCsvSink(output_dir=data_dir, write_corrected=write_corrected)
+                    AsyncSink(PipelineCsvSink(output_dir=data_dir,
+                                              write_corrected=write_corrected))
                 )
             if scan_db_path is not None:
                 # Pre-flight the scan DB before the laser fires. The DB is the
@@ -513,7 +518,11 @@ def _safe_pedestal(sensor) -> float:
                     with self._lock:
                         self._running = False
                     return False
-                default_sinks.append(ScanDBSink(db_path=scan_db_path))
+                # AsyncSink: keep the SQLite commit/fsync off the USB-drain
+                # thread. AsyncSink mirrors `critical`, so a DB on_scan_start
+                # failure still aborts the scan (and the pre-flight above
+                # already caught the common case with the laser off).
+                default_sinks.append(AsyncSink(ScanDBSink(db_path=scan_db_path)))
             # Telemetry CSV: per-scan snapshots from ConsoleTelemetryPoller.
             # Not a pipeline sink — the poller is its own daemon thread that
             # predates the sink-based architecture, so we register a listener
diff --git a/omotion/StreamInterface.py b/omotion/StreamInterface.py
index 1b2552a6..e3c557c6 100644
--- a/omotion/StreamInterface.py
+++ b/omotion/StreamInterface.py
@@ -119,6 +119,19 @@ def __init__(self, dev, interface_index, desc="Stream"):
         self.expected_size = None
         self.isStreaming = False
         self.packets_received: int = 0  # USB transfers queued since last start_streaming
+        # Host-side backpressure stats since last start_streaming. A put that
+        # blocks because the parser/runner stalled is the host-side cause of
+        # the sensor firmware's "HISTO enqueue fail" drops; surfacing it per
+        # scan makes that previously-invisible stall measurable.
+        self.backpressure_stalls: int = 0      # puts that blocked >= warn threshold
+        self.backpressure_stall_s: float = 0.0  # total seconds blocked on those puts
+        self.dropped_chunks: int = 0            # chunks dropped (queue full past timeout)
+        self._last_backpressure_warn_s: float = 0.0
+        # A put blocking this long is a sub-second stall worth counting/warning.
+        self._backpressure_warn_s: float = 0.1
+        # Drop a chunk only after the queue has been full this long (the loop's
+        # original >1 s drop policy). Kept as an attribute so tests can shorten it.
+        self._put_timeout_s: float = 1.0
 
     def start_streaming(self, queue_obj, expected_size):
         # Recover from a stale thread left over by a previous scan whose
@@ -143,6 +156,10 @@ def start_streaming(self, queue_obj, expected_size):
         self.data_queue = queue_obj
         self.expected_size = expected_size
         self.packets_received = 0
+        self.backpressure_stalls = 0
+        self.backpressure_stall_s = 0.0
+        self.dropped_chunks = 0
+        self._last_backpressure_warn_s = 0.0
         self.stop_event.clear()
         self.thread = threading.Thread(target=self._stream_loop, daemon=True)
         self.thread.start()
@@ -168,10 +185,19 @@ def stop_streaming(self):
         self.isStreaming = False
         self.data_queue = None
         self.expected_size = None
-        logger.info(
+        summary = (
             f"{self.desc}: Streaming stopped — "
             f"{self.packets_received} USB read chunk(s) received"
         )
+        if self.backpressure_stalls or self.dropped_chunks:
+            summary += (
+                f"; host backpressure: {self.backpressure_stalls} stall(s) "
+                f">{int(self._backpressure_warn_s * 1000)}ms "
+                f"totalling {self.backpressure_stall_s:.2f}s, "
+                f"{self.dropped_chunks} chunk(s) dropped"
+            )
+        log = logger.warning if self.dropped_chunks else logger.info
+        log(summary)
 
     def flush_stale_data(
         self,
@@ -359,6 +385,51 @@ def _process_packet(self, raw, pkt_count, cmp_count, cmp_errors):
             self.data_queue.put(raw)
         return cmp_count, cmp_errors
 
+    def _enqueue_chunk(self, data_queue, data) -> bool:
+        """Put one USB chunk onto the packet queue, measuring host-side
+        backpressure.
+
+        A put that returns immediately costs nothing; one that blocks means
+        the parser/runner has stalled and is holding the queue full — the
+        host-side cause of the firmware's histo-queue overflow. We time the
+        block: any wait >= ``_backpressure_warn_s`` is counted (and rate-limit
+        warned) so sub-second stalls are visible, not just the >1 s drops.
+
+        Returns True if the chunk was enqueued, False if it was dropped
+        because the queue stayed full past ``_put_timeout_s``.
+        """
+        t0 = time.monotonic()
+        try:
+            data_queue.put(data, timeout=self._put_timeout_s)
+        except queue.Full:
+            self.dropped_chunks += 1
+            # Real host-side data loss — warn (rate-limited) so it's visible
+            # mid-scan, not only in the stop summary.
+            if t0 - self._last_backpressure_warn_s >= 5.0:
+                self._last_backpressure_warn_s = t0
+                logger.warning(
+                    "%s: packet queue full for >%.0fs (parser falling behind) "
+                    "— dropping %d-byte chunk; %d dropped so far this scan",
+                    self.desc, self._put_timeout_s, len(data), self.dropped_chunks,
+                )
+            return False
+        blocked = time.monotonic() - t0
+        self.packets_received += 1
+        if blocked >= self._backpressure_warn_s:
+            self.backpressure_stalls += 1
+            self.backpressure_stall_s += blocked
+            # Rate-limit to at most once per 5 s so a sustained stall doesn't
+            # flood the log; every stall is still counted above.
+            if t0 - self._last_backpressure_warn_s >= 5.0:
+                self._last_backpressure_warn_s = t0
+                logger.warning(
+                    "%s: USB read stalled %.0f ms on a full packet queue "
+                    "(parser/runner falling behind — risks firmware histo "
+                    "drops); %d stall(s) so far this scan",
+                    self.desc, blocked * 1000.0, self.backpressure_stalls,
+                )
+        return True
+
     def _stream_loop(self):
         # Read timeout must exceed the worst-case USB transfer latency for the
         # final frame.  Normal cadence is ~25 ms; the last frame of a scan can
@@ -392,22 +463,15 @@ def _stream_loop(self):
                     timeout=_READ_TIMEOUT_MS,
                 )
                 if data and data_queue is self.data_queue:
-                    # Use a bounded put so the loop can never block forever
-                    # on a stopped/slow parser. With self.stop_event set the
+                    # Bounded put (see _enqueue_chunk) so the loop can never
+                    # block forever on a stopped/slow parser, and so host-side
+                    # backpressure is measured. With self.stop_event set the
                     # parser also drains until empty (see parse_histogram_stream),
-                    # so this drop window is only ever 1s of backlog at scan
-                    # teardown — small price for a guaranteed loop exit.
-                    try:
-                        data_queue.put(bytes(data), timeout=1.0)
-                        self.packets_received += 1
-                    except queue.Full:
-                        if self.stop_event.is_set():
-                            break
-                        logger.warning(
-                            "%s: data_queue full for >1s during streaming "
-                            "(parser falling behind?); dropping %d-byte chunk",
-                            self.desc, len(data),
-                        )
+                    # so a drop here is only ever <=_put_timeout_s of backlog at
+                    # scan teardown — small price for a guaranteed loop exit.
+                    if not self._enqueue_chunk(data_queue, bytes(data)) \
+                            and self.stop_event.is_set():
+                        break
             except usb.core.USBError as e:
                 if e.errno in (110, 10060):
                     # Timeout — no data arrived within the read window.
diff --git a/omotion/pipeline/__init__.py b/omotion/pipeline/__init__.py
index 1a94ea55..56ac563f 100644
--- a/omotion/pipeline/__init__.py
+++ b/omotion/pipeline/__init__.py
@@ -36,6 +36,7 @@
 from .factory import default_pipeline
 from .pedestal import SensorPedestals
 from .sinks import Sink, ScanMetadata, CsvSink, ScanDBSink, DiagnosticsLogSink
+from .async_sink import AsyncSink
 from .sources import Source, LiveUsbSource, CsvReplaySource
 from .telemetry import (
     TelemetryAggregator, TelemetryFeeder, TelemetryIngestStage, TelemetrySample,
@@ -51,7 +52,7 @@
     "default_pipeline",
     "SensorPedestals",
     "Sink", "ScanMetadata",
-    "CsvSink", "ScanDBSink", "DiagnosticsLogSink",
+    "CsvSink", "ScanDBSink", "DiagnosticsLogSink", "AsyncSink",
     "Source", "LiveUsbSource", "CsvReplaySource",
     "TelemetryAggregator", "TelemetryFeeder", "TelemetryIngestStage",
     "TelemetrySample",
diff --git a/omotion/pipeline/async_sink.py b/omotion/pipeline/async_sink.py
new file mode 100644
index 00000000..f0f35c4c
--- /dev/null
+++ b/omotion/pipeline/async_sink.py
@@ -0,0 +1,119 @@
+"""AsyncSink — run a slow sink's consume() on its own worker thread.
+
+The ScanRunner consume loop runs on the same thread that drains the live
+USB batch queue (see ``LiveUsbSource`` / ``StreamInterface``). A sink that
+blocks there — an fsync on a SQLite commit, a CSV buffer flushing to disk —
+stalls that drain, which backpressures the histogram parser and then
+``dev.read``, overflowing the sensor firmware's shallow 4-deep histo queue
+and dropping frames. (Root-caused on the 2026-06 long-scan soak: the
+firmware "HISTO enqueue fail: queue full" drops were brief host-side drain
+stalls, not a USB fault.)
+
+``AsyncSink`` wraps a sink and moves ``consume()`` onto a dedicated worker
+thread fed by a bounded FIFO queue, so a *transient* sink stall is absorbed
+by the buffer instead of the USB path. Sustained overload (a sink that can
+never keep up) still backpressures once the buffer fills — that is correct;
+you cannot outrun the disk forever and dropping persisted data is worse.
+
+Lifecycle:
+  * ``on_scan_start`` runs the wrapped sink synchronously and propagates any
+    exception, so a ``critical`` sink whose start fails still aborts the scan
+    (ScanRunner reads ``.critical``) and the sink is fully initialized before
+    any ``consume`` is enqueued.
+  * ``consume`` enqueues ``(channel, payload)`` (cheap) for the worker.
+  * ``on_complete`` enqueues a sentinel, waits for the worker to drain every
+    buffered item, then calls the wrapped sink's ``on_complete`` — so no
+    buffered data is lost.
+"""
+
+from __future__ import annotations
+
+import logging
+import queue
+import threading
+from typing import Any
+
+logger = logging.getLogger("openmotion.sdk.pipeline.async_sink")
+
+_SENTINEL = object()
+
+
+class AsyncSink:
+    """Threaded buffer in front of a Sink. Duck-types the Sink protocol."""
+
+    def __init__(self, sink: Any, *, max_queue: int = 64,
+                 join_timeout_s: float = 60.0) -> None:
+        self._sink = sink
+        self._queue: queue.Queue = queue.Queue(maxsize=max(1, int(max_queue)))
+        self._join_timeout_s = float(join_timeout_s)
+        self._worker: threading.Thread | None = None
+        self._started = False
+        self._closed = False
+        self._name = type(sink).__name__
+        # Mirror the attributes the runner reads off a sink so AsyncSink is a
+        # drop-in stand-in (channels for routing, critical for abort policy).
+        self.channels = getattr(sink, "channels", set())
+        self.critical = getattr(sink, "critical", False)
+
+    @property
+    def wrapped(self):
+        """The underlying sink, for callers that need to introspect past the
+        async wrapper (e.g. ``isinstance`` checks on the concrete sink type)."""
+        return self._sink
+
+    def on_scan_start(self, meta) -> None:
+        # Synchronous on purpose: a critical sink that fails here must still
+        # raise so ScanRunner aborts before the laser fires, and the sink must
+        # be ready before the first enqueued consume runs on the worker.
+        self._sink.on_scan_start(meta)
+        self._closed = False
+        self._worker = threading.Thread(
+            target=self._run, name=f"AsyncSink-{self._name}", daemon=True,
+        )
+        self._started = True
+        self._worker.start()
+
+    def consume(self, channel: str, payload: Any) -> None:
+        if not self._started or self._closed:
+            # No worker to service the item (start failed/skipped or already
+            # finalized). Dropping is correct: there is nowhere to deliver it.
+            return
+        # Blocking put: absorbs transient stalls up to max_queue, and applies
+        # honest backpressure only on sustained overload.
+        self._queue.put((channel, payload))
+
+    def on_complete(self) -> None:
+        if not self._started:
+            # on_scan_start never ran (or raised) — no worker was created;
+            # finalize the wrapped sink directly so it still tears down.
+            self._sink.on_complete()
+            return
+        if self._closed:
+            return
+        self._closed = True
+        self._queue.put(_SENTINEL)
+        self._worker.join(timeout=self._join_timeout_s)
+        if self._worker.is_alive():
+            logger.error(
+                "AsyncSink(%s): worker did not drain within %.0fs; "
+                "%d buffered item(s) may be lost",
+                self._name, self._join_timeout_s, self._queue.qsize(),
+            )
+        self._sink.on_complete()
+
+    def _run(self) -> None:
+        while True:
+            item = self._queue.get()
+            try:
+                if item is _SENTINEL:
+                    return
+                channel, payload = item
+                try:
+                    self._sink.consume(channel, payload)
+                except Exception:
+                    logger.exception(
+                        "AsyncSink(%s): consume raised on channel %s; "
+                        "continuing", self._name, channel,
+                    )
+            finally:
+                self._queue.task_done()
diff --git a/omotion/pipeline/sources.py b/omotion/pipeline/sources.py
index dcd48c52..caeab73e 100644
--- a/omotion/pipeline/sources.py
+++ b/omotion/pipeline/sources.py
@@ -11,6 +11,7 @@
 
 import csv
 import logging
+import math
 import queue
 import threading
 import time
@@ -20,12 +21,20 @@
 
 import numpy as np
 
+from omotion.config import CAPTURE_HZ
 from .batch import FrameBatch
 from .sinks import ScanMetadata
 
 
 logger = logging.getLogger("openmotion.sdk.pipeline.sources")
 
+# How long a downstream stall (a slow sink, a GC pause, a pipeline CPU spike)
+# the host should absorb before the USB reader is starved and the sensor
+# firmware's 4-deep (~100 ms) histo queue overflows. The batch queue default
+# is sized to hold this many seconds of frames so the stall is buffered here
+# instead of backpressuring dev.read. See the 2026-06 long-scan soak.
+_TARGET_BUFFER_SECONDS = 3.0
+
 
 @runtime_checkable
 class Source(Protocol):
@@ -195,6 +204,7 @@ def __init__(self, *,
                  batch_size_frames: int = 10,
                  flush_interval_s: float = 0.25,
                  packet_queue_size: int = 64,
+                 batch_queue_size: Optional[int] = None,
                  metadata: ScanMetadata):
         super().__init__(metadata=metadata)
         self._console = console
@@ -205,7 +215,20 @@ def __init__(self, *,
             side: queue.Queue(maxsize=packet_queue_size)
             for side, sensor in self._sensors.items() if sensor is not None
         }
-        self._batch_queue: queue.Queue = queue.Queue(maxsize=4)
+        # Parsed-batch buffer between the per-side reader threads and the
+        # runner. Was a hardcoded 4 (~1 s); a transient runner stall (sink
+        # fsync, GC) filled it and backpressured the parser -> dev.read ->
+        # the firmware's 4-deep histo queue overflowed and dropped frames.
+        # Default it to ~_TARGET_BUFFER_SECONDS of frames, derived from the
+        # 40 Hz capture rate and the batch size (so the intent survives a
+        # different batch_size_frames), and keep it tunable. The runner stalls
+        # before the parser does, so this batch buffer is the slack that keeps
+        # dev.read flowing; the 64-deep packet queue adds ~1.6 s more behind
+        # it. See the 2026-06 long-scan soak root-cause + AsyncSink.
+        if batch_queue_size is None:
+            frames = _TARGET_BUFFER_SECONDS * CAPTURE_HZ
+            batch_queue_size = math.ceil(frames / max(1, self._batch_size))
+        self._batch_queue: queue.Queue = queue.Queue(maxsize=max(1, int(batch_queue_size)))
         self._stop = threading.Event()
         self._reader_threads: list[threading.Thread] = []
 
diff --git a/tests/test_pipeline/test_async_sink.py b/tests/test_pipeline/test_async_sink.py
new file mode 100644
index 00000000..cdf21146
--- /dev/null
+++ b/tests/test_pipeline/test_async_sink.py
@@ -0,0 +1,210 @@
+"""Tests for AsyncSink — the threaded wrapper that decouples a slow disk/DB
+sink from the ScanRunner consume loop (and thus the live USB drain).
+
+Regression context: in the 2026-06 long-scan soak the runner thread that
+drains the live USB batch queue also ran every sink.consume(); a SQLite
+commit / CSV flush stalled that drain, backpressured the parser and
+dev.read, and overflowed the sensor firmware's 4-deep histo queue. AsyncSink
+moves consume() onto a worker thread so a transient sink stall is absorbed by
+a buffer instead of the USB path.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+
+import pytest
+
+from omotion.pipeline.async_sink import AsyncSink
+
+
+class _RecordingSink:
+    """Minimal Sink that records the order of lifecycle calls."""
+
+    channels = {"final", "raw"}
+
+    def __init__(self, *, consume_delay: float = 0.0) -> None:
+        self._consume_delay = consume_delay
+        self.events: list = []
+        self._lock = threading.Lock()
+
+    def on_scan_start(self, meta) -> None:
+        with self._lock:
+            self.events.append(("start", meta))
+
+    def consume(self, channel, payload) -> None:
+        if self._consume_delay:
+            time.sleep(self._consume_delay)
+        with self._lock:
+            self.events.append(("consume", channel, payload))
+
+    def on_complete(self) -> None:
+        with self._lock:
+            self.events.append(("complete",))
+
+
+def test_consume_is_processed_in_fifo_order():
+    inner = _RecordingSink()
+    sink = AsyncSink(inner)
+
+    sink.on_scan_start(meta="M")
+    for i in range(5):
+        sink.consume("final", i)
+    sink.on_complete()
+
+    assert inner.events == [
+        ("start", "M"),
+        ("consume", "final", 0),
+        ("consume", "final", 1),
+        ("consume", "final", 2),
+        ("consume", "final", 3),
+        ("consume", "final", 4),
+        ("complete",),
+    ]
+
+
+def test_on_complete_drains_all_buffered_consumes_before_finalizing():
+    # A slow sink: if on_complete did not wait for the worker to drain, the
+    # ("complete",) marker would land before some ("consume", ...) entries.
+    inner = _RecordingSink(consume_delay=0.01)
+    sink = AsyncSink(inner)
+
+    sink.on_scan_start(meta=None)
+    for i in range(10):
+        sink.consume("final", i)
+    sink.on_complete()
+
+    consumes = [e for e in inner.events if e[0] == "consume"]
+    assert len(consumes) == 10
+    assert inner.events[-1] == ("complete",)
+
+
+def test_on_scan_start_propagates_exception_so_critical_abort_still_works():
+    class _BadStart:
+        channels = {"final"}
+        critical = True
+
+        def on_scan_start(self, meta):
+            raise RuntimeError("db open failed")
+
+        def consume(self, channel, payload):
+            pass
+
+        def on_complete(self):
+            pass
+
+    sink = AsyncSink(_BadStart())
+    with pytest.raises(RuntimeError, match="db open failed"):
+        sink.on_scan_start(meta=None)
+
+
+def test_worker_isolates_a_consume_exception_and_keeps_processing():
+    class _FlakySink:
+        channels = {"final"}
+
+        def __init__(self):
+            self.seen = []
+
+        def on_scan_start(self, meta):
+            pass
+
+        def consume(self, channel, payload):
+            if payload == "boom":
+                raise ValueError("transient sink error")
+            self.seen.append(payload)
+
+        def on_complete(self):
+            pass
+
+    inner = _FlakySink()
+    sink = AsyncSink(inner)
+    sink.on_scan_start(meta=None)
+    sink.consume("final", "a")
+    sink.consume("final", "boom")  # raises inside the worker
+    sink.consume("final", "b")     # must still be processed
+    sink.on_complete()
+
+    assert inner.seen == ["a", "b"]
+
+
+def test_exposes_channels_and_critical_from_wrapped_sink():
+    class _CritSink:
+        channels = {"final", "diagnostics"}
+        critical = True
+
+        def on_scan_start(self, meta):
+            pass
+
+        def consume(self, channel, payload):
+            pass
+
+        def on_complete(self):
+            pass
+
+    sink = AsyncSink(_CritSink())
+    assert sink.channels == {"final", "diagnostics"}
+    assert sink.critical is True
+
+    # Default critical is False when the wrapped sink omits it.
+    assert AsyncSink(_RecordingSink()).critical is False
+
+
+def test_wrapped_exposes_the_inner_sink_for_introspection():
+    inner = _RecordingSink()
+    assert AsyncSink(inner).wrapped is inner
+
+
+def test_async_sink_satisfies_the_sink_protocol():
+    from omotion.pipeline import Sink
+    assert isinstance(AsyncSink(_RecordingSink()), Sink)
+
+
+def test_async_sink_is_transparent_through_scan_runner():
+    """Wrapped in AsyncSink and driven by the real ScanRunner, the inner sink
+    still sees on_scan_start, the dispatched event, and on_complete (last)."""
+    import numpy as np
+    from omotion.pipeline import Pipeline, ScanMetadata, ScanRunner
+    from omotion.pipeline.batch import FrameBatch, LiveEmit
+
+    inner = _RecordingSink()  # subscribes {"final", "raw"}
+
+    class _EmitFinalStage:
+        name = "emit_final"
+
+        def process(self, batch):
+            batch.events.append(LiveEmit(channel="final", payload="P"))
+            return batch
+
+        def reset(self):
+            pass
+
+    meta = ScanMetadata(
+        scan_id="x", subject_id="y", operator="z",
+        started_at_iso="2026-06-22T00:00:00Z", duration_sec=1,
+        left_camera_mask=0, right_camera_mask=0, reduced_mode=False,
+    )
+    batch = FrameBatch(
+        cam_ids=np.zeros(1, dtype=np.int8),
+        frame_ids=np.zeros(1, dtype=np.uint8),
+        raw_histograms=np.zeros((1, 2, 8, 1024), dtype=np.uint32),
+        temperature_c=np.zeros((1, 2, 8), dtype=np.float32),
+        timestamp_s=np.zeros(1, dtype=np.float64),
+        pdc=None, tcm=None, tcl=None,
+    )
+
+    class _Src:
+        metadata = meta
+
+        def __iter__(self):
+            yield batch
+
+        def close(self):
+            pass
+
+    ScanRunner(source=_Src(), pipeline=Pipeline([_EmitFinalStage()]),
+               sinks=[AsyncSink(inner)]).run()
+
+    assert ("start", meta) in inner.events
+    assert ("consume", "final", "P") in inner.events
+    assert inner.events[-1] == ("complete",)
diff --git a/tests/test_pipeline/test_public_api.py b/tests/test_pipeline/test_public_api.py
index 617faa14..bd109531 100644
--- a/tests/test_pipeline/test_public_api.py
+++ b/tests/test_pipeline/test_public_api.py
@@ -13,7 +13,7 @@ def test_public_api_symbols_importable() -> None:
         ScanRunner, CriticalSinkError,
         Source, LiveUsbSource, CsvReplaySource,
         Sink, ScanMetadata,
-        CsvSink, ScanDBSink, DiagnosticsLogSink,
+        CsvSink, ScanDBSink, DiagnosticsLogSink, AsyncSink,
         TelemetryAggregator, TelemetryFeeder, TelemetryIngestStage,
         TelemetrySample,
         Tee, default_pipeline,
@@ -25,7 +25,7 @@ def test_public_api_symbols_importable() -> None:
         PipelineError, TimestampMisalignmentWindow,
         ScanRunner, CriticalSinkError,
         Source, LiveUsbSource, CsvReplaySource,
-        Sink, ScanMetadata, CsvSink, ScanDBSink, DiagnosticsLogSink,
+        Sink, ScanMetadata, CsvSink, ScanDBSink, DiagnosticsLogSink, AsyncSink,
         TelemetryAggregator, TelemetryFeeder, TelemetryIngestStage,
         TelemetrySample,
         Tee, default_pipeline, SensorPedestals,
@@ -46,7 +46,7 @@ def test_public_api_all_list_complete() -> None:
         "default_pipeline",
         "SensorPedestals",
         "Sink", "ScanMetadata",
-        "CsvSink", "ScanDBSink", "DiagnosticsLogSink",
+        "CsvSink", "ScanDBSink", "DiagnosticsLogSink", "AsyncSink",
         "Source", "LiveUsbSource", "CsvReplaySource",
         "TelemetryAggregator", "TelemetryFeeder", "TelemetryIngestStage",
         "TelemetrySample",
diff --git a/tests/test_pipeline/test_sources.py b/tests/test_pipeline/test_sources.py
index ef32c3ad..5076b161 100644
--- a/tests/test_pipeline/test_sources.py
+++ b/tests/test_pipeline/test_sources.py
@@ -134,6 +134,33 @@ def test_live_usb_source_close_stops_event():
     assert src._stop.is_set()
 
 
+def test_batch_queue_depth_is_configurable():
+    """The parsed-batch buffer between the reader threads and the runner is
+    sized via batch_queue_size so a slow runner can't backpressure dev.read
+    too quickly (see the 2026-06 histo-drop soak)."""
+    src = LiveUsbSource(
+        console=None, left=None, right=None,
+        metadata=_meta(), batch_queue_size=32,
+    )
+    assert src._batch_queue.maxsize == 32
+
+
+def test_batch_queue_default_targets_three_seconds_regardless_of_batch_size():
+    """Default buffer is sized to absorb a ~3 s downstream stall (so a
+    transient runner stall doesn't backpressure dev.read into the firmware
+    histo-queue overflow), derived from the capture rate + batch size rather
+    than a magic constant — so the intent holds for a different batch size."""
+    from omotion.config import CAPTURE_HZ
+    for batch_frames in (10, 20):
+        src = LiveUsbSource(
+            console=None, left=None, right=None,
+            metadata=_meta(), batch_size_frames=batch_frames,
+        )
+        seconds = src._batch_queue.maxsize * batch_frames / CAPTURE_HZ
+        # ~3 s, rounded up by at most one batch — not over-provisioned.
+        assert 3.0 <= seconds < 3.0 + batch_frames / CAPTURE_HZ
+
+
 def test_live_usb_source_reader_loop_builds_batches_from_packet_queue(monkeypatch):
     """Mock parse_histogram_stream to feed fake samples; verify _reader_loop
     accumulates them into FrameBatches and pushes to the batch queue."""
diff --git a/tests/test_scan_workflow.py b/tests/test_scan_workflow.py
index bfa4c3ab..6551df82 100644
--- a/tests/test_scan_workflow.py
+++ b/tests/test_scan_workflow.py
@@ -125,7 +125,10 @@ def test_start_scan_auto_injects_csv_sink_when_data_dir_set(tmp_path):
         left_camera_mask=0xFF, right_camera_mask=0, reduced_mode=False,
     )
     motion.scan_workflow.start_scan(request)
-    csv_sinks = [s for s in motion.scan_workflow._runner.sinks if isinstance(s, CsvSink)]
+    # Storage sinks are AsyncSink-wrapped (decoupled from the USB-drain
+    # thread); unwrap to assert on the concrete sink type.
+    csv_sinks = [s for s in motion.scan_workflow._runner.sinks
+                 if isinstance(getattr(s, "wrapped", s), CsvSink)]
     assert len(csv_sinks) == 1
 
 
@@ -137,7 +140,8 @@ def test_start_scan_skips_csv_sink_when_data_dir_none():
         left_camera_mask=0xFF, right_camera_mask=0, reduced_mode=False,
     )
     motion.scan_workflow.start_scan(request)
-    csv_sinks = [s for s in motion.scan_workflow._runner.sinks if isinstance(s, CsvSink)]
+    csv_sinks = [s for s in motion.scan_workflow._runner.sinks
+                 if isinstance(getattr(s, "wrapped", s), CsvSink)]
     assert csv_sinks == []
 
 
@@ -150,7 +154,8 @@ def test_start_scan_skips_default_storage_when_request_opts_out(tmp_path):
         skip_default_storage=True,
     )
     motion.scan_workflow.start_scan(request)
-    assert all(not isinstance(s, CsvSink) for s in motion.scan_workflow._runner.sinks)
+    assert all(not isinstance(getattr(s, "wrapped", s), CsvSink)
+               for s in motion.scan_workflow._runner.sinks)
 
 
 def test_start_scan_does_not_auto_wire_pipeline_telemetry_source():
diff --git a/tests/test_stream_interface_backpressure.py b/tests/test_stream_interface_backpressure.py
new file mode 100644
index 00000000..9ffa9e00
--- /dev/null
+++ b/tests/test_stream_interface_backpressure.py
@@ -0,0 +1,87 @@
+"""StreamInterface host-side backpressure instrumentation.
+
+The USB read loop puts each chunk onto the per-side packet queue. When the
+downstream parser/runner stalls, that put blocks. Previously a put only
+surfaced anything after blocking a full second (the drop path); sub-second
+stalls — which are exactly what overflowed the sensor firmware's 4-deep
+histo queue in the 2026-06 soak — were invisible. These tests cover the
+per-scan counters that make those stalls (and any host-side drops) visible.
+"""
+
+from __future__ import annotations
+
+import queue
+import threading
+import time
+
+from omotion.StreamInterface import StreamInterface
+
+
+def _si() -> StreamInterface:
+    # dev is never touched by the enqueue path; construction is side-effect free.
+    return StreamInterface(dev=None, interface_index=0, desc="Test")
+
+
+def test_enqueue_chunk_no_stall_when_queue_has_space():
+    si = _si()
+    q: queue.Queue = queue.Queue(maxsize=4)
+
+    assert si._enqueue_chunk(q, b"frame") is True
+    assert si.backpressure_stalls == 0
+    assert si.dropped_chunks == 0
+    assert si.packets_received == 1
+    assert q.get_nowait() == b"frame"
+
+
+def test_enqueue_chunk_counts_a_sub_second_stall():
+    si = _si()
+    si._backpressure_warn_s = 0.05
+    q: queue.Queue = queue.Queue(maxsize=1)
+    q.put(b"occupied")  # queue starts full so the next put must block
+
+    # Free a slot ~0.12 s later — a sub-second (but >warn) host stall.
+    def _free_slot():
+        time.sleep(0.12)
+        q.get()
+
+    threading.Thread(target=_free_slot, daemon=True).start()
+
+    t0 = time.monotonic()
+    assert si._enqueue_chunk(q, b"frame") is True
+    elapsed = time.monotonic() - t0
+
+    assert elapsed >= 0.05
+    assert si.backpressure_stalls == 1
+    assert si.backpressure_stall_s >= 0.05
+    assert si.dropped_chunks == 0
+    assert si.packets_received == 1
+
+
+def test_enqueue_chunk_drops_when_queue_full_past_timeout():
+    si = _si()
+    si._put_timeout_s = 0.05  # keep the test fast; production drops only after 1 s
+    q: queue.Queue = queue.Queue(maxsize=1)
+    q.put(b"occupied")  # never freed → put times out → chunk dropped
+
+    assert si._enqueue_chunk(q, b"frame") is False
+    assert si.dropped_chunks == 1
+    assert si.backpressure_stalls == 0
+    assert si.packets_received == 0
+
+
+def test_start_streaming_resets_backpressure_counters(monkeypatch):
+    si = _si()
+    si.backpressure_stalls = 7
+    si.backpressure_stall_s = 3.0
+    si.dropped_chunks = 2
+    si.packets_received = 99
+
+    # Don't run the real USB loop — just verify start_streaming clears stats.
+    monkeypatch.setattr(si, "_stream_loop", lambda: None)
+    si.start_streaming(queue.Queue(), expected_size=10)
+    si.stop_streaming()
+
+    assert si.backpressure_stalls == 0
+    assert si.backpressure_stall_s == 0.0
+    assert si.dropped_chunks == 0
+    assert si.packets_received == 0