feat: decouple storage sinks from the USB drain; surface host backpressure

[boringethan]

Refs #116. Superseded by #113 (same work rebased + review fixes); kept open as fallback until #113 merges, then this closes.

Why

On the 2026-06 long-scan soak the sensor firmware logged ~833 HISTO enqueue fail: queue full events (enq-deq=4, txfail=0). Root cause: the ScanRunner consume loop runs on the same thread that drains the live USB histogram batch queue, so a slow disk/DB sink — a SQLite commit/fsync or a CSV flush — stalls that drain. The stall backpressures the parser → dev.read → the firmware's deliberately shallow 4-deep histo queue (~100 ms at 40 Hz) overflows and frames are dropped. These were brief (>100 ms, <1 s) host-side stalls, not a USB fault and not the persistent histo-wedge.

What

Fix #1 — decouple the storage sinks from the USB drain

• New AsyncSink (omotion/pipeline/async_sink.py) wraps a sink and runs consume() on a dedicated worker thread fed by a bounded FIFO buffer, so a transient sink stall is absorbed by the buffer instead of stalling the USB path. on_scan_start stays synchronous (preserves critical-sink abort + ordering before the first consume); on_complete drains the buffer and joins the worker before finalising the wrapped sink, so no persisted data is lost. Sustained overload still backpressures once the buffer fills (correct — you can't outrun the disk forever).
• ScanWorkflow now wraps CsvSink and ScanDBSink in AsyncSink (the live-plot/diagnostics sinks stay synchronous so the GUI trace isn't delayed).
• LiveUsbSource._batch_queue depth is now configurable and defaulted up from 4 (~1 s) to 16, so other runner hiccups (pipeline CPU spike, GC) are absorbed before reaching dev.read.

Fix #3 — make the previously-invisible sub-second backpressure observable

• StreamInterface now times every packet-queue put. A put that blocks ≥100 ms (previously only the >1 s drop was logged) is counted per scan — backpressure_stalls / backpressure_stall_s / dropped_chunks — rate-limit warned mid-scan, and summarised in the stop_streaming line (logged at WARNING if any host-side drops occurred).

Out of scope: bumping the firmware HISTO_QUEUE_SIZE (4→8) — RAM-bound on the sensor, intentionally left alone. This PR removes the host-side cause instead.

Tests

New, test-first:

• test_async_sink.py — FIFO ordering, drain-before-finalise, critical-abort propagation, worker error isolation, Sink protocol conformance, transparency through a real ScanRunner, wrapped accessor.
• test_stream_interface_backpressure.py — no-stall fast path, sub-second stall counting, drop-past-timeout, counter reset on start_streaming.
• test_sources.py — configurable / deepened _batch_queue.

Full pipeline + scan-workflow suites pass. (The 3 pre-existing ScanDBSink "empty session deleted" failures on next are unrelated to this change and untouched here.)

🤖 Generated with Claude Code

[boringethan]

Closing as superseded by #113, which carries both of this PR's commits rebased onto current next plus fixes for the review findings (batch-queue sizing, AsyncSink finalize race, reuse poisoning, warning suppression). Tracking continues on #116.