From ab30a5992a4c1007e64d1ec26d6cc7952d235b20 Mon Sep 17 00:00:00 2001
From: firestar5683 <168790843+firestar5683@users.noreply.github.com>
Date: Sun, 18 Oct 2026 14:11:45 -0500
Subject: [PATCH] Fix FLM training collection

---
 openpilot/starpilot/flm/event_collector.py    |  73 ++++++-
 openpilot/starpilot/flm/event_raw_fit.py      |   3 +-
 openpilot/starpilot/flm/event_restore_fit.py  |   4 +-
 openpilot/starpilot/flm/fullrate_collector.py |  14 +-
 .../test_consumption_history_retention.py     |  99 ++++++++++
 .../flm/tests/test_consumption_join_clocks.py |  92 +++++++++
 .../flm/tests/test_event_pending_storage.py   | 180 ++++++++++++++++++
 openpilot/starpilot/flm/torque_eligibility.py |  20 +-
 8 files changed, 476 insertions(+), 9 deletions(-)
 create mode 100644 openpilot/starpilot/flm/tests/test_consumption_history_retention.py
 create mode 100644 openpilot/starpilot/flm/tests/test_consumption_join_clocks.py
 create mode 100644 openpilot/starpilot/flm/tests/test_event_pending_storage.py

diff --git a/openpilot/starpilot/flm/event_collector.py b/openpilot/starpilot/flm/event_collector.py
index 14cf523479..f79c22da01 100644
--- a/openpilot/starpilot/flm/event_collector.py
+++ b/openpilot/starpilot/flm/event_collector.py
@@ -8,9 +8,73 @@ from openpilot.starpilot.flm.adaptive_profile import replay_accepts
 class EventFullrateCollector(FullrateCollector):
   def __init__(self, *args, **kwargs):
     super().__init__(*args, **kwargs)
-    # At most5,120 joined endpoints; whole events are selected before the
+    # At most5,120 sealed endpoints plus80 current and39 pending;
+    # whole events are selected before the
     # child12,000-frame/24-event export budget, never by arbitrary packet tail.
     self.blocks = deque(self.blocks, maxlen=64)
+    # At most39 already-qualified endpoints awaiting a storage packet. These
+    # never retain an active join, physical freshness or independent episode.
+    self._pending_storage = ()
+    self._pending_storage_lineage = None
+    self._storage_lineage = None
+
+  def _discard_storage(self, reason):
+    self._pending_storage = ()
+    self._pending_storage_lineage = None
+    self._storage_lineage = None
+    super().split(reason, reset=True)
+
+  def _seal_raw(self):
+    if len(self._pending_storage) > 39 or len(self.raw_points) > 80:
+      self._discard_storage('event_storage_pending_bound')
+      return
+    lineage = (self.context, self.physical_context, getattr(self.adapter, 'actual_target_sha256', None))
+    if self.context is None or self.physical_context is None:
+      self._pending_storage = ()
+      self._pending_storage_lineage = None
+      return
+    if self._pending_storage_lineage != lineage:
+      # A new source/physical/accepted-target boundary cannot carry old rows.
+      self._pending_storage = ()
+    rows = self._pending_storage + tuple(self.raw_points)
+    previous, seen = None, set()
+    if self._storage_lineage == lineage and self.blocks and self.blocks[-1].samples:
+      tail = self.blocks[-1].samples[-1]
+      previous = tail.control_ns, tail.apply_ns, tail.mono_ns
+      seen.add(previous)
+    for point in rows:
+      clocks = point.control_ns, point.apply_ns, point.mono_ns
+      if (any(type(stamp) is not int or stamp <= 0 for stamp in clocks) or
+          clocks in seen or previous is not None and
+          (clocks[0] < previous[0] or clocks[1] < previous[1] or clocks[2] <= previous[2]) or
+          point.physical_context != self.physical_context):
+        self._discard_storage('event_storage_lineage_or_chronology')
+        return
+      seen.add(clocks)
+      previous = clocks
+    self._pending_storage = ()
+    self._pending_storage_lineage = None
+    self.raw_points.clear()
+    while len(rows) >= 40:
+      count = min(80, len(rows))
+      self.raw_points.extend(rows[:count])
+      # Original packet writer preserves each RawPoint and its original clocks.
+      super()._seal_raw()
+      rows = rows[count:]
+    self._pending_storage = rows  # Immutable remainder, strictly fewer than40.
+    self._pending_storage_lineage = lineage if rows else None
+    self._storage_lineage = lineage
+
+  def split(self, reason, *, reset=False):
+    if reset:
+      self._pending_storage = ()
+      self._pending_storage_lineage = None
+      self._storage_lineage = None
+    else:
+      self._seal_raw()
+    # Original split still clears ready/current rows/active/physical freshness.
+    # Storage is not episode independence: child complete windows decide support.
+    super().split(reason, reset=reset)
 
   def physical(self, value, now_ns):
     accepted = super().physical(value, now_ns)
