diff --git a/openpilot/starpilot/flm/fullrate_collector.py b/openpilot/starpilot/flm/fullrate_collector.py
index febd76f297..f7df5dfa34 100644
--- a/openpilot/starpilot/flm/fullrate_collector.py
+++ b/openpilot/starpilot/flm/fullrate_collector.py
@@ -425,7 +425,9 @@ class FullrateCollector:
         self.split('changed_duplicate_control', reset=True)
         return False
       return True
-    if self.controls and value.control_ns <= next(reversed(self.controls)):
+    # A joined command leaves controls but remains part of this producer tape.
+    # Cross-stream ingestion must not hide its ordering boundary after a join.
+    if self.recent_controls and value.control_ns <= next(reversed(self.recent_controls)):
       self.split('reordered_control')
       return False
     self._last_live_ingest_ns = now_ns
@@ -877,6 +879,22 @@ class FullrateCollector:
     return self.result
 
 
+def _check_evidence_wire(event):
+  wire_bytes = len(event.as_builder().to_bytes())
+  if wire_bytes > MAX_SNAPSHOT_WIRE_BYTES:
+    raise ValueError('wire_bound')
+  if wire_bytes > MAX_WIRE_BYTES:
+    kind = event.which()
+    control_snapshot = kind == 'starpilotLateralState' and bool(len(event.starpilotLateralState.learningEvidence.replaySnapshot))
+    ford_events = (
+      kind == 'starpilotCarState'
+      and event.starpilotCarState.fordLearningEvidence.version == 1
+      and 0 < len(event.starpilotCarState.fordLearningEvidence.rows) <= 4
+    )
+    if not (control_snapshot or ford_events):
+      raise ValueError('non_snapshot_wire_bound')
+
+
 def drain(socket, decode, receive, collector, now_ns):
   """Nonblocking bounded existing transport; overflow rejects whole maneuver."""
   for _ in range(MAX_DRAIN):
@@ -884,19 +902,7 @@ def drain(socket, decode, receive, collector, now_ns):
     if event is None:
       return True
     try:
-      wire_bytes = len(event.as_builder().to_bytes())
-      if wire_bytes > MAX_SNAPSHOT_WIRE_BYTES:
-        raise ValueError('wire_bound')
-      if wire_bytes > MAX_WIRE_BYTES:
-        kind = event.which()
-        control_snapshot = kind == 'starpilotLateralState' and bool(len(event.starpilotLateralState.learningEvidence.replaySnapshot))
-        ford_events = (
-          kind == 'starpilotCarState'
-          and event.starpilotCarState.fordLearningEvidence.version == 1
-          and 0 < len(event.starpilotCarState.fordLearningEvidence.rows) <= 4
-        )
-        if not (control_snapshot or ford_events):
-          raise ValueError('non_snapshot_wire_bound')
+      _check_evidence_wire(event)
       decode(event, collector, now_ns)
     except (AttributeError, KeyError, TypeError, ValueError, OverflowError):
       collector.split('malformed_evidence', reset=True)
@@ -906,3 +912,50 @@ def drain(socket, decode, receive, collector, now_ns):
     collector.overrun()
     return False
   return True
+
+
+class _DrainOverrun(Exception):
+  pass
+
+
+def drain_pair(control_socket, control, applied_socket, applied, receive, collector, now_ns):
+  """Merge two bounded FIFO tapes by their original publisher headers.
+
+  Only the two queue heads compete; never sort a producer's own tape. A held
+  command can be republished by Card, so embedded control/apply clocks are not
+  transport ordering keys. Headers choose order only: the original callbacks
+  still decode the untouched event and resample source/native/receiver clocks.
+  Equal headers put control first, matching the previous causal preference.
+  Malformed input or either stream overrunning remains a maneuver boundary.
+  """
+  sockets, callbacks = (control_socket, applied_socket), (control, applied)
+  counts = [0, 0]
+
+  def head(index):
+    event = receive(sockets[index])
+    if event is None:
+      return None
+    # This is the same one-message bounded lookahead as the single-stream drain.
+    if counts[index] == MAX_DRAIN:
+      raise _DrainOverrun
+    _check_evidence_wire(event)
+    stamp = event.logMonoTime
+    if type(stamp) is not int or stamp <= 0:
+      raise ValueError('invalid_publisher_header')
+    return event, stamp
+
+  try:
+    pending = [head(0), head(1)]
+    while pending[0] is not None or pending[1] is not None:
+      index = 0 if pending[1] is None or (pending[0] is not None and pending[0][1] <= pending[1][1]) else 1
+      event, _stamp = pending[index]
+      callbacks[index](event, collector, now_ns)
+      counts[index] += 1
+      pending[index] = head(index)
+  except _DrainOverrun:
+    collector.overrun()
+    return False
+  except (AttributeError, KeyError, TypeError, ValueError, OverflowError):
+    collector.split('malformed_evidence', reset=True)
+    return False
+  return True
diff --git a/openpilot/starpilot/flm/steerlearnerd.py b/openpilot/starpilot/flm/steerlearnerd.py
index 6d139ad9a8..215fcf60ce 100644
--- a/openpilot/starpilot/flm/steerlearnerd.py
+++ b/openpilot/starpilot/flm/steerlearnerd.py
@@ -16,7 +16,7 @@ from functools import partial
 from opendbc.car.structs import car
 from openpilot.cereal import messaging
 from openpilot.common.params import Params
-from openpilot.starpilot.flm.fullrate_collector import FullrateCollector, drain, ControlEvidence, AppliedEvidence
+from openpilot.starpilot.flm.fullrate_collector import FullrateCollector, drain_pair, ControlEvidence, AppliedEvidence
 from openpilot.starpilot.flm.learning_wire import read_control, read_applied, result_message, write_source_receipt
 from openpilot.starpilot.flm.runtime_adapter_registry import SOURCE_LEASE_NS
 
@@ -48,7 +48,7 @@ class _SourceDelivery:
 
 
 def source_receiver(services):
-  # Full-rate control/application tapes are drained separately. This optional
+  # Full-rate control/application tapes retain their original FIFO rows. This optional
   # process conflates CarState, so its receive count cannot measure the 100Hz
   # publisher rate while it processes a batch. Keep the actual envelope,
   # source-age, state-bracket and native checks; do not fabricate freq_ok.
