diff --git a/src/syncfield/adapters/_generic.py b/src/syncfield/adapters/_generic.py
index dc8c25c..f3315a6 100644
--- a/src/syncfield/adapters/_generic.py
+++ b/src/syncfield/adapters/_generic.py
@@ -1,12 +1,96 @@
-"""Private internals shared by PollingSensorStream and PushSensorStream."""
+"""Private internals shared by PollingSensorStream and PushSensorStream.
+
+SDK contract for GUI consumers (see syncfield-sensor-onboarding-enhancements §5):
+
+3. **Transient transport hiccup auto-reopen.**  Adapters that own a transport
+   (serial port, BLE connection, USB pipe) MUST attempt up to
+   :data:`TRANSIENT_REOPEN_MAX_ATTEMPTS` reopens with exponential backoff
+   (total wall time ≤ :data:`TRANSIENT_REOPEN_MAX_WAIT_S` seconds) before
+   surfacing a stream-level error.  Use :func:`retry_open` in the adapter's
+   open / reconnect path to satisfy this contract automatically.
+"""
 
 from __future__ import annotations
 
+import logging
+import time
 import threading
-from typing import Optional
+from typing import Callable, Optional, TypeVar
 
 from syncfield.types import StreamCapabilities
 
+logger = logging.getLogger(__name__)
+
+#: Maximum number of reopen attempts before giving up (Contract 3).
+TRANSIENT_REOPEN_MAX_ATTEMPTS: int = 5
+
+#: Maximum total backoff wait across all retry attempts in seconds (Contract 3).
+TRANSIENT_REOPEN_MAX_WAIT_S: float = 30.0
+
+_T = TypeVar("_T")
+
+
+def retry_open(
+    open_fn: Callable[[], _T],
+    *,
+    max_attempts: int = TRANSIENT_REOPEN_MAX_ATTEMPTS,
+    max_wait_s: float = TRANSIENT_REOPEN_MAX_WAIT_S,
+    stream_id: str = "<unknown>",
+) -> _T:
+    """Call *open_fn* up to *max_attempts* times with exponential backoff.
+
+    **SDK contract — transient transport reopen (Contract 3):**  Adapters
+    that own a transport (serial, BLE, USB) MUST use this helper (or
+    equivalent retry logic) in their open/reconnect path.  A single
+    transient ``OSError`` or ``SerialException`` MUST NOT surface
+    immediately as a stream-level failure.  The adapter MUST retry at
+    least :data:`TRANSIENT_REOPEN_MAX_ATTEMPTS` times, sleeping 1 s,
+    2 s, 4 s, … (capped so total wait ≤ *max_wait_s*) between attempts.
+
+    Args:
+        open_fn: Zero-argument callable that opens the transport and
+            returns a handle (e.g. a ``serial.Serial`` instance).
+            May raise any exception on transient failure.
+        max_attempts: How many attempts before re-raising.  Defaults to
+            :data:`TRANSIENT_REOPEN_MAX_ATTEMPTS` (5).
+        max_wait_s: Hard cap on total sleep time across all retries.
+            Defaults to :data:`TRANSIENT_REOPEN_MAX_WAIT_S` (30 s).
+        stream_id: Stream identifier for log messages.
+
+    Returns:
+        The value returned by *open_fn* on success.
+
+    Raises:
+        Exception: The last exception raised by *open_fn* once all
+            attempts are exhausted.
+    """
+    delay = 1.0
+    last_exc: Optional[Exception] = None
+    total_waited = 0.0
+    for attempt in range(1, max_attempts + 1):
+        try:
+            return open_fn()
+        except Exception as exc:
+            last_exc = exc
+            if attempt >= max_attempts:
+                break
+            # Compute sleep duration capped by max_wait_s budget.
+            # Even when the budget is exhausted we still retry up to
+            # max_attempts times — max_wait_s limits sleep duration only,
+            # not the number of attempts.
+            remaining_budget = max_wait_s - total_waited
+            wait = min(delay, max(remaining_budget, 0.0))
+            logger.warning(
+                "[%s] transient open error (attempt %d/%d): %s — retrying in %.1fs",
+                stream_id, attempt, max_attempts, exc, wait,
+            )
+            if wait > 0:
+                time.sleep(wait)
+            total_waited += wait
+            delay = delay * 2
+    assert last_exc is not None
+    raise last_exc
+
 
 class _SensorWriteCore:
     """Frame counter and timing tracker for generic sensor helpers.
diff --git a/src/syncfield/adapters/polling_sensor.py b/src/syncfield/adapters/polling_sensor.py
index e0e0451..69b17cf 100644
--- a/src/syncfield/adapters/polling_sensor.py
+++ b/src/syncfield/adapters/polling_sensor.py
@@ -1,4 +1,13 @@
-"""PollingSensorStream — generic helper for sensors with a read() function."""
+"""PollingSensorStream — generic helper for sensors with a read() function.
+
+SDK contract for GUI consumers (see syncfield-sensor-onboarding-enhancements §5):
+
+3. **Transient transport hiccup auto-reopen.**  When an ``open`` callback is
+   provided, ``PollingSensorStream`` MUST attempt up to
+   :data:`~syncfield.adapters._generic.TRANSIENT_REOPEN_MAX_ATTEMPTS` reopens
+   with exponential backoff before surfacing a stream-level error.  This is
+   implemented via :func:`~syncfield.adapters._generic.retry_open`.
+"""
 
 from __future__ import annotations
 
@@ -10,6 +19,7 @@
 from syncfield.adapters._generic import (
     _SensorWriteCore,
     _resolve_capabilities,
+    retry_open,
 )
 from syncfield.clock import SessionClock
 from syncfield.stream import DeviceKey, StreamBase
@@ -24,7 +34,19 @@
 
 
 class PollingSensorStream(StreamBase):
-    """Generic helper that polls a user read() function on a fixed hz."""
+    """Generic helper that polls a user read() function on a fixed hz.
+
+    SDK contract for GUI consumers (see syncfield-sensor-onboarding-enhancements §5):
+
+    3. **Transient transport hiccup auto-reopen.**  When an ``open`` callback
+       is supplied, ``PollingSensorStream.connect()`` MUST attempt up to
+       :data:`~syncfield.adapters._generic.TRANSIENT_REOPEN_MAX_ATTEMPTS` (5)
+       reopens with exponential backoff (≤ 30 s total) before surfacing a
+       stream-level error.  A single ``OSError`` or ``SerialException`` MUST
+       NOT propagate immediately — the adapter retries automatically so that
+       brief USB re-enumerations or connection blips are transparent to GUI
+       users.
+    """
 
     def __init__(
         self,
@@ -173,8 +195,18 @@ def _capture_loop(self) -> None:
     # ------------------------------------------------------------------
 
     def connect(self) -> None:
+        """Open the transport and start the polling loop.
+
+        **SDK contract — transient transport reopen (Contract 3):**
+        When an ``open`` callback was provided, this method MUST retry
+        the open up to :data:`~syncfield.adapters._generic.TRANSIENT_REOPEN_MAX_ATTEMPTS`
+        times with exponential backoff before propagating an exception.
+        """
         if self._open is not None:
-            self._handle = self._open()
+            self._handle = retry_open(
+                self._open,
+                stream_id=self.id,
+            )
         self._stop_event.clear()
         self._thread = threading.Thread(
             target=self._capture_loop,
diff --git a/src/syncfield/adapters/push_sensor.py b/src/syncfield/adapters/push_sensor.py
index 0ef967a..71591a2 100644
--- a/src/syncfield/adapters/push_sensor.py
+++ b/src/syncfield/adapters/push_sensor.py
@@ -1,10 +1,27 @@
-"""PushSensorStream — generic helper for callback/asyncio/external-thread sources."""
+"""PushSensorStream — generic helper for callback/asyncio/external-thread sources.
+
+SDK contracts for GUI consumers (see syncfield-sensor-onboarding-enhancements §5):
+
+1. **on_connect callback is non-blocking.**  ``PushSensorStream`` MUST NOT
+   synchronously wait on the callback the user supplied via ``on_connect=``.
+   The SDK fires it in a background daemon thread; the calling thread (and
+   the orchestrator's connect loop) return immediately.  If the user's
+   callback itself blocks, that blocks ONLY the callback's background thread,
+   NOT the stream lifecycle or the viewer's "Connecting…" indicator.
+
+2. **Burst-aware capture_ns interpolation.**  When the user's producer reads
+   N > 1 samples in a single USB/BLE tick, it MUST NOT pass the same
+   ``capture_ns`` to all N ``push()`` calls.  Instead it SHOULD call
+   :func:`burst_timestamps` to obtain per-sample host-monotonic timestamps
+   spaced by ``dt = 1e9 / expected_hz`` nanoseconds, anchored so the *last*
+   sample lands at the actual read instant.
+"""
 
 from __future__ import annotations
 
 import threading
 import time
-from typing import Callable, Optional
+from typing import Callable, List, Optional
 
 from syncfield.adapters._generic import _SensorWriteCore, _resolve_capabilities
 from syncfield.clock import SessionClock
@@ -15,8 +32,62 @@
 )
 
 
+def burst_timestamps(n: int, *, anchor_ns: Optional[int] = None, expected_hz: float) -> List[int]:
+    """Compute per-sample host-monotonic timestamps for a burst read.
+
+    When a single USB/BLE read returns *n* samples that were collected at a
+    known fixed rate, this helper distributes timestamps so the **last**
+    sample lands at *anchor_ns* (or ``time.monotonic_ns()`` if omitted) and
+    earlier samples step backward by ``1e9 / expected_hz`` nanoseconds.
+
+    The SDK contract for ``PushSensorStream`` requires callers to use this
+    helper (or equivalent arithmetic) rather than passing the same
+    ``capture_ns`` to all ``push()`` calls in a burst.  Clustering N samples
+    at one tick degrades timestamp quality at high rates (1 kHz+) and defeats
+    the sync alignment that depends on per-sample spread.
+
+    Args:
+        n: Number of samples in the burst.  Must be >= 1.
+        anchor_ns: Host monotonic nanosecond timestamp for the *last* sample
+            in the burst.  Defaults to ``time.monotonic_ns()`` at call time.
+        expected_hz: Expected sensor sample rate in Hz.
+
+    Returns:
+        A list of *n* integer nanosecond timestamps in ascending order, with
+        ``timestamps[-1] == anchor_ns`` and adjacent deltas equal to
+        ``round(1e9 / expected_hz)``.
+
+    Example::
+
+        ts = burst_timestamps(5, anchor_ns=recv_ns, expected_hz=1000.0)
+        for i, (sample, capture_ns) in enumerate(zip(burst, ts)):
+            stream.push(sample, capture_ns=capture_ns)
+    """
+    if n < 1:
+        raise ValueError(f"burst_timestamps: n must be >= 1, got {n}")
+    if expected_hz <= 0:
+        raise ValueError(f"burst_timestamps: expected_hz must be > 0, got {expected_hz}")
+    if anchor_ns is None:
+        anchor_ns = time.monotonic_ns()
+    dt_ns = round(1e9 / expected_hz)
+    return [anchor_ns - (n - 1 - i) * dt_ns for i in range(n)]
+
+
 class PushSensorStream(StreamBase):
-    """Generic helper for sensors driven by user-owned producer threads."""
+    """Generic helper for sensors driven by user-owned producer threads.
+
+    SDK contracts for GUI consumers (see syncfield-sensor-onboarding-enhancements §5):
+
+    1. **on_connect callback is non-blocking.**  The SDK fires the callback in
+       a background daemon thread so the orchestrator's connect loop MUST NOT
+       be blocked even if the user's ``on_connect`` coroutine/function takes
+       time to complete.  See :func:`burst_timestamps` for Contract 2.
+
+    2. **Burst-aware capture_ns interpolation.**  Callers who receive N > 1
+       samples per USB/BLE tick MUST distribute timestamps using
+       :func:`burst_timestamps` rather than passing the same ``capture_ns``
+       to every ``push()`` in a burst.
+    """
 
     def __init__(
         self,
@@ -49,9 +120,23 @@ def device_key(self) -> Optional[DeviceKey]:
     # ------------------------------------------------------------------
 
     def connect(self) -> None:
+        """Open the stream for pushing.
+
+        **SDK contract — non-blocking on_connect:**  The user-supplied
+        ``on_connect`` callback MUST NOT block the calling thread.  The
+        SDK fires it in a background daemon thread so the orchestrator's
+        connect loop (and the GUI's "Connecting…" indicator) proceed
+        immediately regardless of how long the callback takes.
+        """
         self._connected = True
         if self._on_connect is not None:
-            self._on_connect(self)
+            t = threading.Thread(
+                target=self._on_connect,
+                args=(self,),
+                name=f"push-sensor-on-connect-{self.id}",
+                daemon=True,
+            )
+            t.start()
 
     def start_recording(self, session_clock: SessionClock) -> None:
         self._begin_recording_window(session_clock)
diff --git a/src/syncfield/orchestrator.py b/src/syncfield/orchestrator.py
index f2d979e..2c25eb7 100644
--- a/src/syncfield/orchestrator.py
+++ b/src/syncfield/orchestrator.py
@@ -288,6 +288,26 @@ class SessionOrchestrator:
             require an audio stream — disabling this flag in a
             multi-host session means you must add your own
             audio-capable stream explicitly.
+
+    SDK contracts for GUI consumers (see syncfield-sensor-onboarding-enhancements §5):
+
+    4. **SyncToneConfig.silent() MUST NOT register a host_audio stream.**
+       When the session's ``sync_tone`` has ``suppress_host_audio=True``
+       (set automatically by :meth:`~syncfield.tone.SyncToneConfig.silent`),
+       :meth:`_maybe_preregister_host_audio` and :meth:`_maybe_inject_host_audio`
+       MUST return immediately without registering or connecting any
+       :class:`~syncfield.adapters.host_audio.HostAudioStream`.
+
+    5. **Stream errors MUST NOT propagate to the SessionOrchestrator.**
+       An unhandled exception inside a stream's capture loop (i.e. inside
+       the adapter's background thread that calls ``push()`` or runs the
+       polling loop) MUST NOT raise through the orchestrator or cause the
+       session to transition out of ``RECORDING``.  Each stream runs in
+       its own daemon thread — errors are isolated per stream.  The session
+       continues collecting data from all remaining healthy streams.
+       Errors raised during ``start_recording`` fan-out are handled by the
+       orchestrator's rollback logic but do NOT affect other streams
+       mid-session.
     """
 
     def __init__(
@@ -2753,9 +2773,21 @@ def _maybe_preregister_host_audio(self) -> None:
         registered. Only registers the stream (no device open) so the
         viewer can display the audio card immediately. The actual device
         connection happens in :meth:`connect` along with all other streams.
+
+        **SDK contract (Contract 4):** When the session's
+        :class:`~syncfield.tone.SyncToneConfig` has
+        ``suppress_host_audio=True`` (set automatically by
+        :meth:`~syncfield.tone.SyncToneConfig.silent`), this method MUST
+        skip host-audio registration so that a silent-mode session never
+        shows a ghost ``host_audio`` stream in the GUI.
         """
         if not self._enable_host_audio:
             return
+        # Contract 4: SyncToneConfig.silent() sets suppress_host_audio=True
+        # to signal that no acoustic sync path is desired. Honour it here so
+        # a ghost host_audio stream never appears in the GUI viewer.
+        if self._sync_tone.suppress_host_audio:
+            return
 
         try:
             from syncfield.adapters.host_audio import (
@@ -2784,9 +2816,18 @@ def _maybe_inject_host_audio(self) -> None:
         this is a no-op (connect loop handles it). If not yet added
         (e.g. user skipped add() and went straight to connect()), this
         adds and connects it now.
+
+        **SDK contract (Contract 4):** Mirrors the ``suppress_host_audio``
+        check in :meth:`_maybe_preregister_host_audio` — both guards MUST
+        be present to prevent late injection during the connect phase.
         """
         if not self._enable_host_audio:
             return
+        # Contract 4: honour SyncToneConfig.silent()'s suppress_host_audio
+        # flag at connect() time as well, in case the caller bypassed add()
+        # and went straight to connect() (e.g. legacy one-shot path).
+        if self._sync_tone.suppress_host_audio:
+            return
 
         has_audio = any(
             s.capabilities.provides_audio_track
diff --git a/src/syncfield/tone.py b/src/syncfield/tone.py
index e0325a9..9d09c98 100644
--- a/src/syncfield/tone.py
+++ b/src/syncfield/tone.py
@@ -216,6 +216,22 @@ class SyncToneConfig:
         pre_stop_tail_margin_ms: Extra wait time (on top of the stop
             chirp's own duration) before stopping streams so the chirp
             tail is fully captured in any recording audio track.
+        suppress_host_audio: When ``True``, the
+            :class:`~syncfield.orchestrator.SessionOrchestrator` MUST NOT
+            auto-register a ``host_audio`` stream, even when a microphone
+            is detected.  Automatically set by :meth:`silent` so that
+            "silent mode" truly produces no audio-related streams.
+
+    SDK contract for GUI consumers (see syncfield-sensor-onboarding-enhancements §5):
+
+    4. **SyncToneConfig.silent() MUST NOT register a host_audio stream.**
+       When :meth:`silent` is used, the orchestrator MUST skip auto-injection
+       of :class:`~syncfield.adapters.host_audio.HostAudioStream`.  This is
+       enforced via ``suppress_host_audio=True`` on the config object, which
+       the orchestrator reads in both its pre-register and inject helpers.
+       Callers who construct ``SyncToneConfig(enabled=False)`` directly retain
+       the old behaviour (host_audio may still be injected) to preserve
+       backward compatibility.
     """
 
     enabled: bool = True
@@ -226,6 +242,8 @@ class SyncToneConfig:
     )
     post_start_stabilization_ms: int = 200
     pre_stop_tail_margin_ms: int = 200
+    #: Set by :meth:`silent` to suppress host_audio auto-injection (Contract 4).
+    suppress_host_audio: bool = False
 
     @classmethod
     def default(cls) -> "SyncToneConfig":
@@ -260,7 +278,16 @@ def audible(cls) -> "SyncToneConfig":
 
     @classmethod
     def silent(cls) -> "SyncToneConfig":
-        """Construct with the start/stop chirps disabled.
+        """Construct with the start/stop chirps disabled and host_audio suppressed.
+
+        **SDK contract (Contract 4) — silent() MUST NOT register host_audio.**
+        This factory sets ``suppress_host_audio=True`` so the
+        :class:`~syncfield.orchestrator.SessionOrchestrator` MUST skip
+        auto-injection of
+        :class:`~syncfield.adapters.host_audio.HostAudioStream`.  In a
+        truly silent session there is no acoustic sync path, so capturing
+        a microphone track would produce a ghost ``host_audio`` stream
+        that serves no purpose and confuses GUI users.
 
         The 3/2/1 countdown tick still plays — disabling the
         ultrasonic / audible chirps does not mean "no audio at all":
@@ -272,7 +299,7 @@ def silent(cls) -> "SyncToneConfig":
         itself is unacceptable (clinical, audio-sensitive subjects)
         but the operator still benefits from a 3/2/1 audible cue.
         """
-        return cls(enabled=False)
+        return cls(enabled=False, suppress_host_audio=True)
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/unit/test_sdk_contracts.py b/tests/unit/test_sdk_contracts.py
new file mode 100644
index 0000000..47635e4
--- /dev/null
+++ b/tests/unit/test_sdk_contracts.py
@@ -0,0 +1,585 @@
+"""SDK contract tests for GUI no-code onboarding (Phase 2).
+
+Five contracts documented in syncfield-sensor-onboarding-enhancements §5:
+
+1. on_connect callback is non-blocking.
+2. Burst-aware capture_ns interpolation via burst_timestamps().
+3. Transient transport hiccup auto-reopen with backoff.
+4. SyncToneConfig.silent() MUST NOT register a host_audio stream.
+5. A stream-level error MUST NOT propagate to SessionOrchestrator.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from typing import Any, List, Optional
+from unittest.mock import patch
+
+import pytest
+
+from syncfield.adapters._generic import (
+    TRANSIENT_REOPEN_MAX_ATTEMPTS,
+    retry_open,
+)
+from syncfield.adapters.polling_sensor import PollingSensorStream
+from syncfield.adapters.push_sensor import PushSensorStream, burst_timestamps
+from syncfield.orchestrator import SessionOrchestrator
+from syncfield.testing import FakeStream
+from syncfield.tone import SyncToneConfig
+from syncfield.types import SessionState
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _session(tmp_path, **kwargs) -> SessionOrchestrator:
+    """Build a test session with auto-countdown stripped."""
+    session = SessionOrchestrator(
+        host_id=kwargs.pop("host_id", "test_host"),
+        output_dir=tmp_path,
+        sync_tone=kwargs.pop("sync_tone", SyncToneConfig.silent()),
+        **kwargs,
+    )
+    real_start = session.start
+
+    def _fast_start(*args, **start_kwargs):
+        start_kwargs.setdefault("countdown_s", 0)
+        return real_start(*args, **start_kwargs)
+
+    session.start = _fast_start  # type: ignore[method-assign]
+    return session
+
+
+# ---------------------------------------------------------------------------
+# Contract 1 — on_connect callback is non-blocking
+# ---------------------------------------------------------------------------
+
+
+class TestContract1OnConnectNonBlocking:
+    """on_connect MUST NOT block the stream's connect() call.
+
+    If a user's on_connect sleeps for 2 s, the stream.connect() call itself
+    must return quickly (< 0.5 s wall time), and the callback runs in the
+    background daemon thread.
+    """
+
+    def test_slow_on_connect_does_not_block_connect_call(self):
+        """connect() returns in < 0.5 s even when on_connect sleeps for 2 s."""
+        callback_started = threading.Event()
+        callback_done = threading.Event()
+
+        def slow_on_connect(stream: PushSensorStream) -> None:
+            callback_started.set()
+            time.sleep(2.0)
+            callback_done.set()
+
+        stream = PushSensorStream("test_sensor", on_connect=slow_on_connect)
+
+        t0 = time.monotonic()
+        stream.connect()
+        elapsed = time.monotonic() - t0
+
+        # connect() must return well before the callback finishes
+        assert elapsed < 0.5, (
+            f"connect() blocked for {elapsed:.2f}s; "
+            "on_connect callback must run in background thread"
+        )
+        # The background thread should have started the callback
+        assert callback_started.wait(timeout=0.5), (
+            "on_connect callback never started"
+        )
+        # Cleanup — don't leave the background thread running forever
+        callback_done.wait(timeout=3.0)
+
+    def test_connect_still_sets_connected_flag(self):
+        """Stream is immediately ready for push() after connect() returns."""
+        connected_states: List[bool] = []
+
+        def on_connect(stream: PushSensorStream) -> None:
+            time.sleep(0.1)
+
+        stream = PushSensorStream("test_sensor", on_connect=on_connect)
+        stream.connect()
+        # _connected must be True immediately after connect() returns
+        assert stream._connected is True
+
+    def test_no_on_connect_callback_still_works(self):
+        """Streams without on_connect connect instantly and are usable."""
+        stream = PushSensorStream("no_cb_sensor")
+        stream.connect()
+        assert stream._connected is True
+
+    def test_on_connect_exception_does_not_propagate_to_connect(self):
+        """Even if on_connect raises, connect() must not raise.
+
+        The daemon thread that runs on_connect may raise, but that exception
+        must stay confined to that thread — it must not propagate to connect()
+        or to any other thread.  We wrap the thread's excepthook for this test
+        to absorb the expected RuntimeError without triggering a pytest warning.
+        """
+        done = threading.Event()
+        exceptions: List[Exception] = []
+        original_excepthook = threading.excepthook
+
+        def absorbing_excepthook(args):
+            if (
+                args.exc_type is RuntimeError
+                and str(args.exc_value) == "bad callback"
+            ):
+                exceptions.append(args.exc_value)
+            else:
+                original_excepthook(args)
+
+        threading.excepthook = absorbing_excepthook
+        try:
+            def bad_on_connect(stream: PushSensorStream) -> None:
+                raise RuntimeError("bad callback")
+
+            stream = PushSensorStream("err_sensor", on_connect=bad_on_connect)
+            stream.connect()
+            done.wait(timeout=0.5)  # wait for thread to start
+            # Give it a moment to raise and be caught
+            time.sleep(0.1)
+        finally:
+            threading.excepthook = original_excepthook
+
+        assert stream._connected is True
+        # The exception was absorbed (isolated to the thread)
+        assert len(exceptions) == 1
+
+
+# ---------------------------------------------------------------------------
+# Contract 2 — Burst-aware capture_ns interpolation
+# ---------------------------------------------------------------------------
+
+
+class TestContract2BurstTimestamps:
+    """burst_timestamps() must distribute N timestamps uniformly.
+
+    The last timestamp must equal anchor_ns; adjacent deltas must equal
+    round(1e9 / expected_hz).
+    """
+
+    def test_single_sample_returns_anchor(self):
+        anchor = 1_000_000_000_000
+        ts = burst_timestamps(1, anchor_ns=anchor, expected_hz=1000.0)
+        assert ts == [anchor]
+
+    def test_five_samples_at_1khz_spaced_by_1ms(self):
+        anchor = 1_000_000_000_000
+        ts = burst_timestamps(5, anchor_ns=anchor, expected_hz=1000.0)
+        assert len(ts) == 5
+        assert ts[-1] == anchor
+        dt_ns = round(1e9 / 1000.0)  # 1_000_000 ns = 1 ms
+        for i in range(4):
+            delta = ts[i + 1] - ts[i]
+            assert delta == dt_ns, (
+                f"expected delta {dt_ns} ns between sample {i} and {i+1}, "
+                f"got {delta} ns"
+            )
+
+    def test_burst_at_200hz_8_samples(self):
+        """BLE IMU profile: 8 samples per notification at 200 Hz."""
+        anchor = 5_000_000_000_000
+        ts = burst_timestamps(8, anchor_ns=anchor, expected_hz=200.0)
+        assert ts[-1] == anchor
+        dt_ns = round(1e9 / 200.0)  # 5_000_000 ns = 5 ms
+        for i in range(7):
+            assert ts[i + 1] - ts[i] == dt_ns
+
+    def test_ascending_order(self):
+        anchor = 9_000_000_000_000
+        ts = burst_timestamps(10, anchor_ns=anchor, expected_hz=500.0)
+        assert ts == sorted(ts), "timestamps must be in ascending order"
+
+    def test_invalid_n_raises(self):
+        with pytest.raises(ValueError, match="n must be >= 1"):
+            burst_timestamps(0, anchor_ns=0, expected_hz=100.0)
+
+    def test_invalid_hz_raises(self):
+        with pytest.raises(ValueError, match="expected_hz must be > 0"):
+            burst_timestamps(5, anchor_ns=0, expected_hz=0.0)
+
+    def test_anchor_defaults_to_now(self):
+        """When anchor_ns is omitted, the last timestamp is close to now."""
+        before = time.monotonic_ns()
+        ts = burst_timestamps(3, expected_hz=100.0)
+        after = time.monotonic_ns()
+        # Last timestamp should be within a reasonable range around call time
+        assert before <= ts[-1] <= after + 1_000_000  # 1 ms slack
+
+    def test_timestamps_fed_to_push_have_correct_spacing(self):
+        """push() accepts burst-spaced timestamps and stores them correctly."""
+        anchor = 2_000_000_000_000
+        ts = burst_timestamps(3, anchor_ns=anchor, expected_hz=1000.0)
+        stream = PushSensorStream("burst_test")
+        stream.connect()
+
+        captured: List[int] = []
+        stream.on_sample(lambda ev: captured.append(ev.capture_ns))
+
+        for t in ts:
+            stream.push({"x": 1.0}, capture_ns=t)
+
+        assert captured == ts, "pushed capture_ns values must match burst timestamps"
+
+
+# ---------------------------------------------------------------------------
+# Contract 3 — Transient transport hiccup auto-reopen
+# ---------------------------------------------------------------------------
+
+
+class TestContract3TransientReopen:
+    """retry_open must retry up to TRANSIENT_REOPEN_MAX_ATTEMPTS times."""
+
+    def test_retry_open_succeeds_on_third_attempt(self):
+        """retry_open must retry and return the handle on eventual success."""
+        call_count = [0]
+        sentinel = object()
+
+        def flaky_open():
+            call_count[0] += 1
+            if call_count[0] < 3:
+                raise OSError("transient failure")
+            return sentinel
+
+        result = retry_open(
+            flaky_open,
+            max_attempts=5,
+            max_wait_s=0.0,  # no real sleep in unit test
+            stream_id="test_sensor",
+        )
+        assert result is sentinel
+        assert call_count[0] == 3
+
+    def test_retry_open_raises_after_max_attempts(self):
+        """retry_open must re-raise after exhausting all attempts."""
+        call_count = [0]
+
+        def always_fails():
+            call_count[0] += 1
+            raise OSError("persistent failure")
+
+        with pytest.raises(OSError, match="persistent failure"):
+            retry_open(
+                always_fails,
+                max_attempts=TRANSIENT_REOPEN_MAX_ATTEMPTS,
+                max_wait_s=0.0,
+                stream_id="persistent_sensor",
+            )
+        assert call_count[0] == TRANSIENT_REOPEN_MAX_ATTEMPTS
+
+    def test_retry_open_succeeds_on_first_attempt(self):
+        """When open_fn succeeds immediately, result is returned with 1 call."""
+        handle = object()
+        call_count = [0]
+
+        def good_open():
+            call_count[0] += 1
+            return handle
+
+        result = retry_open(good_open, stream_id="ok_sensor")
+        assert result is handle
+        assert call_count[0] == 1
+
+    def test_polling_sensor_retries_open_on_transient_error(self, monkeypatch):
+        """PollingSensorStream.connect() must retry a flaky open callback."""
+        open_count = [0]
+        handle = object()
+
+        def flaky_open():
+            open_count[0] += 1
+            if open_count[0] < 3:
+                raise OSError("transient USB hiccup")
+            return handle
+
+        # Patch time.sleep inside _generic so the test doesn't actually wait
+        monkeypatch.setattr("syncfield.adapters._generic.time.sleep", lambda _s: None)
+
+        stream = PollingSensorStream(
+            "serial_sensor",
+            read=lambda h: {"x": 1.0},
+            hz=10.0,
+            open=flaky_open,
+        )
+        stream.connect()
+        assert open_count[0] == 3, (
+            f"expected 3 open attempts (2 transient + 1 success), "
+            f"got {open_count[0]}"
+        )
+        stream.disconnect()
+
+    def test_polling_sensor_raises_after_max_failed_opens(self, monkeypatch):
+        """PollingSensorStream.connect() re-raises when all retries are exhausted."""
+        open_count = [0]
+
+        def always_fails():
+            open_count[0] += 1
+            raise OSError("persistent hardware failure")
+
+        monkeypatch.setattr("syncfield.adapters._generic.time.sleep", lambda _s: None)
+
+        stream = PollingSensorStream(
+            "dead_sensor",
+            read=lambda h: {"x": 1.0},
+            hz=10.0,
+            open=always_fails,
+        )
+        with pytest.raises(OSError, match="persistent hardware failure"):
+            stream.connect()
+
+        assert open_count[0] == TRANSIENT_REOPEN_MAX_ATTEMPTS
+
+
+# ---------------------------------------------------------------------------
+# Contract 4 — SyncToneConfig.silent() MUST NOT register host_audio
+# ---------------------------------------------------------------------------
+
+
+class TestContract4SilentNoHostAudio:
+    """SyncToneConfig.silent() must suppress auto-injection of host_audio."""
+
+    def test_silent_config_has_suppress_host_audio_true(self):
+        """silent() factory must set suppress_host_audio=True."""
+        cfg = SyncToneConfig.silent()
+        assert cfg.suppress_host_audio is True
+
+    def test_default_config_does_not_suppress_host_audio(self):
+        """default() and audible() must leave suppress_host_audio=False."""
+        assert SyncToneConfig.default().suppress_host_audio is False
+        assert SyncToneConfig.audible().suppress_host_audio is False
+
+    def test_manual_enabled_false_does_not_suppress_host_audio(self):
+        """Existing code using SyncToneConfig(enabled=False) keeps old behaviour."""
+        cfg = SyncToneConfig(enabled=False)
+        assert cfg.suppress_host_audio is False
+
+    def test_silent_session_excludes_host_audio_from_add(self, tmp_path):
+        """add() must not pre-register host_audio when sync_tone is silent()."""
+        # Simulate an environment where a mic IS available so we can confirm
+        # the suppress flag is what's blocking injection, not absence of HW.
+        with patch(
+            "syncfield.adapters.host_audio.is_audio_available", return_value=True
+        ), patch(
+            "syncfield.adapters.host_audio.HostAudioStream"
+        ) as mock_has:
+            session = SessionOrchestrator(
+                host_id="rig",
+                output_dir=tmp_path,
+                sync_tone=SyncToneConfig.silent(),
+            )
+            session.add(FakeStream("cam"))
+
+        assert "host_audio" not in session._streams
+        mock_has.assert_not_called()
+
+    def test_silent_session_excludes_host_audio_from_connect(self, tmp_path):
+        """connect() must not inject host_audio when sync_tone is silent()."""
+        with patch(
+            "syncfield.adapters.host_audio.is_audio_available", return_value=True
+        ), patch(
+            "syncfield.adapters.host_audio.HostAudioStream"
+        ) as mock_has:
+            session = SessionOrchestrator(
+                host_id="rig",
+                output_dir=tmp_path,
+                sync_tone=SyncToneConfig.silent(),
+            )
+            session.add(FakeStream("cam"))
+            session.connect()
+
+        assert "host_audio" not in session._streams
+        mock_has.assert_not_called()
+        session.disconnect()
+
+    def test_non_silent_session_suppress_host_audio_is_false(self):
+        """Sessions with default() sync_tone have suppress_host_audio=False.
+
+        This is the structural contract: the preregister helper checks
+        suppress_host_audio, and for non-silent configs the flag is False.
+        We verify the flag directly since the conftest autouse fixture
+        patches out the helper itself (to avoid needing real audio HW in
+        every test).
+        """
+        cfg = SyncToneConfig.default()
+        assert cfg.suppress_host_audio is False, (
+            "default() sync_tone must NOT suppress host_audio injection"
+        )
+        # The orchestrator reads _sync_tone.suppress_host_audio in both helper
+        # methods — confirm it would be honoured at the config level.
+        cfg2 = SyncToneConfig.audible()
+        assert cfg2.suppress_host_audio is False
+
+
+# ---------------------------------------------------------------------------
+# Contract 5 — Stream errors MUST NOT propagate to SessionOrchestrator
+# ---------------------------------------------------------------------------
+
+
+class _FailingCaptureStream(FakeStream):
+    """FakeStream whose capture 'thread' raises on first read after start_recording.
+
+    Simulates a push-sensor whose user code crashes after recording starts.
+    The crash must not bubble out of the session.
+    """
+
+    def __init__(self, id: str, **kwargs: Any) -> None:
+        super().__init__(id=id, **kwargs)
+        self._capture_thread: Optional[threading.Thread] = None
+        self.captured_error: Optional[Exception] = None
+        self.emitted_samples = 0
+        self._stop_event = threading.Event()
+
+    def connect(self) -> None:
+        """No-op; capture starts in start_recording for simplicity."""
+        pass
+
+    def start_recording(self, session_clock) -> None:  # type: ignore[override]
+        super().start(session_clock)  # reuse legacy start for counters
+        self._stop_event.clear()
+        self._capture_thread = threading.Thread(
+            target=self._failing_loop,
+            daemon=True,
+        )
+        self._capture_thread.start()
+
+    def _failing_loop(self) -> None:
+        try:
+            raise RuntimeError("simulated sensor hardware crash")
+        except RuntimeError as exc:
+            self.captured_error = exc
+            # The error stays in the thread — it must not propagate to
+            # the orchestrator. FakeStream.stop() returns the report.
+
+    def stop_recording(self) -> Any:
+        self._stop_event.set()
+        if self._capture_thread is not None:
+            self._capture_thread.join(timeout=1.0)
+        return super().stop()
+
+    def disconnect(self) -> None:
+        pass
+
+
+class _StableStream(FakeStream):
+    """FakeStream that emits samples steadily for the duration of a test."""
+
+    def __init__(self, id: str, **kwargs: Any) -> None:
+        super().__init__(id=id, **kwargs)
+        self.received_samples = 0
+        self._sample_thread: Optional[threading.Thread] = None
+        self._stop_event = threading.Event()
+
+    def on_sample(self, callback) -> None:  # type: ignore[override]
+        super().on_sample(callback)
+
+    def connect(self) -> None:
+        pass
+
+    def start_recording(self, session_clock) -> None:  # type: ignore[override]
+        super().start(session_clock)
+        self._stop_event.clear()
+        self._sample_thread = threading.Thread(
+            target=self._emit_loop, daemon=True
+        )
+        self._sample_thread.start()
+
+    def _emit_loop(self) -> None:
+        while not self._stop_event.wait(timeout=0.01):
+            self.push_sample(frame_number=self.received_samples, capture_ns=time.monotonic_ns())
+            self.received_samples += 1
+
+    def stop_recording(self) -> Any:
+        self._stop_event.set()
+        if self._sample_thread is not None:
+            self._sample_thread.join(timeout=1.0)
+        return super().stop()
+
+    def disconnect(self) -> None:
+        pass
+
+
+class TestContract5StreamErrorIsolation:
+    """A stream's runtime crash MUST NOT kill the session.
+
+    After one stream's capture thread raises, other streams must still
+    receive data and the session state must remain RECORDING until stop().
+    """
+
+    def test_failing_stream_does_not_kill_session_state(self, tmp_path):
+        """Session stays in RECORDING even when one stream crashes."""
+        stable = _StableStream("healthy_cam")
+        crashing = _FailingCaptureStream("crashing_sensor")
+
+        session = _session(tmp_path)
+        session.add(stable)
+        session.add(crashing)
+        session.start()
+
+        # Allow the crash to happen
+        time.sleep(0.05)
+
+        # Session must still be in RECORDING
+        assert session.state is SessionState.RECORDING, (
+            f"Session should stay RECORDING after one stream crashes; "
+            f"got state={session.state}"
+        )
+
+        session.stop()
+        session.disconnect()
+
+    def test_failing_stream_error_stays_in_stream_thread(self, tmp_path):
+        """The crash is captured in the stream thread, not the session."""
+        crashing = _FailingCaptureStream("crashing_sensor")
+
+        session = _session(tmp_path)
+        session.add(crashing)
+        session.start()
+
+        time.sleep(0.05)
+
+        # The error must be captured inside the stream, not raised externally
+        assert crashing.captured_error is not None
+        assert isinstance(crashing.captured_error, RuntimeError)
+
+        session.stop()
+        session.disconnect()
+
+    def test_healthy_stream_receives_samples_while_sibling_crashes(self, tmp_path):
+        """Healthy stream continues emitting after its sibling crashes."""
+        stable = _StableStream("healthy_cam")
+        crashing = _FailingCaptureStream("crashing_sensor")
+
+        session = _session(tmp_path)
+        session.add(stable)
+        session.add(crashing)
+        session.start()
+
+        # Give healthy stream time to produce samples
+        time.sleep(0.1)
+
+        assert stable.received_samples > 0, (
+            f"Healthy stream should have produced samples by now; "
+            f"got {stable.received_samples}"
+        )
+
+        session.stop()
+        session.disconnect()
+
+    def test_stop_is_clean_when_one_stream_crashed(self, tmp_path):
+        """stop() must not raise even if one stream's capture thread crashed."""
+        crashing = _FailingCaptureStream("crashing_sensor")
+
+        session = _session(tmp_path)
+        session.add(crashing)
+        session.start()
+
+        time.sleep(0.05)
+
+        # stop() must succeed without raising
+        report = session.stop()
+        assert report is not None
+        session.disconnect()