@@ -49,6 +113,9 @@ class EventFullrateCollector(FullrateCollector):
     self.ready.clear()
     self.points.clear()
     self.raw_points.clear()
+    self._pending_storage = ()
+    self._pending_storage_lineage = None
+    self._storage_lineage = None
     self.blocks.clear()
     self.cohort = None
     self.sealed = False
@@ -84,6 +151,10 @@ class EventFullrateCollector(FullrateCollector):
       return None
     control, output = self.ready[-1]
     self.sequence += 1
+    # None is a completed selection attempt too. Reconstructing the same whole
+    # source history on every20Hz parent loop cannot create independent support.
+    # Retain the existing60s cadence for all expensive snapshot attempts.
+    self.last_fit = now_ns
     try:
       return self.adapter.fit_job_snapshot(tuple(self.blocks), now_ns=now_ns, sequence=self.sequence,
         prior=self.prior_coefficients, control=control, output=output, physical_context=self.physical_context)
diff --git a/openpilot/starpilot/flm/event_raw_fit.py b/openpilot/starpilot/flm/event_raw_fit.py
index b741b53611..8a14853aa0 100644
--- a/openpilot/starpilot/flm/event_raw_fit.py
+++ b/openpilot/starpilot/flm/event_raw_fit.py
@@ -10,7 +10,7 @@ from openpilot.starpilot.flm.event_physical import fit_event_physical
 from openpilot.starpilot.flm.event_windows import regroup_whole_events
 from openpilot.starpilot.flm.physical_response import PhysicalRow
 from openpilot.starpilot.flm.runtime_adapter_registry import canonical_physical_context, physical_model_revision
-from openpilot.starpilot.flm.torque_eligibility import exact_unclipped_pair
+from openpilot.starpilot.flm.torque_eligibility import exact_unclipped_pair, consumed_sources_current
 from openpilot.starpilot.flm.torque_emission_replay import delivered
 from openpilot.starpilot.flm.torque_raw_fit import control_from_dict, output_from_dict
 
@@ -63,6 +63,7 @@ def original_event_evidence(payload, engine, objective):
           command.limited and not exact_unclipped_pair(command, output) or command.model_wobble or
           command.controller_chatter and not retain_chatter or any(command.applied_coordinates) or
           not output.valid or not output.active or output.override or output.fault or output.limited or not output.native_allowed or
+          not consumed_sources_current(command, output) or
           output.control_ns != command.control_ns or point['control_ns'] != command.control_ns or point['apply_ns'] != output.apply_ns or
           canonical_physical_context(point['physical_context']) != context or
           not engine.applied_matches(command, output) or delivered(output) != point['physical']['applied_command']):
diff --git a/openpilot/starpilot/flm/event_restore_fit.py b/openpilot/starpilot/flm/event_restore_fit.py
index eebded1336..3e792d32a3 100644
--- a/openpilot/starpilot/flm/event_restore_fit.py
+++ b/openpilot/starpilot/flm/event_restore_fit.py
@@ -16,7 +16,7 @@ from openpilot.starpilot.flm.event_windows import regroup_whole_events
 from openpilot.starpilot.flm.full_surface_profile import RESTORE_POLICY, envelope_from_dict
 from openpilot.starpilot.flm.physical_response import PhysicalRow
 from openpilot.starpilot.flm.runtime_adapter_registry import canonical_physical_context, physical_model_revision