@@ -645,8 +645,7 @@ def run(
               return
           _collect_applied_output(collector, applied, sm, CP, received_now)
 
-      drain(control_socket, control, messaging.recv_one_or_none, sink, now)
-      drain(applied_socket, output, messaging.recv_one_or_none, sink, now)
+      drain_pair(control_socket, control, applied_socket, output, messaging.recv_one_or_none, sink, now)
       if collector is not None:
         # Decoding queued frames may outlast the sampled shared inputs. Consume
         # real pending deliveries before pose/native checks; keep their original
diff --git a/openpilot/starpilot/flm/tests/_drain_fixture.py b/openpilot/starpilot/flm/tests/_drain_fixture.py
new file mode 100644
index 0000000000..2d1824d819
--- /dev/null
+++ b/openpilot/starpilot/flm/tests/_drain_fixture.py
@@ -0,0 +1,12 @@
+"""Adapt declared callback schedules in daemon units, not production transport.
+
+These tests deliberately drive callback time/source/native boundaries directly.
+The actual FIFO merge and bounded wire receiver have their own transport tests.
+"""
+
+
+def separate_drain_fixture(single):
+  def paired(control_socket, control, applied_socket, applied, receive, collector, now_ns):
+    single(control_socket, control, receive, collector, now_ns)
+    single(applied_socket, applied, receive, collector, now_ns)
+  return paired
diff --git a/openpilot/starpilot/flm/tests/test_drain_receipt_clock.py b/openpilot/starpilot/flm/tests/test_drain_receipt_clock.py
index e8fd27d4b3..f093e4db70 100644
--- a/openpilot/starpilot/flm/tests/test_drain_receipt_clock.py
+++ b/openpilot/starpilot/flm/tests/test_drain_receipt_clock.py
@@ -1,4 +1,5 @@
 """Real learner run callback clocks: deterministic transport and monotonic clock."""
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from types import SimpleNamespace
 from unittest import mock
 
@@ -44,7 +45,7 @@ def exercise(*, control_offset=10_000_000, callback_offset=20_000_000,
       callback(applied, sink, captured_now)
       raise StopIteration  # Stop only after both real run callbacks execute.
 
-  with mock.patch.object(steerlearnerd, 'drain', drain), \
+  with mock.patch.object(steerlearnerd, 'drain_pair', separate_drain_fixture(drain)), \
        mock.patch.object(steerlearnerd, 'read_control', return_value=evidence), \
        mock.patch.object(steerlearnerd, 'read_applied', return_value=output_value), \
        mock.patch.object(steerlearnerd, 'FullrateCollector', return_value=collector):
@@ -100,7 +101,7 @@ def test_post_drain_health_and_fit_use_current_time():
     assert captured_now == NOW
     receipt_clock[0] += 10_000_000
 
-  with mock.patch.object(steerlearnerd, 'drain', drain):
+  with mock.patch.object(steerlearnerd, 'drain_pair', separate_drain_fixture(drain)):
     with pytest.raises(StopIteration):
       steerlearnerd.run(SimpleNamespace(), sm, object(), object(), mock.Mock(),
         mock.Mock(), mock.Mock(), mock.Mock(return_value=source), mock.Mock(),
diff --git a/openpilot/starpilot/flm/tests/test_event_application_schedule.py b/openpilot/starpilot/flm/tests/test_event_application_schedule.py
index c1eac47097..789ee4fbb4 100644
--- a/openpilot/starpilot/flm/tests/test_event_application_schedule.py
+++ b/openpilot/starpilot/flm/tests/test_event_application_schedule.py
@@ -4,6 +4,7 @@ The child output is an opaque proposal deliberately rejected by adapter install.
 Real typed control/output/physical collector/source guards still run. Synthetic
 native health is explicit; this is not controller/native/physical qualification.
 """
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from dataclasses import replace
 from types import SimpleNamespace
 from unittest import mock
@@ -107,7 +108,7 @@ def exercise(*, failure=None):
   publisher=mock.Mock()
   publisher.send.side_effect=send
   proposal=(SimpleNamespace(profile=object()), object())
-  with mock.patch.object(steerlearnerd,'drain',drain), \
+  with mock.patch.object(steerlearnerd,'drain_pair',separate_drain_fixture(drain)), \
        mock.patch('openpilot.starpilot.flm.event_collector.EventFullrateCollector',Collector), \
        mock.patch('openpilot.starpilot.flm.native_health.current_native_health',
          side_effect=lambda *args: not (failure == 'native' and phase[0] >= 5)):
diff --git a/openpilot/starpilot/flm/tests/test_event_application_success.py b/openpilot/starpilot/flm/tests/test_event_application_success.py
index e78cdd2345..a04481a734 100644
--- a/openpilot/starpilot/flm/tests/test_event_application_success.py
+++ b/openpilot/starpilot/flm/tests/test_event_application_success.py
@@ -3,6 +3,7 @@
 NOT a route-fit/registered-native/model qualification. BoundaryAdapter is the
 existing synthetic lifecycle adapter; scientific validation is not invented.
 """
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from dataclasses import asdict, replace
 from types import SimpleNamespace
 from unittest import mock
@@ -132,7 +133,7 @@ def exercise(*, no_new_pose=False):
     sent.append((phase[0], admission))
   publisher = mock.Mock()
   publisher.send.side_effect = send
-  with mock.patch.object(steerlearnerd, 'drain', drain), \
+  with mock.patch.object(steerlearnerd, 'drain_pair', separate_drain_fixture(drain)), \
        mock.patch('openpilot.starpilot.flm.event_collector.EventFullrateCollector', Collector), \
        mock.patch('openpilot.starpilot.flm.native_health.current_native_health', return_value=True):
     with pytest.raises(StopIteration):
diff --git a/openpilot/starpilot/flm/tests/test_event_fit_reporting.py b/openpilot/starpilot/flm/tests/test_event_fit_reporting.py
index f6060343e6..be79b18093 100644
--- a/openpilot/starpilot/flm/tests/test_event_fit_reporting.py
+++ b/openpilot/starpilot/flm/tests/test_event_fit_reporting.py
@@ -1,4 +1,5 @@
 """Actual optional publisher cache reports child rejection without admission."""
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from types import SimpleNamespace
 from unittest import mock
 
@@ -81,7 +82,7 @@ def test_actual_event_cached_child_reports_reason_only_for_exact_current_prior(s
     return source
 
   with mock.patch('openpilot.starpilot.flm.event_collector.EventFullrateCollector', return_value=collector), \
-       mock.patch.object(steerlearnerd, 'drain', drain), \
+       mock.patch.object(steerlearnerd, 'drain_pair', separate_drain_fixture(drain)), \
        mock.patch.object(steerlearnerd, 'read_control', return_value=evidence), \
        mock.patch.object(collector, 'control'), mock.patch.object(collector, 'physical'), \
        mock.patch.object(collector, 'job_snapshot', side_effect=[{'job': 'bounded'}, None, None]), \
diff --git a/openpilot/starpilot/flm/tests/test_fit_reporting.py b/openpilot/starpilot/flm/tests/test_fit_reporting.py
index 26b03d960e..3e4e83e340 100644
--- a/openpilot/starpilot/flm/tests/test_fit_reporting.py
+++ b/openpilot/starpilot/flm/tests/test_fit_reporting.py
@@ -1,4 +1,5 @@
 """Actual optional publisher cache reports child rejection without admission."""
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from types import SimpleNamespace
 from unittest import mock
 
@@ -78,7 +79,7 @@ def test_actual_cached_child_without_profile_reports_reason_without_admission(st
     return source
 
   with mock.patch.object(steerlearnerd, 'FullrateCollector', return_value=collector), \
-       mock.patch.object(steerlearnerd, 'drain', drain), \
+       mock.patch.object(steerlearnerd, 'drain_pair', separate_drain_fixture(drain)), \
        mock.patch.object(steerlearnerd, 'read_control', return_value=evidence), \
        mock.patch.object(collector, 'control'), mock.patch.object(collector, 'physical'), \
        mock.patch.object(collector, 'job_snapshot', side_effect=[{'job': 'bounded'}, None, None]), \
diff --git a/openpilot/starpilot/flm/tests/test_fresh_publication.py b/openpilot/starpilot/flm/tests/test_fresh_publication.py
index fa05c1e782..dfad0440a9 100644
--- a/openpilot/starpilot/flm/tests/test_fresh_publication.py
+++ b/openpilot/starpilot/flm/tests/test_fresh_publication.py
@@ -32,7 +32,7 @@ def publication(*, fresh, changes=None, snapshot_cost=0):
     raise StopIteration
 
   publisher.send.side_effect = send
-  with mock.patch.object(steerlearnerd, 'drain', return_value=None):
+  with mock.patch.object(steerlearnerd, 'drain_pair', return_value=None):
     with pytest.raises(StopIteration):
       steerlearnerd.run(SimpleNamespace(), sm, object(), object(), publisher,
         mock.Mock(), mock.Mock(), mock.Mock(return_value=old), mock.Mock(), mock.Mock(),
diff --git a/openpilot/starpilot/flm/tests/test_learner_unavailable_reason.py b/openpilot/starpilot/flm/tests/test_learner_unavailable_reason.py
index 282c3f9954..3d0d7b8705 100644
--- a/openpilot/starpilot/flm/tests/test_learner_unavailable_reason.py
+++ b/openpilot/starpilot/flm/tests/test_learner_unavailable_reason.py
@@ -1,4 +1,5 @@
 """Existing status publications must name settings blocks accurately."""
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from dataclasses import replace
 from types import SimpleNamespace
 from unittest import mock
@@ -36,7 +37,7 @@ def publication(source, evidence=None, adapter=None, sources=None):
                               starpilotLateralState=SimpleNamespace(learningEvidence=SimpleNamespace(version=1)))
       callback(event, now, now)
 
-  with mock.patch.object(steerlearnerd, 'drain', drain), mock.patch.object(steerlearnerd, 'read_control', return_value=evidence):
+  with mock.patch.object(steerlearnerd, 'drain_pair', separate_drain_fixture(drain)), mock.patch.object(steerlearnerd, 'read_control', return_value=evidence):
     with pytest.raises(StopIteration):
       steerlearnerd.run(SimpleNamespace(), sm, control_socket, applied_socket, pm, factory, mock.Mock(),
                         source_reader, lambda item: (('feedforward', 'low', 'steady'),),
diff --git a/openpilot/starpilot/flm/tests/test_live_tracking_daemon.py b/openpilot/starpilot/flm/tests/test_live_tracking_daemon.py
index 4fa0e68ca0..6b556e4d65 100644
--- a/openpilot/starpilot/flm/tests/test_live_tracking_daemon.py
+++ b/openpilot/starpilot/flm/tests/test_live_tracking_daemon.py
@@ -1,4 +1,5 @@
 """Actual daemon loop and cereal publication; fake actor does not model permission."""
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 
 from dataclasses import replace
 from types import SimpleNamespace
@@ -85,7 +86,7 @@ def test_actual_daemon_publishes_separate20_and4_without_legacy_fit(monkeypatch,
 
   monkeypatch.setattr(actor, 'close', actor_close)
   # No control/native is invented: this exercises actual collecting publications.
-  monkeypatch.setattr(steerlearnerd, 'drain', lambda *args: None)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', lambda *args: None)
   with pytest.raises(LoopComplete if shutdown_reaped else RuntimeError):
     steerlearnerd.run(
       SimpleNamespace(),
@@ -169,7 +170,7 @@ def _run_poll_script(monkeypatch, rows, *, expire_during_drain=False):
       now[0] += 600_000_000
       expired = True
 
-  monkeypatch.setattr(steerlearnerd, 'drain', drain)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', separate_drain_fixture(drain))
   with pytest.raises(LoopComplete):
     steerlearnerd.run(SimpleNamespace(), sm, object(), object(), pm, None, None,
       lambda stamp: source(stamp), None, None, None, None, fitter, None,
@@ -250,7 +251,7 @@ def test_actual_nonlive_run_does_not_read_tracking_poll_metadata(monkeypatch):
   # Minimal SM intentionally lacks tracking metadata; the legacy path is intact.
   sent = []
   pm = SimpleNamespace(send=lambda topic, message: sent.append(topic))
-  monkeypatch.setattr(steerlearnerd, 'drain', lambda *args: None)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', lambda *args: None)
   fitter = SimpleNamespace(poll=lambda **kwargs: None, close=lambda reason: True, child=None)
   with pytest.raises(LoopComplete):
     steerlearnerd.run(SimpleNamespace(), SM(), object(), object(), pm, None, None,
diff --git a/openpilot/starpilot/flm/tests/test_neutral_status_cadence.py b/openpilot/starpilot/flm/tests/test_neutral_status_cadence.py
index 5fd85c7da7..de4b943d00 100644
--- a/openpilot/starpilot/flm/tests/test_neutral_status_cadence.py
+++ b/openpilot/starpilot/flm/tests/test_neutral_status_cadence.py
@@ -25,7 +25,7 @@ def test_neutral_status_meets_declared_frequency_without_renewing_physical_proof
   def send(topic, message):
     if topic == 'starpilotSteeringTracking':
       sends.append((now[0], message.as_reader()))
-  monkeypatch.setattr(steerlearnerd, 'drain', lambda *args: None)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', lambda *args: None)
   actor, fitter = RecordingActor(), NoLegacyFitter()
   fitter.cancel = lambda reason: None
   actor.interrupt = lambda reason, **kwargs: None
@@ -74,7 +74,7 @@ def test_explicit_learning_off_still_publishes_withdrawal_on_the_next_source_pac
   def send(topic, message):
     if topic == 'starpilotSteeringTracking':
       sends.append((now[0], message.as_reader()))
-  monkeypatch.setattr(steerlearnerd, 'drain', lambda *args: None)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', lambda *args: None)
   with pytest.raises(LoopComplete):
     steerlearnerd.run(SimpleNamespace(), SM(), object(), object(), SimpleNamespace(send=send),
       None, None, current, None, None, None, None, fitter, None, clock=lambda: now[0], live_actor=actor)
@@ -107,7 +107,7 @@ def test_missing_source_does_not_enter_the_explicit_withdrawal_serializer(monkey
   def send(topic, message):
     if topic == 'starpilotSteeringTracking':
       sends.append((now[0], message.as_reader()))
-  monkeypatch.setattr(steerlearnerd, 'drain', lambda *args: None)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', lambda *args: None)
   with pytest.raises(LoopComplete):
     steerlearnerd.run(SimpleNamespace(), SM(), object(), object(), SimpleNamespace(send=send),
       None, None, current, None, None, None, None, fitter, None, clock=lambda: now[0], live_actor=actor)
diff --git a/openpilot/starpilot/flm/tests/test_paired_drain_daemon.py b/openpilot/starpilot/flm/tests/test_paired_drain_daemon.py
new file mode 100644
index 0000000000..baf959f3e8
--- /dev/null
+++ b/openpilot/starpilot/flm/tests/test_paired_drain_daemon.py
@@ -0,0 +1,103 @@
+"""Real daemon/transport/public-Panda parser with declared health, not road proof."""
+from collections import deque
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from opendbc.car.car_helpers import interfaces
+from opendbc.car.toyota.values import CAR
+
+from openpilot.cereal import messaging
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.starpilot.flm import steerlearnerd
+from openpilot.starpilot.flm.fullrate_collector import FullrateCollector
+from openpilot.starpilot.flm.runtime_adapter_registry import SourceState, car_params_digest
+from openpilot.starpilot.flm.tests.test_fullrate_pipeline import SyntheticAdapter, snapshots, serialized_control, serialized_output
+
+BASE = 10_000_000_000
+
+
+@pytest.mark.parametrize('native_allowed', (True, False))
+def test_actual_merged_daemon_callbacks_keep_fresh_tail_and_current_native_checks(monkeypatch, native_allowed):
+  class Finished(Exception):
+    pass
+
+  now = [BASE + 160_000_000]
+  cp = interfaces[CAR.TOYOTA_RAV4_TSS2].get_non_essential_params(CAR.TOYOTA_RAV4_TSS2)
+  source = SourceState(True, False, '0', '2' * 64, BASE - 10_000_000, BASE + 490_000_000,
+    car_params_digest(cp), (0., 0., 0.), 'c' * 32, 'd' * 64)
+  calls, instances, physical = [], [], []
+
+  class Adapter(SyntheticAdapter):
+    def select_objective(self, key):
+      pass
+
+    def prepare_control(self, value):
+      return value
+
+  class Collector(FullrateCollector):
+    def control(self, value, stamp):
+      accepted = super().control(value, stamp)
+      calls.append(('control', value.control_ns, self._learning_current_native_allowed, accepted, stamp))
+      return accepted
+
+    def output(self, value, stamp):
+      accepted = super().output(value, stamp)
+      calls.append(('output', value.apply_ns, self._learning_current_native_allowed, accepted, stamp))
+      return accepted
+
+  def factory(adapter):
+    collector = Collector(adapter)
+    instances.append(collector)
+    return collector
+
+  sockets = object(), object()
+  rows = [snapshots(BASE + offset)[:2] for offset in (0, 150_000_000)]
+  queues = {sockets[0]: deque(serialized_control(c) for c, _ in rows),
+            sockets[1]: deque(serialized_output(o) for _, o in rows)}
+
+  def receive(socket):
+    return queues[socket].popleft() if queues[socket] else None
+
+  def clock():
+    now[0] += 100
+    return now[0]
+
+  def observe(_sm, _cp, collector, stamp):
+    physical.append((tuple(collector.ready), stamp, collector._learning_current_native_allowed))
+    return None  # No physical evidence is invented for this acquisition unit.
+
+  fitter = Mock(child=object())
+  fitter.poll.side_effect = Finished
+  monkeypatch.setattr(steerlearnerd, 'FullrateCollector', factory)
+  monkeypatch.setattr(steerlearnerd.messaging, 'recv_one_or_none', receive)
+  with OpenpilotPrefix():
+    sm = messaging.SubMaster(['pandaStates', 'deviceMotion'])
+    # Actual public message decoding/checks, with explicitly declared test state.
+    for stamp in (BASE - 40_000_000, BASE + 60_000_000, BASE + 150_000_000):
+      event = messaging.new_message('pandaStates', len(cp.safetyConfigs), valid=True, logMonoTime=stamp)
+      for panda, config in zip(event.pandaStates, cp.safetyConfigs, strict=True):
+        panda.safetyModel, panda.safetyParam = config.safetyModel, config.safetyParam
+        panda.alternativeExperience = cp.alternativeExperience
+        panda.controlsAllowed = native_allowed and str(config.safetyModel) not in ('silent', 'noOutput')
+      sm.update_msgs(stamp / 1e9, [messaging.log_from_bytes(event.to_bytes())])
+    assert sm.freq_ok['pandaStates']
+    updates = []
+    monkeypatch.setattr(sm, 'update', lambda timeout=None: updates.append(timeout))
+    with pytest.raises(Finished):
+      steerlearnerd.run(cp, sm, *sockets, SimpleNamespace(send=lambda *args: None),
+        lambda *args: Adapter(), observe, lambda stamp: source,
+        lambda row: ('fixture',), lambda row: 'fixture', lambda *args: None,
+        None, fitter, None, clock=clock)
+  assert len(instances) == 1 and all(not queue for queue in queues.values())
+  assert [kind for kind, *_ in calls] == ['control', 'output', 'control', 'output']
+  assert [accepted for *_, accepted, _stamp in calls] == [False, False, True, True]
+  assert [native for _kind, _original, native, _accepted, _stamp in calls] == [native_allowed, False, native_allowed, native_allowed]
+  assert all(stamp >= BASE + 160_000_000 for *_rest, stamp in calls)
+  assert updates == [None, 0]  # Final pending source/native refresh remains mandatory.
+  assert len(physical) == 1 and len(physical[0][0]) == 1
+  assert physical[0][0][0][0].control_ns == BASE + 150_000_000
+  assert physical[0][2] is native_allowed  # Acquisition does not create native permission.
+  assert fitter.poll.call_args.kwargs['token'] is None
+  assert not instances[0].ready  # Neither branch supplied a physical witness.
+  fitter.close.assert_called_once_with('learner_parent_exit')
diff --git a/openpilot/starpilot/flm/tests/test_paired_evidence_drain.py b/openpilot/starpilot/flm/tests/test_paired_evidence_drain.py
new file mode 100644
index 0000000000..8d36ab4349
--- /dev/null
+++ b/openpilot/starpilot/flm/tests/test_paired_evidence_drain.py
@@ -0,0 +1,301 @@
+"""Bounded actual wire/collector transport units; no road or learned-fit claim."""
+from collections import deque
+from dataclasses import replace
+from types import SimpleNamespace
+
+import pytest
+
+from openpilot.starpilot.flm.fullrate_collector import (
+  FullrateCollector, MAX_DRAIN, MAX_SNAPSHOT_WIRE_BYTES, MAX_WIRE_BYTES, drain, drain_pair,
+)
+from openpilot.starpilot.flm.learning_wire import read_applied, read_control
+from openpilot.starpilot.flm.tests.test_fullrate_pipeline import SyntheticAdapter, snapshots, serialized_control, serialized_output
+
+BASE = 10_000_000_000
+
+
+def transport(controls=(), outputs=()):
+  sockets = object(), object()
+  queues = {sockets[0]: deque(controls), sockets[1]: deque(outputs)}
+  reads = dict.fromkeys(sockets, 0)
+
+  def receive(socket):
+    reads[socket] += 1
+    return queues[socket].popleft() if queues[socket] else None
+
+  return sockets, receive, reads
+
+
+def exercise(controls, outputs, now, *, paired=True):
+  collector = FullrateCollector(SyntheticAdapter())
+  sockets, receive, reads = transport(controls, outputs)
+  seen = []
+
+  def control(event, current, stamp):
+    row = read_control(event)
+    accepted = current.control(row, stamp)
+    seen.append(('control', row.control_ns, accepted))
+
+  def output(event, current, stamp):
+    row = read_applied(event)
+    accepted = current.output(row, stamp)
+    seen.append(('output', row.apply_ns, accepted))
+
+  if paired:
+    result = drain_pair(sockets[0], control, sockets[1], output, receive, collector, now)
+  else:
+    result = drain(sockets[0], control, receive, collector, now)
+    result = drain(sockets[1], output, receive, collector, now) and result
+  return result, collector, seen, reads
+
+
+@pytest.mark.parametrize('offsets,expected', [((0, 150_000_000), 1), ((140_000_000, 150_000_000), 2), ((0, 10_000_000), 0)])
+def test_original_mixed_fresh_and_stale_rows_keep_same_rejections(offsets, expected):
+  pairs = [snapshots(BASE + offset)[:2] for offset in offsets]
+  controls = [serialized_control(c) for c, _ in pairs]
+  outputs = [serialized_output(o) for _, o in pairs]
+  result, collector, seen, _ = exercise(controls, outputs, BASE + 160_000_000)
+  old_result, old, old_seen, _ = exercise(controls, outputs, BASE + 160_000_000, paired=False)
+  assert result and old_result
+  assert len(collector.ready) == expected
+  assert sorted(seen) == sorted(old_seen)  # Neither stale original row was admitted.
+  assert collector.training_interruptions == old.training_interruptions
+  assert all(row.control_ns >= BASE + 60_000_000 for row, _ in collector.ready)
+  if offsets == (0, 150_000_000):
+    assert not old.ready  # The old output used to erase the fresh queued control.
+
+
+def small_event(stamp, *, size=8, kind='fixture', snapshot=(), ford_rows=(), ford_version=0):
+  return SimpleNamespace(logMonoTime=stamp, as_builder=lambda: SimpleNamespace(to_bytes=lambda: bytes(size)),
+    which=lambda: kind,
+    starpilotLateralState=SimpleNamespace(learningEvidence=SimpleNamespace(replaySnapshot=snapshot)),
+    starpilotCarState=SimpleNamespace(fordLearningEvidence=SimpleNamespace(version=ford_version, rows=ford_rows)))
+
+
+def ordered(controls, outputs):
+  collector = FullrateCollector(SyntheticAdapter())
+  sockets, receive, reads = transport(controls, outputs)
+  seen = []
+
+  def callback(kind):
+    return lambda event, _collector, now: seen.append((kind, event, now))
+
+  result = drain_pair(sockets[0], callback('control'), sockets[1], callback('output'), receive, collector, BASE)
+  return result, collector, seen, sockets, reads
+
+
+def test_equal_publisher_headers_choose_control_first_without_reordering_either_fifo():
+  controls = [small_event(BASE), small_event(BASE)]
+  outputs = [small_event(BASE), small_event(BASE)]
+  result, _, seen, _, _ = ordered(controls, outputs)
+  assert result
+  assert [(kind, event) for kind, event, _ in seen] == [('control', e) for e in controls] + [('output', e) for e in outputs]
+  assert all(now == BASE for _, _, now in seen)
+
+
+def test_only_queue_heads_compete_never_sort_a_producers_regression_away():
+  controls = [small_event(BASE + 30), small_event(BASE + 10)]
+  outputs = [small_event(BASE + 20), small_event(BASE + 40)]
+  result, _, seen, _, _ = ordered(controls, outputs)
+  assert result
+  assert [(kind, event.logMonoTime) for kind, event, _ in seen] == [
+    ('output', BASE + 20), ('control', BASE + 30), ('control', BASE + 10), ('output', BASE + 40)]
+
+
+def test_real_control_sequence_regression_still_rejects_and_clears():
+  controls = [serialized_control(snapshots(BASE + offset)[0]) for offset in (20_000_000, 10_000_000)]
+  result, collector, seen, _ = exercise(controls, [], BASE + 30_000_000)
+  assert result and [accepted for _, _, accepted in seen] == [True, False]
+  assert collector.reason == 'reordered_control' and not collector.controls and not collector.ready
+
+
+def test_original_publisher_header_not_repeated_consumed_command_clock_is_ordering_key():
+  control, output, _ = snapshots(BASE)
+  event = serialized_control(control).as_builder()
+  event.logMonoTime = BASE + 5_000_000  # Card can publish before Controls finishes its feedback.
+  applied = serialized_output(output).as_builder()
+  applied.logMonoTime = BASE + 2_000_000
+  result, _, seen, _, _ = ordered([event.as_reader()], [applied.as_reader()])
+  assert result and [kind for kind, _, _ in seen] == ['output', 'control']
+  # The transport never restamps or replaces the embedded original clocks.
+  assert read_control(seen[1][1]).control_ns == BASE
+  assert read_applied(seen[0][1]).apply_ns == output.apply_ns
+
+
+@pytest.mark.parametrize('absent', ('control', 'output', 'both'))
+def test_one_producer_absent_is_nonblocking_and_does_not_create_evidence(absent):
+  controls = [] if absent in ('control', 'both') else [small_event(BASE)]
+  outputs = [] if absent in ('output', 'both') else [small_event(BASE)]
+  result, collector, seen, sockets, reads = ordered(controls, outputs)
+  assert result and len(seen) == len(controls) + len(outputs)
+  assert reads[sockets[0]] == len(controls) + 1 and reads[sockets[1]] == len(outputs) + 1
+  assert not collector.ready and not collector.controls and not collector.applied
+
+
+@pytest.mark.parametrize('field,value', [('logMonoTime', 0), ('logMonoTime', -1), ('logMonoTime', True), ('logMonoTime', 1.0)])
+def test_invalid_ordering_header_is_malformed_not_a_freshness_substitute(field, value):
+  event = small_event(BASE)
+  setattr(event, field, value)
+  result, collector, _, _, _ = ordered([event], [])
+  assert not result and collector.reason == 'malformed_evidence'
+
+
+@pytest.mark.parametrize('side', (0, 1))
+@pytest.mark.parametrize('size,kind,snapshot,ford_rows,ford_version,allowed', [
+  (MAX_WIRE_BYTES, 'fixture', (), (), 0, True),
+  (MAX_WIRE_BYTES + 1, 'fixture', (), (), 0, False),
+  (MAX_WIRE_BYTES + 1, 'starpilotLateralState', (), (), 0, False),
+  (MAX_SNAPSHOT_WIRE_BYTES, 'starpilotLateralState', (1,), (), 0, True),
+  (MAX_SNAPSHOT_WIRE_BYTES + 1, 'starpilotLateralState', (1,), (), 0, False),
+  (MAX_WIRE_BYTES + 1, 'starpilotCarState', (), (1,), 1, True),
+  (MAX_WIRE_BYTES + 1, 'starpilotCarState', (), (1,), 0, False),
+  (MAX_WIRE_BYTES + 1, 'starpilotCarState', (), (1,) * 5, 1, False),
+])
+def test_exact_existing_wire_bounds_on_both_streams(side, size, kind, snapshot, ford_rows, ford_version, allowed):
+  rows = [[], []]
+  rows[side] = [small_event(BASE, size=size, kind=kind, snapshot=snapshot, ford_rows=ford_rows, ford_version=ford_version)]
+  result, collector, seen, _, _ = ordered(*rows)
+  assert result is allowed
+  assert len(seen) == int(allowed)
+  if not allowed:
+    assert collector.reason == 'malformed_evidence' and not collector.ready
+
+
+@pytest.mark.parametrize('side', (0, 1))
+@pytest.mark.parametrize('error', (AttributeError, KeyError, TypeError, ValueError, OverflowError))
+def test_malformed_callback_keeps_original_reset_boundary(side, error):
+  collector = FullrateCollector(SyntheticAdapter())
+  c, o, _ = snapshots(BASE)
+  collector.control(c, BASE + 1_000_000)
+  collector.output(o, BASE + 1_000_000)
+  assert collector.ready
+  rows = [[], []]
+  rows[side] = [small_event(BASE)]
+  sockets, receive, _ = transport(*rows)
+
+  def malformed(*_args):
+    raise error('declared malformed fixture')
+
+  assert not drain_pair(sockets[0], malformed, sockets[1], malformed, receive, collector, BASE)
+  assert collector.reason == 'malformed_evidence' and not collector.ready and not collector.blocks
+
+
+@pytest.mark.parametrize('side', (0, 1))
+@pytest.mark.parametrize('count', (MAX_DRAIN, MAX_DRAIN + 1, MAX_DRAIN + 2))
+def test_per_stream_bound_and_single_lookahead_are_not_silently_conflated(side, count):
+  rows = [[], []]
+  rows[side] = [small_event(BASE + i) for i in range(count)]
+  result, collector, seen, sockets, reads = ordered(*rows)
+  assert len(seen) == MAX_DRAIN
+  assert reads[sockets[side]] == MAX_DRAIN + 1
+  assert reads[sockets[1 - side]] == 1
+  assert result is (count == MAX_DRAIN)
+  assert collector.dropped == int(count > MAX_DRAIN)
+  if count > MAX_DRAIN:
+    assert collector.reason == 'fullrate_overrun' and not collector.ready
+
+
+def test_both_streams_have_independent_bounds():
+  controls = [small_event(BASE + i * 2) for i in range(MAX_DRAIN)]
+  outputs = [small_event(BASE + i * 2 + 1) for i in range(MAX_DRAIN)]
+  result, _, seen, sockets, reads = ordered(controls, outputs)
+  assert result and len(seen) == 2 * MAX_DRAIN
+  assert list(reads.values()) == [MAX_DRAIN + 1] * 2
+  assert [event.logMonoTime for _, event, _ in seen] == list(range(BASE, BASE + 2 * MAX_DRAIN))
+
+
+def test_output_native_denial_is_not_bypassed_by_fresh_cross_queue_join():
+  c, o, _ = snapshots(BASE)
+  result, collector, _, _ = exercise([serialized_control(c)], [serialized_output(replace(o, native_allowed=False))], BASE + 2_000_000)
+  assert result and not collector.ready and not collector.points
+
+
+def test_monotonic_publisher_headers_cannot_hide_regressed_control_after_an_intervening_join():
+  new_c, new_o, _ = snapshots(BASE + 20_000_000)
+  old_c, old_o, _ = snapshots(BASE + 10_000_000)
+
+  def header(event, stamp):
+    message = event.as_builder()
+    message.logMonoTime = stamp
+    return message.as_reader()
+
+  controls = [serialized_control(new_c), header(serialized_control(old_c), BASE + 30_000_000)]
+  outputs = [serialized_output(new_o), header(serialized_output(old_o), BASE + 31_000_000)]
+  result, collector, seen, _ = exercise(controls, outputs, BASE + 40_000_000)
+  assert result
+  assert seen[0][0] == 'control' and seen[1][0] == 'output'
+  assert seen[2] == ('control', old_c.control_ns, False)
+  assert collector.training_interruptions['reordered_control'] == 1
+  assert not collector.ready and old_c.control_ns not in collector.recent_controls
+
+
+@pytest.mark.parametrize('mutated', (False, True))
+def test_repeated_applied_header_and_duplicate_body_are_not_rewritten(mutated):
+  control, output, _ = snapshots(BASE)
+  duplicate = replace(output, applied=output.applied + 0.1) if mutated else output
+  result, collector, seen, _ = exercise([serialized_control(control)],
+    [serialized_output(output), serialized_output(duplicate)], BASE + 2_000_000)
+  assert result
+  assert seen[-1] == ('output', output.apply_ns, not mutated)
+  assert len(collector.ready) == int(not mutated)
+  if mutated:
+    assert collector.reason == 'changed_duplicate_output'
+
+
+def test_applied_before_feedback_can_join_without_rewriting_original_clocks():
+  collector = FullrateCollector(SyntheticAdapter())
+  seed = snapshots(BASE - 10_000_000)[0]
+  assert collector.control(seed, BASE + 10_000_000)
+  control, output, _ = snapshots(BASE)
+  feedback = serialized_control(control).as_builder()
+  feedback.logMonoTime = BASE + 5_000_000
+  sockets, receive, _ = transport([feedback.as_reader()], [serialized_output(output)])
+  order = []
+
+  def consume_control(event, current, stamp):
+    order.append('control')
+    assert current.control(read_control(event), stamp)
+
+  def consume_output(event, current, stamp):
+    order.append('output')
+    assert current.output(read_applied(event), stamp)
+    assert not current.ready
+
+  assert drain_pair(sockets[0], consume_control, sockets[1], consume_output, receive, collector, BASE + 10_000_000)
+  assert order == ['output', 'control']
+  assert list(collector.ready) == [(control, output)]
+
+
+@pytest.mark.parametrize('source_offset,header_offset,reason', [
+  (0, 1_000_000_000, 'stale_control'),
+  (200_000_000, 10_000_000, 'invalid_control'),
+])
+def test_publisher_ordering_header_cannot_replace_stale_or_future_original_source_clock(source_offset, header_offset, reason):
+  control = snapshots(BASE + source_offset)[0]
+  event = serialized_control(control).as_builder()
+  event.logMonoTime = BASE + header_offset
+  result, collector, seen, _ = exercise([event.as_reader()], [], BASE + 160_000_000)
+  assert result and seen == [('control', control.control_ns, False)]
+  assert collector.reason == reason and not collector.ready and not collector.controls
+
+
+@pytest.mark.parametrize('overrun_side', (0, 1))
+def test_mixed_queue_overrun_remains_whole_maneuver_reset(overrun_side):
+  collector = FullrateCollector(SyntheticAdapter())
+  c, o, _ = snapshots(BASE)
+  assert collector.control(c, BASE + 1_000_000)
+  assert collector.output(o, BASE + 1_000_000) and collector.ready
+  rows = [[small_event(BASE + 2 * i + side) for i in range(MAX_DRAIN + 1 if side == overrun_side else 8)] for side in (0, 1)]
+  sockets, receive, reads = transport(*rows)
+  counts = [0, 0]
+
+  def callback(side):
+    def consume(_event, _current, _now):
+      counts[side] += 1
+    return consume
+
+  assert not drain_pair(sockets[0], callback(0), sockets[1], callback(1), receive, collector, BASE)
+  assert counts[overrun_side] == MAX_DRAIN and counts[1 - overrun_side] == 8
+  assert reads[sockets[overrun_side]] == MAX_DRAIN + 1 and reads[sockets[1 - overrun_side]] == 9
+  assert collector.dropped == 1 and collector.reason == 'fullrate_overrun'
+  assert not collector.controls and not collector.applied and not collector.ready and not collector.points
diff --git a/openpilot/starpilot/flm/tests/test_paired_hkg_emission.py b/openpilot/starpilot/flm/tests/test_paired_hkg_emission.py
new file mode 100644
index 0000000000..864a43a62b
--- /dev/null
+++ b/openpilot/starpilot/flm/tests/test_paired_hkg_emission.py
@@ -0,0 +1,133 @@
+"""Registered Ioniq6 CAN encoding/decoding; declared state, no road/native claim."""
+from dataclasses import replace
+from types import SimpleNamespace
+
+import pytest
+from opendbc.car import structs
+from opendbc.car.car_helpers import interfaces
+from opendbc.car.hyundai.values import CAR
+from opendbc.car.steering_projection import TorqueProjectionObserver
+
+from openpilot.starpilot.flm.fullrate_collector import FullrateCollector, AppliedEvidence, drain_pair
+from openpilot.starpilot.flm.learning_wire import read_control, read_applied
+from openpilot.starpilot.flm.torque_emission_replay import EmissionHistory
+from openpilot.starpilot.flm.tests.test_fullrate_pipeline import snapshots, serialized_control, serialized_output
+from openpilot.starpilot.flm.tests.test_paired_evidence_drain import transport
+
+BASE = 10_000_000_000
+
+
+def original_rows():
+  factory = interfaces[CAR.HYUNDAI_IONIQ_6]
+  cp = factory.get_non_essential_params(CAR.HYUNDAI_IONIQ_6)
+  ci = factory(cp)
+  state = structs.CarState.new_message(canValid=True, canTimeout=False, vEgo=20., vEgoRaw=20.,
+    steeringAngleDeg=0., steeringRateDeg=0., steeringTorque=0., standstill=False)
+  state.cruiseState.enabled = state.cruiseState.available = True
+  state.cruiseState.speed = 20.
+  ci.CS.out = state.as_reader()
+  observer = TorqueProjectionObserver(cp)
+  ci.CC.learning_projection_observer = observer
+  command = structs.CarControl.new_message(enabled=True, latActive=True, longActive=False)
+  command.actuators.torque = 0.
+  rows = []
+  for index in range(17):
+    stamp = BASE + index * 10_000_000
+    control = replace(snapshots(stamp)[0], domain='normalized_torque', requested=0., baseline_prediction=0.)
+    actuators, packets = ci.apply(command.as_reader(), stamp + 1_000_000)
+    projection = observer.current
+    assert projection is not None and projection.steer_requested
+    assert any(msg[0] == projection.can_address and msg[1] == projection.can_data and msg[2] == projection.can_bus for msg in packets)
+    output = AppliedEvidence(stamp, stamp, stamp + 1_000_000, 'normalized_torque', float(actuators.torque),
+      True, True, False, False, False, True, projection, ci.CC.frame, stamp + 1_000_000)
+    rows.append((control, output))
+  return cp, rows
+
+
+def exercise(rows, cp, now):
+  history = EmissionHistory(cp, identity=rows[0][0].identity, baseline_revision=rows[0][0].baseline_revision)
+  adapter = SimpleNamespace(command_domain='normalized_torque', record_actual_carrier=history.record_carrier,
+    record_actual_emission=history.record)
+  collector = FullrateCollector(adapter)
+  sockets, receive, _ = transport([serialized_control(c) for c, _ in rows], [serialized_output(o) for _, o in rows])
+  accepted = []
+
+  def control(event, current, stamp):
+    accepted.append(('control', current.control(read_control(event), stamp)))
+
+  def output(event, current, stamp):
+    accepted.append(('output', current.output(read_applied(event), stamp)))
+
+  assert drain_pair(sockets[0], control, sockets[1], output, receive, collector, now)
+  for command, emission in collector.ready:
+    history.verify(command, emission)
+  return collector, accepted
+
+
+def test_mixed_age_original_hkg_frames_join_with_actual_can_decoder():
+  cp, rows = original_rows()
+  collector, accepted = exercise(rows, cp, BASE + 165_000_000)
+  assert len(collector.ready) == 10
+  assert sum(not good for kind, good in accepted if kind == 'control') == 7
+  assert sum(not good for kind, good in accepted if kind == 'output') == 7
+  assert collector.training_interruptions['stale_control'] == 7
+  assert collector.training_interruptions['stale_applied_output'] == 7
+
+
+@pytest.mark.parametrize('failure', ('gap', 'regression', 'future', 'denied'))
+def test_original_hkg_carrier_and_native_denials_still_invalidate(failure):
+  cp, rows = original_rows()
+  rows = rows[-3:]
+  c, o = rows[-1]
+  changes = {'gap': {'carrier_frame': o.carrier_frame + 1},
+             'regression': {'carrier_frame': o.carrier_frame - 2},
+             'future': {'carrier_observed_ns': BASE + 200_000_000},
+             'denied': {'native_allowed': False}}[failure]
+  rows[-1] = c, replace(o, **changes)
+  collector, accepted = exercise(rows, cp, BASE + 165_000_000)
+  assert not collector.ready
+  if failure == 'denied':
+    assert collector.reason == 'applied_native_witness_unavailable'
+  else:
+    assert accepted[-1] == ('output', False)
+    assert collector.reason == 'actual_carrier_gap_or_clock'
+
+
+@pytest.mark.parametrize('native_current', (True, False))
+def test_paired_collector_reaches_real_ingress_without_inventing_scores_or_trials(native_current):
+  from openpilot.starpilot.flm.canonical_surface_adapter import NAMES
+  from openpilot.starpilot.flm.live_tracking_ingress import LiveTrackingIngress
+  from openpilot.starpilot.flm.tests.test_tracking_source_context import context
+
+  cp, rows = original_rows()
+  first = rows[0][0]
+  # Explicit source-context unit seam. No physical pose, response witness,
+  # probe result or fitted episode is supplied by this transport test.
+  binding = replace(context().binding, identity=first.identity, source_epoch=first.source_epoch,
+    baseline_revision=first.baseline_revision, controller_revision=first.controller_revision,
+    native_identity=first.native_identity, native_required=first.native_required)
+  certificate = context(binding)
+  ingress = LiveTrackingIngress(certificate.binding, certificate, tuple((name, 0.) for name in NAMES), controller_domain='torque')
+  rows = [(replace(c, features=(1., 1.), applied_coordinates=(0., 0.),
+    stage_values=(('setpoint', .1),), replay_inputs=(('desired_curvature', .1 / 400.), ('vEgo', 20.),
+      ('roll', 0.), ('curvature_limited', 0.), ('control_frame', float(index)))), o) for index, (c, o) in enumerate(rows)]
+  history = EmissionHistory(cp, identity=first.identity, baseline_revision=first.baseline_revision)
+  adapter = SimpleNamespace(command_domain='normalized_torque', record_actual_carrier=history.record_carrier,
+    record_actual_emission=history.record)
+  collector = FullrateCollector(adapter)
+  collector.live_tracking = ingress
+  collector._learning_current_native_allowed = native_current  # Declared unit authority only.
+  sockets, receive, _ = transport([serialized_control(c) for c, _ in rows], [serialized_output(o) for _, o in rows])
+  now = BASE + 165_000_000
+  assert drain_pair(sockets[0], lambda e, c, n: c.control(read_control(e), n),
+    sockets[1], lambda e, c, n: c.output(read_applied(e), n), receive, collector, now)
+  for control, output in collector.ready:
+    history.verify(control, output)
+  expected = [c.control_ns for c, _ in rows[7:]] if native_current else []
+  assert list(ingress.episodes.ticks) == expected
+  assert [c.control_ns for c in ingress.control_history] == expected
+  assert ingress.episodes.emission_count == len(expected)
+  assert not ingress.episodes.endpoints and not ingress.episodes.completed and not ingress.episodes.ready
+  assert ingress.monitor.witness(now) is None
+  assert ingress.optimizer.offer(now) is None
+  assert not ingress.optimizer.before and not ingress.optimizer.future
diff --git a/openpilot/starpilot/flm/tests/test_parent_physical_pending.py b/openpilot/starpilot/flm/tests/test_parent_physical_pending.py
index 8a99d515e1..dd713ed7e1 100644
--- a/openpilot/starpilot/flm/tests/test_parent_physical_pending.py
+++ b/openpilot/starpilot/flm/tests/test_parent_physical_pending.py
@@ -3,6 +3,7 @@
 The SM/native world is declared synthetic. PoseCalibrator, observer, wire readers,
 control/output joins and the actual parent loop/collector are not replaced.
 """
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from dataclasses import replace
 from types import SimpleNamespace
 from unittest import mock
@@ -118,7 +119,7 @@ def exercise(*, failure=None, polls=86, pending=True):
   fitter.poll.return_value = None
   persistence = mock.Mock()
   with mock.patch.object(steerlearnerd, 'FullrateCollector', side_effect=collector_factory), \
-       mock.patch.object(steerlearnerd, 'drain', side_effect=drain), \
+       mock.patch.object(steerlearnerd, 'drain_pair', side_effect=separate_drain_fixture(drain)), \
        mock.patch('openpilot.starpilot.flm.native_health.current_native_health', side_effect=lambda *args: native[0]), \
        mock.patch('openpilot.starpilot.flm.runtime_adapter_registry.time.monotonic_ns', side_effect=lambda: clock[0]), \
        mock.patch('openpilot.starpilot.flm.runtime_adapter_registry.time.clock_gettime_ns', side_effect=boot_clock):
diff --git a/openpilot/starpilot/flm/tests/test_pending_native_delivery.py b/openpilot/starpilot/flm/tests/test_pending_native_delivery.py
index cf2164727d..1c1f730353 100644
--- a/openpilot/starpilot/flm/tests/test_pending_native_delivery.py
+++ b/openpilot/starpilot/flm/tests/test_pending_native_delivery.py
@@ -1,4 +1,5 @@
 """Actual daemon/native-parser boundary units; declared inputs, no road/fit proof."""
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from dataclasses import replace
 from types import SimpleNamespace
 
@@ -155,7 +156,7 @@ def exercise(monkeypatch, mode):
     return now[0]
 
   monkeypatch.setattr(steerlearnerd, 'FullrateCollector', factory)
-  monkeypatch.setattr(steerlearnerd, 'drain', drain)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', separate_drain_fixture(drain))
   monkeypatch.setattr(steerlearnerd._SourceDelivery, 'next', delivery_next)
   with pytest.raises(Finished):
     steerlearnerd.run(cp, sm, control_socket, output_socket, pm, lambda *args: Adapter(),
diff --git a/openpilot/starpilot/flm/tests/test_postdrain_sources.py b/openpilot/starpilot/flm/tests/test_postdrain_sources.py
index 56687e84b1..44e0de44f1 100644
--- a/openpilot/starpilot/flm/tests/test_postdrain_sources.py
+++ b/openpilot/starpilot/flm/tests/test_postdrain_sources.py
@@ -1,4 +1,5 @@
 """Actual daemon scheduling only; synthetic collector is not fit/native proof."""
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from types import SimpleNamespace as S
 import pytest
 from openpilot.starpilot.flm import steerlearnerd
@@ -46,7 +47,7 @@ def test_pose_observer_consumes_delivery_that_arrived_during_decoding(monkeypatc
     else:
       now[0] += 80_000_000
       sm.pending = True
-  monkeypatch.setattr(steerlearnerd, 'drain', drain)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', separate_drain_fixture(drain))
   actor=S(ingress=None, interrupt=lambda *a,**k: None, close=lambda *a: True)
   fitter=S(cancel=lambda *a: None, close=lambda *a: True)
   with pytest.raises(Stop):
diff --git a/openpilot/starpilot/flm/tests/test_receipt_callback_actual.py b/openpilot/starpilot/flm/tests/test_receipt_callback_actual.py
index be61bb45fc..d2385f3d59 100644
--- a/openpilot/starpilot/flm/tests/test_receipt_callback_actual.py
+++ b/openpilot/starpilot/flm/tests/test_receipt_callback_actual.py
@@ -1,4 +1,5 @@
 """Parent-boundary source renewal with genuine collector admission and blocks."""
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from dataclasses import replace
 from unittest.mock import Mock, patch
 
@@ -119,7 +120,7 @@ def test_parent_renews_actual_source_and_retains_real_collector_block(tmp_path,
 
   persistence = Mock()
   with patch.object(steerlearnerd, 'FullrateCollector', side_effect=factory), \
-       patch.object(steerlearnerd, 'drain', drain), \
+       patch.object(steerlearnerd, 'drain_pair', separate_drain_fixture(drain)), \
        patch('openpilot.starpilot.flm.native_health.current_native_health', native):
     with pytest.raises(StopIteration):
       steerlearnerd.run(cp, sm, control_socket, output_socket, Mock(), Mock(return_value=Adapter()), Mock(),
diff --git a/openpilot/starpilot/flm/tests/test_regular_publication_cadence.py b/openpilot/starpilot/flm/tests/test_regular_publication_cadence.py
index f5cd724554..f541006437 100644
--- a/openpilot/starpilot/flm/tests/test_regular_publication_cadence.py
+++ b/openpilot/starpilot/flm/tests/test_regular_publication_cadence.py
@@ -62,7 +62,7 @@ def _run_regular(monkeypatch, offsets, *, mode, delay_first_publication=0):
       now[0] += delay_first_publication
     return read(now[0])
 
-  monkeypatch.setattr(steerlearnerd, 'drain', lambda *args: None)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', lambda *args: None)
   with pytest.raises(LoopComplete):
     steerlearnerd.run(
       SimpleNamespace(),
diff --git a/openpilot/starpilot/flm/tests/test_source_publication_latency.py b/openpilot/starpilot/flm/tests/test_source_publication_latency.py
index 5cc0f48a01..5f73792663 100644
--- a/openpilot/starpilot/flm/tests/test_source_publication_latency.py
+++ b/openpilot/starpilot/flm/tests/test_source_publication_latency.py
@@ -50,7 +50,7 @@ def test_real_pose_status_precedes_optional_work_and_late_writes_use_current_clo
 
   monkeypatch.setattr(actor, 'tick', tick)
   monkeypatch.setattr(steerlearnerd, 'tracking_message', write_source)
-  monkeypatch.setattr(steerlearnerd, 'drain', lambda *args: None)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', lambda *args: None)
   fitter = NoLegacyFitter()
   with pytest.raises(LoopComplete):
     steerlearnerd.run(SimpleNamespace(), SM(), object(), object(), SimpleNamespace(send=send),
diff --git a/openpilot/starpilot/flm/tests/test_source_uncertainty.py b/openpilot/starpilot/flm/tests/test_source_uncertainty.py
index d3fe21b298..e13bf06f9b 100644
--- a/openpilot/starpilot/flm/tests/test_source_uncertainty.py
+++ b/openpilot/starpilot/flm/tests/test_source_uncertainty.py
@@ -1,4 +1,5 @@
 """Actual source reader/parent loop; synthetic transport, no fitted admission."""
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from dataclasses import replace
 from types import SimpleNamespace
 from typing import cast
@@ -134,7 +135,7 @@ def exercise(recovery=None, *, publication_uncertain=False, publication_only_rec
   persistence, publisher = mock.Mock(), mock.Mock()
   persistence.save_current.side_effect = lambda *args, **kwargs: saves.append(phase[0])
   publisher.send.side_effect = send
-  with mock.patch.object(steerlearnerd, 'drain', drain), \
+  with mock.patch.object(steerlearnerd, 'drain_pair', separate_drain_fixture(drain)), \
        mock.patch.object(steerlearnerd, 'FullrateCollector', side_effect=factory), \
        mock.patch('openpilot.starpilot.flm.native_health.current_native_health', return_value=True):
     with pytest.raises(StopIteration):
diff --git a/openpilot/starpilot/flm/tests/test_tracking_publication_receipt.py b/openpilot/starpilot/flm/tests/test_tracking_publication_receipt.py
index 8c2d752ab3..248ea0e9e3 100644
--- a/openpilot/starpilot/flm/tests/test_tracking_publication_receipt.py
+++ b/openpilot/starpilot/flm/tests/test_tracking_publication_receipt.py
@@ -49,7 +49,7 @@ def publication(monkeypatch, *, changes=None, read_cost=0, retained=None, origin
       raise LoopComplete
 
   sm = SimpleNamespace(updated={'deviceMotion': False}, logMonoTime={'deviceMotion': 0}, update=lambda *args: None)
-  monkeypatch.setattr(steerlearnerd, 'drain', lambda *args: None)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', lambda *args: None)
   with pytest.raises(LoopComplete):
     steerlearnerd.run(SimpleNamespace(), sm, object(), object(), SimpleNamespace(send=send),
       None, None, lambda stamp: old, None, None, None, None, fitter, None,
@@ -176,7 +176,7 @@ def test_actual_daemon_coalesces_refreshed_status_within_ordinary_receiver_frequ
       sent.append((now[0], message.as_reader()))
 
   monkeypatch.setattr(actor, 'tick', tick)
-  monkeypatch.setattr(steerlearnerd, 'drain', lambda *args: None)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', lambda *args: None)
   with pytest.raises(LoopComplete):
     steerlearnerd.run(SimpleNamespace(), SM(), object(), object(), SimpleNamespace(send=send),
       None, None, lambda stamp: cached[0], None, None, None, None, fitter, None,
diff --git a/openpilot/starpilot/flm/tests/test_training_report.py b/openpilot/starpilot/flm/tests/test_training_report.py
index acf42a0405..cfe7f09b6f 100644
--- a/openpilot/starpilot/flm/tests/test_training_report.py
+++ b/openpilot/starpilot/flm/tests/test_training_report.py
@@ -1,4 +1,5 @@
 """Actual optional writer/collector and typed application, never control authority."""
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 from copy import deepcopy
 from dataclasses import replace
 from types import SimpleNamespace
@@ -218,7 +219,7 @@ def test_actual_parent_publication_also_writes_training_snapshot(tmp_path):
   def drain(socket, callback, _receive, sink, now):
     callback(serialized_control(control) if socket is control_socket else serialized_output(applied), sink, now)
 
-  with mock.patch.object(steerlearnerd, 'drain', drain), \
+  with mock.patch.object(steerlearnerd, 'drain_pair', separate_drain_fixture(drain)), \
        mock.patch('openpilot.starpilot.flm.native_health.current_native_health', return_value=True):
     with pytest.raises(StopIteration):
       steerlearnerd.run(SimpleNamespace(), sm, control_socket, output_socket, publisher,
diff --git a/openpilot/starpilot/flm/tests/test_upstream_producer_transport.py b/openpilot/starpilot/flm/tests/test_upstream_producer_transport.py
index 41761aeb16..aace896a31 100644
--- a/openpilot/starpilot/flm/tests/test_upstream_producer_transport.py
+++ b/openpilot/starpilot/flm/tests/test_upstream_producer_transport.py
@@ -1,4 +1,5 @@
 """Installed source closure only; no kernel or model-output equivalence proof."""
+from openpilot.starpilot.flm.tests._drain_fixture import separate_drain_fixture
 import hashlib
 from pathlib import Path
 
@@ -139,7 +140,7 @@ def test_actual_daemon_reports_startup_source_refusal_without_issuing_a_trial(mo
       identity='f'*64, generation=1, domain='normalized_torque', stage_revision='stage', response_revision='response',
       controller_mode='starpilot', policy='declared-unit-policy', baseline_revision='baseline', controller_revision='controller', turn_assist=False)
 
-  monkeypatch.setattr(steerlearnerd, 'drain', drain)
+  monkeypatch.setattr(steerlearnerd, 'drain_pair', separate_drain_fixture(drain))
   monkeypatch.setattr(steerlearnerd, 'read_control', read_control)
   with pytest.raises(LoopComplete):
     steerlearnerd.run(SimpleNamespace(), SM(), control_socket, output_socket,
diff --git a/prebuilt.json b/prebuilt.json
index 7091d19448..fbaf16b2c1 100644
--- a/prebuilt.json
+++ b/prebuilt.json
@@ -1 +1 @@
-{"protected_inventory_sha256": "88b9b51a1975ea2c9475fbcbc7c9c54800f64a18e0fd6e421c55f2c073048c90", "schema_version": 1}
+{"protected_inventory_sha256": "c0f6a6c1510c120277f18045a93a8e928616844d6c574a8096ad7ca09a254fe2", "schema_version": 1}