-from openpilot.starpilot.flm.torque_eligibility import exact_unclipped_pair
+from openpilot.starpilot.flm.torque_eligibility import exact_unclipped_pair, consumed_sources_current
 from openpilot.starpilot.flm.torque_emission_replay import delivered
 from openpilot.starpilot.flm.torque_raw_fit import control_from_dict, output_from_dict
 
@@ -45,7 +45,7 @@ def _restore_original_evidence(payload, engine, objective):
           command.baseline_revision != engine.baseline_revision or command.source_epoch != envelope.source_epoch or
           not 0 < command.source_receipt_ns <= command.control_ns <= command.source_receipt_ns + 500_000_000 or
           not 0 < command.state_ns <= command.control_ns <= command.state_ns + 30_000_000 or
-          not command.state_ns <= output.state_ns <= output.apply_ns or output.state_ns - command.state_ns > 30_000_000 or
+          not consumed_sources_current(command, output) or
           not command.control_ns <= output.apply_ns <= command.control_ns + 100_000_000 or
           command.native_required and (not command.native_allowed or not command.native_identity or
             not 0 < command.native_observed_ns <= command.control_ns <= output.apply_ns <= command.native_expires_ns or
diff --git a/openpilot/starpilot/flm/fullrate_collector.py b/openpilot/starpilot/flm/fullrate_collector.py
index 3a73c48e80..a2c9f9b949 100644
--- a/openpilot/starpilot/flm/fullrate_collector.py
+++ b/openpilot/starpilot/flm/fullrate_collector.py
@@ -6,7 +6,7 @@ controller selection is inferred here. Domain adapters qualify response inverse.
 """
 
 from collections import Counter, OrderedDict, deque
-from openpilot.starpilot.flm.torque_eligibility import asynchronous_hint, bounded_context_output, exact_unclipped_pair
+from openpilot.starpilot.flm.torque_eligibility import asynchronous_hint, bounded_context_output, exact_unclipped_pair, consumed_sources_current
 
 from dataclasses import dataclass, replace
 import math
@@ -408,11 +408,17 @@ class FullrateCollector:
         self._join_emission(command, output)
 
   def _join_emission(self, command, output):
-    if (
-      output.domain != self.adapter.command_domain or not command.state_ns <= output.state_ns <= output.apply_ns or output.state_ns - command.state_ns > JOIN_NS
-    ):
+    if output.domain != self.adapter.command_domain:
       self.split('consumed_source_mismatch', reset=True)
       return
+    current_sources = (consumed_sources_current(command, output) if command.domain == 'normalized_torque' else
+                       command.state_ns <= output.state_ns <= output.apply_ns and output.state_ns - command.state_ns <= JOIN_NS)
+    if not current_sources:
+      # This pair is unusable, but it does not change the identity/provenance
+      # of independently sealed historical samples. Withdraw current admission
+      # and discard the partial window; real context changes still hard-reset.
+      self.split('consumed_source_mismatch', reset=command.domain != 'normalized_torque')
+      return
     if command.domain == 'normalized_torque':
       projection = output.projection
       if (
diff --git a/openpilot/starpilot/flm/tests/test_consumption_history_retention.py b/openpilot/starpilot/flm/tests/test_consumption_history_retention.py
new file mode 100644
index 0000000000..002b93a150
--- /dev/null
+++ b/openpilot/starpilot/flm/tests/test_consumption_history_retention.py
@@ -0,0 +1,99 @@
+"""Collector split-policy regressions; synthetic physical fixture, no fit claim.
+
+History is sealed through the real collector methods using the existing explicit
+synthetic transport fixture. Direct _join_emission calls isolate reset policy;
+these tests do not claim registered controller/native/physical qualification.
+"""
+from dataclasses import replace
+
+import pytest
+
+from openpilot.starpilot.flm.adaptive_profile import Result
+from openpilot.starpilot.flm.fullrate_collector import FullrateCollector
+from openpilot.starpilot.flm.tests.test_fullrate_pipeline import BASE, SyntheticAdapter, snapshots
+
+
+def history():
+  collector = FullrateCollector(SyntheticAdapter())
+  stamp = BASE
+  for _ in range(281):
+    stamp += 50_000_000
+    control, output, physical = snapshots(stamp)
+    assert collector.control(control, output.apply_ns)
+    assert collector.output(output, output.apply_ns)
+    collector.physical(physical, output.apply_ns)
+  assert len(collector.blocks) == 1
+  assert collector.ready and collector.last_qualified_response
+  assert collector.fit_result is None  # Never manufacture a fitted result.
+  collector.result = Result('unavailable', 'synthetic test state', None)
+  return collector, control, output
+
+
+def withdrawn(collector, now_ns):
+  assert not collector.controls and not collector.applied
+  assert not collector.recent_controls and not collector.recent_applied
+  assert not collector.ready and not collector.points and not collector.raw_points
+  assert collector.active_since == collector.last_apply == collector.last_response == 0
+  assert collector.last_qualified_response == 0
+  assert collector.result is None and collector.last_validation is None
+  assert collector.revalidate_fit(now_ns) is None
+
+
+@pytest.mark.parametrize('failure', ('card_source', 'dispatch', 'source_order'))
+def test_normalized_clock_rejection_preserves_exact_sealed_history_only(failure):
+  collector, control, output = history()
+  saved = tuple(collector.blocks)
+  samples = tuple(saved[0].samples)
+  physical_context = collector.physical_context
+  # The direct call tests only normalized join-reset dispatch, before any
+  # encoder/native branch can run; the invalid pair must never be admitted.
+  collector.adapter.command_domain = 'normalized_torque'
+  control = replace(control, domain='normalized_torque')
+  output = replace(output, domain='normalized_torque')
+  if failure == 'card_source':
+    output = replace(output, state_ns=control.state_ns, apply_ns=control.control_ns + 30_000_000)
+  elif failure == 'dispatch':
+    output = replace(output, state_ns=control.control_ns + 30_000_001,
+                     apply_ns=control.control_ns + 30_000_001)
+  else:
+    output = replace(output, state_ns=control.state_ns - 1)
+  collector._join_emission(control, output)
+  assert collector.training_interruptions['consumed_source_mismatch'] == 1
+  assert tuple(collector.blocks) == saved
+  assert collector.blocks[0] is saved[0]
+  assert all(a is b for a, b in zip(collector.blocks[0].samples, samples, strict=True))
+  assert collector.physical_context == physical_context
+  withdrawn(collector, output.apply_ns)
+
+
+def test_wrong_output_domain_hard_resets_sealed_history():
+  collector, control, output = history()
+  collector._join_emission(control, replace(output, domain='wrong-domain'))
+  assert collector.training_interruptions['consumed_source_mismatch'] == 1
+  assert not collector.blocks and collector.physical_context is None
+  assert collector.fit_result is collector.fit_validation is None
+  withdrawn(collector, output.apply_ns)
+
+
+@pytest.mark.parametrize('field,value', (('identity', '4' * 64), ('source_epoch', '5' * 64),
+                                      ('baseline_revision', '6' * 64)))
+def test_actual_control_context_change_still_hard_resets_history(field, value):
+  collector, control, output = history()
+  later, _, _ = snapshots(control.control_ns + 50_000_000)
+  changed = replace(later, **{field: value})
+  assert collector.control(changed, changed.control_ns)
+  assert collector.training_interruptions['actual_controller_identity_changed'] == 2
+  assert not collector.blocks and collector.physical_context is None
+  assert collector.fit_result is collector.fit_validation is None
+  assert collector.result is None and collector.last_validation is None
+
+
+def test_non_normalized_clock_mismatch_keeps_original_hard_reset_behavior():
+  collector, control, output = history()
+  assert control.domain != 'normalized_torque'
+  stale = replace(output, state_ns=control.state_ns + 30_000_001,
+                  apply_ns=control.state_ns + 30_000_002)
+  collector._join_emission(control, stale)
+  assert collector.training_interruptions['consumed_source_mismatch'] == 1
+  assert not collector.blocks and collector.physical_context is None
+  withdrawn(collector, stale.apply_ns)
diff --git a/openpilot/starpilot/flm/tests/test_consumption_join_clocks.py b/openpilot/starpilot/flm/tests/test_consumption_join_clocks.py
new file mode 100644
index 0000000000..1ec74fabda
--- /dev/null
+++ b/openpilot/starpilot/flm/tests/test_consumption_join_clocks.py
@@ -0,0 +1,92 @@
+"""Recorded route7c clocks and strict independent consumption boundaries.
+
+No supplied physical response, native permission or fit is created here.
+"""
+from types import SimpleNamespace
+
+import pytest
+from openpilot.starpilot.flm.torque_eligibility import consumed_sources_current
+
+
+def pair():
+  return (SimpleNamespace(state_ns=1681782046348, control_ns=1681789927746),
+          SimpleNamespace(state_ns=1681812490090, apply_ns=1681814665097))
+
+
+def test_recorded_route7c_individually_fresh_consumption_not_cross_source_age():
+  control, output = pair()
+  assert output.state_ns - control.state_ns == 30_443_742
+  assert control.control_ns - control.state_ns == 7_881_398
+  assert output.apply_ns - output.state_ns == 2_175_007
+  assert output.apply_ns - control.control_ns == 24_737_351
+  assert consumed_sources_current(control, output)
+
+
+@pytest.mark.parametrize('boundary', ('control_state', 'card_state', 'dispatch'))
+@pytest.mark.parametrize('excess', (0, 1))
+def test_each_original_thirty_ms_budget_is_inclusive_but_never_extended(boundary, excess):
+  end = 10_100_000_000
+  control = SimpleNamespace(state_ns=end - 30_000_000, control_ns=end)
+  output = SimpleNamespace(state_ns=end, apply_ns=end)
+  if boundary == 'control_state':
+    control.state_ns -= excess
+  elif boundary == 'card_state':
+    output.apply_ns = end + 30_000_000 + excess
+  else:
+    output.state_ns = output.apply_ns = end + 30_000_000 + excess
+  assert consumed_sources_current(control, output) is (excess == 0)
+
+
+@pytest.mark.parametrize('field,value', (('state_ns', 0), ('state_ns', -1), ('state_ns', True),
+                                      ('control_ns', 1.0), ('control_ns', 0)))
+def test_invalid_control_clocks_fail_closed(field, value):
+  control, output = pair()
+  setattr(control, field, value)
+  assert not consumed_sources_current(control, output)
+
+
+def test_card_state_precedes_control_source_or_follows_application_fails():
+  control, output = pair()
+  output.state_ns = control.state_ns - 1
+  assert not consumed_sources_current(control, output)
+  output.state_ns = output.apply_ns + 1
+  assert not consumed_sources_current(control, output)
+
+
+def test_recorded_clock_relationship_reaches_actual_registered_encoder_join(tmp_path):
+  from dataclasses import replace
+  from opendbc.car.toyota.values import CAR
+  from openpilot.starpilot.flm.fullrate_collector import FullrateCollector
+  from openpilot.starpilot.flm.runtime_adapter_registry import read_source
+  from openpilot.starpilot.flm.torque_emission_replay import EmissionHistory
+  from openpilot.starpilot.flm.tests.test_torque_emission_join import BASE, interface, settle, cc, emitted
+
+  ci = interface(CAR.TOYOTA_COROLLA_TSS2)
+  settle(ci, cc(0.01))
+  dispatch = 24_737_351
+  apply_ns = BASE + ci.CC.frame * 10_000_000
+  original_control_ns = apply_ns - dispatch
+  control, output = emitted(ci, cc(0.01), tmp_path, control_ns=original_control_ns)
+  assert control.control_ns == original_control_ns and output.apply_ns == apply_ns
+  # emitted/source constructs the source at the original command clock. Its
+  # compact encoder fixture omits receipt metadata; obtain the actual typed
+  # source with the identical fixture settings rather than inventing a stamp.
+  actual_source = read_source(ci.CP.as_reader(),
+    SimpleNamespace(get_param_path=lambda key: str(tmp_path / key)),
+    now_ns=original_control_ns, calibration=(0.0, 0.0, 0.0), vehicle_session='fixture-drive')
+  assert actual_source.source_epoch == control.source_epoch
+  control.source_receipt_ns = actual_source.observed_ns
+  assert 0 < control.source_receipt_ns <= control.control_ns <= actual_source.expires_ns
+  assert control.control_ns <= control.source_receipt_ns + 500_000_000
+  control.state_ns = control.control_ns - 7_881_398
+  control.domain = 'normalized_torque'
+  output = replace(output, control_ns=control.control_ns, state_ns=output.apply_ns - 2_175_007)
+  history = EmissionHistory(ci.CP, identity=control.identity, baseline_revision=control.baseline_revision)
+  assert history.record_carrier(output)
+  collector = FullrateCollector(SimpleNamespace(command_domain='normalized_torque', record_actual_emission=history.record))
+  collector._join_emission(control, output)
+  assert list(collector.ready) == [(control, output)]
+  history.verify(control, output)
+  collector._join_emission(control, replace(output, state_ns=output.apply_ns - 30_000_001))
+  assert not collector.ready
+  assert collector.training_interruptions['consumed_source_mismatch'] == 1
diff --git a/openpilot/starpilot/flm/tests/test_event_pending_storage.py b/openpilot/starpilot/flm/tests/test_event_pending_storage.py
new file mode 100644
index 0000000000..c910cab2c6
--- /dev/null
+++ b/openpilot/starpilot/flm/tests/test_event_pending_storage.py
@@ -0,0 +1,180 @@
+"""Storage-only synthetic row identity, not physical/native qualification."""
+import pytest
+from openpilot.starpilot.flm.event_collector import EventFullrateCollector
+from openpilot.starpilot.flm.fullrate_collector import RawPoint
+from openpilot.starpilot.flm.tests.test_fullrate_pipeline import SyntheticAdapter, snapshots, BASE
+
+
+def setup():
+  collector = EventFullrateCollector(SyntheticAdapter())
+  collector.context = ('source',)
+  collector.physical_context = ('physical',)
+  collector.cohort = 'event-storage.source'
+  control, output, _ = snapshots(BASE)
+  def rows(start, count):
+    return tuple(RawPoint(control.control_ns, output.apply_ns, BASE + start + i,
+      control, output, {'original': start + i}, ('physical',)) for i in range(count))
+  return collector, rows
+
+
+@pytest.mark.parametrize('first,second', ((1,39),(20,20),(39,1),(39,40),(39,80)))
+def test_original_fragments_survive_only_as_storage_and_reconstruct_exact_rows(first, second):
+  collector, rows = setup()
+  a, b = rows(0, first), rows(first, second)
+  collector.raw_points.extend(a)
+  collector.active_since = BASE
+  collector.last_apply = BASE + 1
+  collector.split('physical-interruption')
+  assert collector._pending_storage == a and not collector.blocks
+  assert collector.active_since == collector.last_apply == 0 and not collector.ready
+  collector.raw_points.extend(b)
+  collector.split('another-interruption')
+  saved = tuple(point for block in collector.blocks for point in block.samples)
+  actual = saved + collector._pending_storage
+  assert actual == a + b
+  assert all(x is y for x, y in zip(actual, a + b, strict=True))
+  assert len(collector._pending_storage) < 40
+  assert all(40 <= len(block.samples) <= 80 for block in collector.blocks)
+  before = actual
+  collector.split('empty-interruption')
+  assert tuple(point for block in collector.blocks for point in block.samples) + collector._pending_storage == before
+
+
+@pytest.mark.parametrize('count', (1, 39, 40, 79, 80))
+def test_hard_reset_discards_pending_current_and_sealed_history(count):
+  collector, rows = setup()
+  collector.raw_points.extend(rows(0, 39))
+  collector.split('ordinary-break')
+  collector.raw_points.extend(rows(39, count))
+  collector.split('identity-boundary', reset=True)
+  assert not collector._pending_storage and not collector.raw_points and not collector.blocks
+  assert collector.last_qualified_response == 0 and not collector.ready
+
+
+def test_accepted_target_rollover_drops_pending_and_current_storage():
+  collector, rows = setup()
+  collector.raw_points.extend(rows(0, 39))
+  collector.split('ordinary-break')
+  collector.raw_points.extend(rows(39, 20))
+  collector._rollover_training()
+  assert not collector._pending_storage and not collector.raw_points and not collector.blocks
+  assert collector.active_since == 0 and collector.last_qualified_response == 0
+
+
+def test_empty_context_cannot_park_unqualified_rows_and_packet_history_stays_bounded():
+  collector, rows = setup()
+  collector.context = None
+  collector.raw_points.extend(rows(0, 39))
+  collector.split('no-context')
+  assert not collector._pending_storage and not collector.raw_points
+  collector.context = ('source',)
+  for i in range(70):
+    collector.raw_points.extend(rows(i * 80, 80))
+    collector.split('ordinary-break')
+  assert len(collector.blocks) == 64 and not collector._pending_storage
+
+
+@pytest.mark.parametrize('boundary', ('source', 'physical', 'target'))
+def test_pending_never_crosses_source_physical_or_accepted_target_change(boundary):
+  collector, rows = setup()
+  old = rows(0, 39)
+  collector.raw_points.extend(old)
+  collector.split('ordinary-break')
+  assert collector._pending_storage == old
+  if boundary == 'source':
+    collector.context = ('new-source',)
+  elif boundary == 'physical':
+    collector.physical_context = ('new-physical',)
+  else:
+    collector.adapter.actual_target_sha256 = 'new-accepted-target'
+  from dataclasses import replace
+  new = replace(rows(39, 1)[0], physical_context=collector.physical_context)
+  collector.raw_points.append(new)
+  collector.split('new-boundary')
+  assert not collector.blocks
+  assert collector._pending_storage == (new,)
+  assert collector._pending_storage[0] is new
+
+
+@pytest.mark.parametrize('fault', ('duplicate', 'backward-control', 'backward-apply', 'backward-response', 'foreign-physical'))
+def test_duplicate_backward_and_foreign_rows_fail_closed_without_export(fault):
+  from dataclasses import replace
+  collector, rows = setup()
+  first = rows(0, 39)
+  collector.raw_points.extend(first)
+  collector.split('ordinary-break')
+  new = rows(39, 1)[0]
+  if fault == 'duplicate':
+    new = first[-1]
+  elif fault == 'backward-control':
+    new = replace(new, control_ns=new.control_ns - 1)
+  elif fault == 'backward-apply':
+    new = replace(new, apply_ns=new.apply_ns - 1)
+  elif fault == 'backward-response':
+    new = replace(new, mono_ns=first[-1].mono_ns - 1)
+  else:
+    new = replace(new, physical_context=('foreign',))
+  collector.raw_points.append(new)
+  collector.split('malformed-fragment')
+  assert collector.training_interruptions['event_storage_lineage_or_chronology'] == 1
+  assert not collector._pending_storage and not collector.blocks
+
+
+def test_thirty_nine_unexportable_rows_never_become_a_packet_by_themselves():
+  collector, rows = setup()
+  original = rows(0, 39)
+  collector.raw_points.extend(original)
+  for _ in range(20):
+    collector.split('ordinary-break')
+    assert not collector.blocks
+    assert collector._pending_storage == original
+
+
+
+def test_late_row_cannot_precede_previously_sealed_packet_when_pending_is_empty():
+  collector, rows = setup()
+  collector.raw_points.extend(rows(0, 40))
+  collector.split('first-packet')
+  assert len(collector.blocks) == 1 and not collector._pending_storage
+  collector.raw_points.append(rows(39, 1)[0])
+  collector.split('late-duplicate')
+  assert not collector.blocks and not collector._pending_storage
+  assert collector.training_interruptions['event_storage_lineage_or_chronology'] == 1
+
+
+
+def snapshot_ready_collector(result):
+  from unittest.mock import Mock
+  collector, rows = setup()
+  for i in range(8):
+    collector.raw_points.extend(rows(i * 40, 40))
+    collector.split('storage-boundary')
+  control, output, _ = snapshots(BASE)
+  collector.ready.append((control, output))
+  collector.adapter.fit_job_snapshot = Mock(return_value=result)
+  return collector
+
+
+def test_none_snapshot_selection_is_bounded_by_original_sixty_second_cadence():
+  collector = snapshot_ready_collector(None)
+  assert collector.job_snapshot(BASE) is None
+  for elapsed in (0, 1, 50_000_000, 59_999_999_999):
+    assert collector.job_snapshot(BASE + elapsed) is None
+  assert collector.adapter.fit_job_snapshot.call_count == 1
+  assert collector.job_snapshot(BASE + 60_000_000_000) is None
+  assert collector.adapter.fit_job_snapshot.call_count == 2
+  assert collector.sequence == 2
+  assert collector.last_fit == BASE + 60_000_000_000
+  assert collector.fit_block_sequence == 0  # No actual submission fabricated.
+
+
+def test_snapshot_result_still_requires_original_explicit_submission_marker():
+  sentinel = {'storage_test_only': True}
+  collector = snapshot_ready_collector(sentinel)
+  assert collector.job_snapshot(BASE) is sentinel
+  assert collector.fit_block_sequence == 0
+  collector.mark_job_submitted(BASE)
+  assert collector.fit_block_sequence == collector.block_sequence
+  assert collector.last_fit == BASE
+  assert collector.job_snapshot(BASE + 60_000_000_000) is None
+  assert collector.adapter.fit_job_snapshot.call_count == 1
diff --git a/openpilot/starpilot/flm/torque_eligibility.py b/openpilot/starpilot/flm/torque_eligibility.py
index 8882b7c0db..092ab8bbec 100644
--- a/openpilot/starpilot/flm/torque_eligibility.py
+++ b/openpilot/starpilot/flm/torque_eligibility.py
@@ -29,6 +29,24 @@ def bounded_context_output(output):
     return False
 
 
+def consumed_sources_current(control, output):
+  """Two actual consumption instants, never a cross-process source-age proxy.
+
+  Each state is fresh at its own consumer; the original command itself also
+  reaches Card within30ms. This is evidence qualification, not driving authority.
+  """
+  try:
+    stamps = control.state_ns, control.control_ns, output.state_ns, output.apply_ns
+    return bool(all(type(stamp) is int and stamp > 0 for stamp in stamps) and
+                control.state_ns <= control.control_ns <= output.apply_ns and
+                control.state_ns <= output.state_ns <= output.apply_ns and
+                control.control_ns - control.state_ns <= 30_000_000 and
+                output.apply_ns - output.state_ns <= 30_000_000 and
+                output.apply_ns - control.control_ns <= 30_000_000)
+  except (AttributeError, TypeError, ValueError, OverflowError):
+    return False
+
+
 def exact_unclipped_pair(control, output):
   """Resolve one async hint only from its original consumed source and leases."""
   try:
@@ -36,7 +54,7 @@ def exact_unclipped_pair(control, output):
     if (not asynchronous_hint(control) or not isinstance(witness, TorqueProjection) or
         not output.valid or not output.active or output.override or output.fault or output.limited or not output.native_allowed or
         output.domain != control.domain or output.control_ns != control.control_ns or
-        not control.state_ns <= output.state_ns <= output.apply_ns or output.state_ns - control.state_ns > 30_000_000 or
+        not consumed_sources_current(control, output) or
         not 0 < control.source_receipt_ns <= control.control_ns <= control.source_receipt_ns + 500_000_000 or
         witness.requested_torque != control.requested or witness.reported_torque != output.applied or
         not witness.steer_requested or witness.unlimited_raw != witness.final_raw):