Preserve FLM learning windows across unchanged steering repeats

The d5 Ioniq 6 logs repeatedly reset source tracking when Card repeats a command that already arrived on time. Discard only a recent unchanged repeat after the existing native CAN verifier succeeds; add no sample, application time or permission.

Use independently admitted pending CarState brackets to complete original poses without waiting for that command to reach Card. Classify stale control source clocks as transient stale evidence, preserving the original bounded study deadline. Replace Python character loops with equivalent strict text predicates.

Validation: focused regression suite and paced d5 replay with original Card command references; actual controller/CAN trial application and worsening-trial rollback. No timing/acceptance threshold changes or claimed road improvement.

openpilot/starpilot/flm/fullrate_collector.py

@@ -327,7 +327,7 @@ class FullrateCollector:
       or type(value.control_ns) is not int
       or type(value.state_ns) is not int
       or not 0 < value.state_ns <= value.control_ns <= now_ns
-      or value.control_ns - value.state_ns > JOIN_NS
+      or value.domain != 'normalized_torque' and value.control_ns - value.state_ns > JOIN_NS
       or not 1 <= len(value.features) <= 8
       or not all(finite(v) for v in (value.requested, value.baseline_prediction, *value.features, *value.applied_coordinates))
       or len(value.stage_values) > 40
@@ -410,9 +410,9 @@ class FullrateCollector:
     ):
       self.split('ineligible_control')
       return False
-    if now_ns - value.control_ns > MAX_AGE_NS:
-      # Expiry alone rejects this row, but may retain an already suspended
-      # numerical study. Explicit original native refusal is not a timing gap.
+    if now_ns - value.control_ns > MAX_AGE_NS or value.control_ns - value.state_ns > JOIN_NS:
+      # Receiver or normalized source-state delay still rejects this row.
+      # Only inert study data may survive; native refusal is not a timing gap.
       if value.native_required and (not value.native_allowed or not identity(value.native_identity) or
           not 0 < value.native_observed_ns <= value.control_ns <= value.native_expires_ns <= value.native_observed_ns + 200_000_000):
         self.split('original_native_admission_unavailable')
@@ -518,6 +518,11 @@ class FullrateCollector:
     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:
+      from openpilot.starpilot.flm.unchanged_output import redundant_late_output
+      if redundant_late_output(self, command, output):
+        # The original on-time pair remains untouched. This repeat creates no
+        # ready row, sample, application time, physical witness or permission.
+        return
       # 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.

openpilot/starpilot/flm/live_trial_contract.py

@@ -13,11 +13,11 @@ MAX_TRIAL_NS = 120_000_000_000
 
 
 def _text(value, maximum=96):
-  return type(value) is str and 0 < len(value) <= maximum and value.isascii() and all(32 <= ord(c) < 127 for c in value)
+  return type(value) is str and 0 < len(value) <= maximum and value.isascii() and value.isprintable()
 
 
 def _digest(value):
-  return type(value) is str and len(value) == 64 and all(c in '0123456789abcdef' for c in value)
+  return type(value) is str and len(value) == 64 and not value.strip('0123456789abcdef')
 
 
 @dataclass(frozen=True)

openpilot/starpilot/flm/original_pose_brackets.py

@@ -0,0 +1,38 @@
+"""Original consumed vehicle states; no actuation or physical permission."""
+
+from openpilot.starpilot.flm.fullrate_collector import ControlEvidence, JOIN_NS, MAX_AGE_NS
+
+
+def state_brackets(collector, identity, now_ns):
+  states = {}
+
+  def add(control):
+    if (control.identity, control.generation, control.source_epoch, control.baseline_revision,
+        control.controller_revision, control.native_identity) != identity:
+      return
+    values = dict(control.replay_inputs)
+    if 'vEgo' in values and 'steeringAngleDeg' in values:
+      row = (float(values['vEgo']), float(values['steeringAngleDeg']))
+      if control.state_ns in states and states[control.state_ns] != row:
+        raise ValueError('conflicting original vehicle state brackets')
+      states[control.state_ns] = row
+
+  for control, _ in collector.ready:
+    add(control)
+  # These controls already passed collector admission. Their independently
+  # recorded CarState can bracket a pose before Card emits their command.
+  # The observer still requires its original joined native pair; these rows
+  # supply only speed/angle, and never enter ready or renew any witness.
+  for control in collector.controls.values():
+    if (type(control) is not ControlEvidence or control.domain != collector.ready[-1][0].domain
+        or control.native_required is not collector.ready[-1][0].native_required
+        or not control.enabled or not control.active or not control.valid
+        or control.manual or control.override or control.fault or control.saturated
+        or not 0 < control.state_ns <= control.control_ns <= now_ns
+        or control.control_ns - control.state_ns > JOIN_NS or now_ns - control.control_ns > MAX_AGE_NS):
+      continue
+    if control.native_required and (not control.native_allowed or
+        not 0 < control.native_observed_ns <= control.control_ns <= control.native_expires_ns <= control.native_observed_ns + 200_000_000):
+      continue
+    add(control)
+  return states

openpilot/starpilot/flm/runtime_adapter_registry.py

@@ -877,13 +877,8 @@ def original_source_pose_observer(sm, cp, collector, now_ns, *, distinguish_pend
     if queue:
       # No original pose is waiting after a completed publication. Rebuild
       # state brackets only for retained/new rows, after all current guards.
-      for command, _ in ready:
-        if (command.identity, command.generation, command.source_epoch, command.baseline_revision,
-            command.controller_revision, command.native_identity) != current_identity:
-          continue  # State brackets cannot bridge a command/native epoch.
-        values = dict(command.replay_inputs)
-        if 'vEgo' in values and 'steeringAngleDeg' in values:
-          history[command.state_ns] = (float(values['vEgo']), float(values['steeringAngleDeg']))
+      from openpilot.starpilot.flm.original_pose_brackets import state_brackets
+      history = state_brackets(collector, current_identity, now_ns)
     for index, evidence in enumerate(queue):
       before = [stamp for stamp in history if stamp <= (evidence.source_lower_ns or evidence.mono_ns)]
       after = [stamp for stamp in history if stamp >= (evidence.source_upper_ns or evidence.mono_ns)]

openpilot/starpilot/flm/steerlearnerd.py

@@ -132,13 +132,13 @@ def _source_config_known(source, now_ns):
     type(source.observed_ns) is int and type(source.expires_ns) is int and
     0 < source.observed_ns <= now_ns <= source.expires_ns <= source.observed_ns + SOURCE_LEASE_NS and
     type(source.cp_sha256) is str and len(source.cp_sha256) == 64 and
-    all(char in '0123456789abcdef' for char in source.cp_sha256) and
+    not source.cp_sha256.strip('0123456789abcdef') and
     type(source.reset) is str and 1 <= len(source.reset) <= 64 and
-    all(char in '0123456789abcdef' for char in source.reset) and
+    not source.reset.strip('0123456789abcdef') and
     type(source.calibration_generation) is str and (not source.calibration_generation or
-      len(source.calibration_generation) == 32 and all(char in '0123456789abcdef' for char in source.calibration_generation)) and
+      len(source.calibration_generation) == 32 and not source.calibration_generation.strip('0123456789abcdef')) and
     type(source.persistent_source_epoch) is str and (not source.persistent_source_epoch or
-      len(source.persistent_source_epoch) == 64 and all(char in '0123456789abcdef' for char in source.persistent_source_epoch)) and
+      len(source.persistent_source_epoch) == 64 and not source.persistent_source_epoch.strip('0123456789abcdef')) and
     type(source.manual_baseline) is tuple and len(source.manual_baseline) in (0, 3) and
     all(type(value) in (int, float) and math.isfinite(value) for value in source.manual_baseline))
 
@@ -433,7 +433,7 @@ def run(
       return False, now
     valid = bool(known and source.enabled and not source.manual and
       type(source.source_epoch) is str and len(source.source_epoch) == 64 and
-      all(char in '0123456789abcdef' for char in source.source_epoch) and
+      not source.source_epoch.strip('0123456789abcdef') and
       (source_cp_sha256 is None or source.cp_sha256 == source_cp_sha256))
     if not valid:
       source_rejected = True

openpilot/starpilot/flm/tests/test_control_source_clock_recovery.py

@@ -0,0 +1,94 @@
+"""Timing rejection and inert study recovery; numerical fixtures, no road proof."""
+from dataclasses import replace
+from types import SimpleNamespace
+
+import pytest
+
+from openpilot.starpilot.flm.fullrate_collector import FullrateCollector
+from openpilot.starpilot.flm.tests.test_fullrate_pipeline import SyntheticAdapter, snapshots
+from openpilot.starpilot.flm.tests.test_source_study_gaps import active_study
+from openpilot.starpilot.flm.tests.test_tracking_source_context import refresh, source_ack
+
+
+def subject(age=43_305_053, *, pause=True):
+  owner, now = active_study()
+  if pause:
+    owner.interrupt('expired_response_witness')
+  control, _, _ = snapshots(now)
+  control = replace(control, domain='normalized_torque', state_ns=now-age, native_required=True,
+    native_allowed=True, native_identity='4'*64, native_observed_ns=now-50_000_000,
+    native_expires_ns=now+150_000_000)
+  collector = FullrateCollector(SyntheticAdapter())
+  collector.adapter.command_domain = 'normalized_torque'
+  collector.context = tuple(getattr(control, name) for name in (
+    'identity', 'generation', 'domain', 'stage_revision', 'response_revision', 'source_epoch',
+    'controller_mode', 'policy', 'baseline_revision', 'controller_revision', 'turn_assist'))
+  collector.live_tracking = SimpleNamespace(interrupt=owner.interrupt)
+  return collector, owner, control, now+37_720_875
+
+
+@pytest.mark.parametrize('age', (30_000_001, 43_305_053))
+@pytest.mark.parametrize('pause', (False, True))
+def test_normalized_source_clock_gap_rejects_row_without_erasing_numerical_study(age, pause):
+  collector, owner, control, now = subject(age, pause=pause)
+  original = (owner.paused_intent[0] if pause else owner.offer_state).copy()
+  before, future = owner.paused_intent[1:] if pause else (owner.before, tuple(owner.future))
+  assert not collector.control(control, now)
+  assert collector.reason == 'stale_control'
+  assert not collector.controls and not collector.ready and not collector.recent_controls
+  assert collector.last_qualified_response == collector.last_apply == 0
+  assert owner.offer_state['phase'] == 'rollback' and owner.settled_ns is None
+  assert owner.paused_intent is not None
+  assert owner.paused_intent[1:] == (before, future)
+  assert not owner.before and not owner.future and not any(owner.accepted.values())
+  refresh(owner, now+200_000_000)
+  rollback = owner.offer(now+200_000_000)
+  assert rollback is not None and rollback.phase == 'rollback'
+  assert source_ack(owner, rollback, now+210_000_000)
+  resumed = owner.offer_state
+  assert resumed['phase'] == 'trial' and resumed['trial_id'] != original['trial_id']
+  assert resumed['target_sequence'] > original['target_sequence'] and resumed['published'] is False
+  for name in ('started_ns', 'deadline_ns', 'alpha', 'required_episodes', 'look_quotas', 'target', 'baseline'):
+    assert resumed[name] == original[name]
+  assert (owner.before, tuple(owner.future)) == (before, future) and owner.settled_ns is None
+
+
+@pytest.mark.parametrize('change,reason', (
+  ({'identity': 'bad'}, 'invalid_control'),
+  ({'requested': float('nan')}, 'invalid_control'),
+  ({'valid': False}, 'ineligible_control'),
+  ({'fault': True}, 'ineligible_control'),
+  ({'override': True}, 'ineligible_control'),
+  ({'manual': True}, 'ineligible_control'),
+  ({'active': False}, 'ineligible_control'),
+  ({'saturated': True}, 'ineligible_control'),
+  ({'native_allowed': False}, 'original_native_admission_unavailable'),
+  ({'native_expires_ns': 1}, 'original_native_admission_unavailable'),
+  ({'source_receipt_ns': 1}, 'invalid_control'),
+  ({'domain': 'torque'}, 'invalid_control'),
+))
+def test_clock_gap_never_masks_fault_native_structural_or_other_domain_failure(change, reason):
+  collector, owner, control, now = subject()
+  assert owner.paused_intent is not None
+  assert not collector.control(replace(control, **change), now)
+  assert collector.reason == reason and owner.paused_intent is None
+  assert not collector.ready and not collector.controls and not any(owner.accepted.values())
+
+
+@pytest.mark.parametrize('change', ({'identity': '5'*64}, {'source_epoch': '6'*64}, {'baseline_revision': '7'*64}))
+def test_clock_gap_with_changed_context_cannot_preserve_old_study(change):
+  collector, owner, control, now = subject()
+  assert not collector.control(replace(control, **change), now)
+  assert collector.training_interruptions['actual_controller_identity_changed'] == 1
+  assert owner.paused_intent is None and not collector.ready
+
+
+def test_original_deadline_still_expires_after_source_clock_gap():
+  collector, owner, control, now = subject()
+  deadline = owner.paused_intent[0]['deadline_ns']
+  assert not collector.control(control, now)
+  refresh(owner, deadline+1)
+  rollback = owner.offer(deadline+1)
+  assert rollback is not None and source_ack(owner, rollback, deadline+10_000_001)
+  assert owner.paused_intent is None and owner.offer_state['phase'] == 'rollback'
+  assert owner.offer_state['accepted'] and not any(owner.accepted.values())

openpilot/starpilot/flm/tests/test_exact_text_validation.py

@@ -0,0 +1,41 @@
+"""Equivalent bounded text predicates, including malformed wire text."""
+from dataclasses import replace
+from types import SimpleNamespace
+
+from openpilot.starpilot.flm.live_trial_contract import _digest, _text
+from openpilot.starpilot.flm.runtime_adapter_registry import SourceState
+from openpilot.starpilot.flm.steerlearnerd import _source_config_known
+
+
+class TextSubclass(str):
+  pass
+
+
+def old_text(value, maximum):
+  return type(value) is str and 0 < len(value) <= maximum and value.isascii() and all(32 <= ord(c) < 127 for c in value)
+
+
+def old_digest(value):
+  return type(value) is str and len(value) == 64 and all(c in '0123456789abcdef' for c in value)
+
+
+def test_text_and_digest_match_exact_prior_predicates():
+  values = [None, False, 0, b'a', [], {}, SimpleNamespace(), TextSubclass('a'), '', 'a' * 193]
+  alphabet = [chr(i) for i in range(256)] + ['\u200b', '\u2028', '\ud800', '\uffff', '\U0001f642']
+  for char in alphabet:
+    values.extend((char, char * 64, 'a' * 31 + char + '0' * 32, 'a' * 63 + char, char + 'a' * 63))
+  for value in values:
+    assert _digest(value) is old_digest(value)
+    for bound in (0, 1, 32, 64, 96, 192):
+      assert _text(value, bound) is old_text(value, bound)
+
+
+def test_source_binding_hex_remains_strict():
+  source = SourceState(True, False, '0', 'a' * 64, 100, 500_000_100, 'b' * 64,
+    calibration_generation='c' * 32, persistent_source_epoch='d' * 64)
+  assert _source_config_known(source, 200)
+  for field in ('cp_sha256', 'reset', 'calibration_generation', 'persistent_source_epoch'):
+    original = getattr(source, field)
+    for bad in ('g', 'A', ' ', '\n', '\x00', '\u00e9'):
+      assert not _source_config_known(replace(source, **{field: original[:-1] + bad}), 200)
+  assert not _source_config_known(source, 500_000_101)

openpilot/starpilot/flm/tests/test_original_pose_pending_state.py

@@ -0,0 +1,63 @@
+"""A consumed state is usable before its command's independent CAN receipt."""
+from dataclasses import replace
+
+import pytest
+
+from openpilot.starpilot.flm.original_source_capture import OriginalSourcePose, validate_pose
+from openpilot.starpilot.flm.physical_observation import PendingPhysical
+from openpilot.starpilot.flm.tests.test_original_pose_work import BASE, fixture, observe, packet, pair
+
+
+def pending_right(collector, sm, **changes):
+  stamp = BASE + 10_000_000
+  sm.update_msgs(stamp / 1e9, [packet('carState', stamp).as_reader()])
+  control, _ = pair(stamp, speed=22., angle=2.)
+  control = replace(control, **changes)
+  collector.controls[control.control_ns] = control
+  return control
+
+
+def test_pose_uses_original_pending_state_without_admitting_its_actuation():
+  with fixture() as (collector, cp, sm):
+    assert type(observe(collector, cp, sm, BASE)) is PendingPhysical
+    original = tuple(collector.ready)
+    control = pending_right(collector, sm)
+    row = observe(collector, cp, sm, BASE + 12_000_000)
+    assert type(row) is OriginalSourcePose
+    validate_pose(row, BASE + 12_000_000)
+    assert (row.speed_mps, row.steering_angle_deg) == (21., 1.)
+    assert row.right_state_ns == control.state_ns
+    assert (row.native_control_ns, row.native_apply_ns) == (original[-1][0].control_ns, original[-1][1].apply_ns)
+    assert tuple(collector.ready) == original
+    assert collector.controls[control.control_ns] is control
+    assert (row.mono_ns, row.received_ns, row.observed_ns) == (BASE, BASE, BASE)
+
+
+@pytest.mark.parametrize('changes', [
+  {'identity': '9' * 64}, {'generation': 2}, {'source_epoch': '9' * 64},
+  {'baseline_revision': 'other'}, {'controller_revision': 'other'}, {'native_identity': '9' * 64},
+  {'enabled': False}, {'active': False}, {'valid': False}, {'manual': True},
+  {'override': True}, {'fault': True}, {'saturated': True}, {'native_allowed': False},
+  {'control_ns': BASE + 12_000_001}, {'control_ns': BASE + 41_000_000},
+  {'native_expires_ns': BASE + 10_000_000}, {'native_required': False}, {'domain': 'steering_angle'},
+])
+def test_unqualified_pending_state_cannot_complete_pose(changes):
+  with fixture() as (collector, cp, sm):
+    assert type(observe(collector, cp, sm, BASE)) is PendingPhysical
+    pending_right(collector, sm, **changes)
+    assert type(observe(collector, cp, sm, BASE + 12_000_000)) is PendingPhysical
+
+
+def test_pending_state_does_not_replace_current_native_permission():
+  with fixture() as (collector, cp, sm):
+    pending_right(collector, sm)
+    collector._learning_current_native_allowed = False
+    assert observe(collector, cp, sm, BASE + 12_000_000) is None
+
+
+def test_conflicting_source_state_is_refused():
+  with fixture() as (collector, cp, sm):
+    pending_right(collector, sm)
+    control, _ = pair(BASE - 10_000_000, speed=99.)
+    collector.controls[control.control_ns] = control
+    assert observe(collector, cp, sm, BASE + 12_000_000) is None

openpilot/starpilot/flm/tests/test_unchanged_output.py

@@ -0,0 +1,137 @@
+"""Split policy with a declared native verifier fixture; no CAN or road claim."""
+from dataclasses import replace
+from types import SimpleNamespace
+from typing import cast
+from unittest.mock import Mock
+
+import pytest
+from opendbc.car.steering_projection import TorqueProjection
+from openpilot.starpilot.flm.fullrate_collector import FullrateCollector
+from openpilot.starpilot.flm.tests.test_fullrate_pipeline import BASE, SyntheticAdapter, snapshots
+from openpilot.starpilot.flm.torque_eligibility import consumed_sources_current
+from openpilot.starpilot.flm.unchanged_output import redundant_late_output
+
+
+def fixture():
+  control, output, _ = snapshots(BASE)
+  control = replace(control, domain='normalized_torque', native_required=True, native_allowed=True,
+    native_identity='4' * 64, native_observed_ns=BASE - 50_000_000, native_expires_ns=BASE + 150_000_000)
+  limits = tuple(zip(('STEER_DELTA_UP', 'STEER_DELTA_DOWN', 'STEER_DRIVER_ALLOWANCE',
+    'STEER_DRIVER_MULTIPLIER', 'STEER_DRIVER_FACTOR'), (10., 10., 50., 2., 1.), strict=True))
+  witness = TorqueProjection(1, 'hyundai', 'synthetic', 0, 'hyundai.LKAS11', 'round', 10,
+    .2, .2, 200, 200, 1000., (), (), 'driver', 200., 0., limits, (), True, 832, 0, b'12345678', .2)
+  previous = replace(output, domain='normalized_torque', state_ns=BASE + 19_000_000,
+    apply_ns=BASE + 20_000_000, projection=witness)
+  later = replace(previous, state_ns=BASE + 29_000_000, apply_ns=BASE + 31_000_000,
+    projection=witness._replace(frame=11, can_data=b'87654321'))
+  collector = FullrateCollector(SyntheticAdapter())
+  collector.adapter.command_domain = 'normalized_torque'
+  collector.adapter.replay = SimpleNamespace(applied_matches=Mock(return_value=True))
+  collector.live_tracking = SimpleNamespace(episodes=SimpleNamespace(source_tracking=True), interrupt=Mock())
+  collector._learning_current_native_allowed = True
+  collector.ready.append((control, previous))
+  collector.last_apply = previous.apply_ns
+  collector.last_response = collector.last_qualified_response = BASE + 21_000_000
+  return collector, control, previous, later
+
+
+def test_verified_unchanged_repeat_does_not_erase_or_renew_existing_evidence():
+  collector, control, previous, later = fixture()
+  assert consumed_sources_current(control, previous)
+  assert not consumed_sources_current(control, later)
+  prior = (tuple(collector.ready), collector.last_apply, collector.last_response, collector.last_qualified_response)
+  collector._join_emission(control, later)
+  assert (tuple(collector.ready), collector.last_apply, collector.last_response, collector.last_qualified_response) == prior
+  collector.live_tracking.interrupt.assert_not_called()
+  collector.adapter.replay.applied_matches.assert_called_once_with(control, later)
+  assert not collector.training_interruptions
+
+
+@pytest.mark.parametrize('case', ('no_prior', 'different_control', 'plant', 'native', 'decode', 'changed_output',
+  'limited', 'override', 'fault', 'inactive', 'invalid', 'card_age', 'source_age', 'expired_native',
+  'old_prior', 'codec', 'requested', 'not_late', 'native_disabled'))
+def test_repeat_exception_cannot_admit_changed_unqualified_or_unverified_input(case):
+  collector, control, previous, later = fixture()
+  if case == 'no_prior':
+    collector.ready.clear()
+  elif case == 'different_control':
+    control = replace(control, generation=control.generation + 1)
+  elif case == 'plant':
+    collector.live_tracking.episodes.source_tracking = False
+  elif case == 'native':
+    collector._learning_current_native_allowed = False
+  elif case == 'decode':
+    collector.adapter.replay.applied_matches.return_value = False
+  elif case == 'changed_output':
+    later = replace(later, applied=.201, projection=cast(TorqueProjection, later.projection)._replace(final_raw=201, reported_torque=.201))
+  elif case in ('limited', 'override', 'fault'):
+    later = replace(later, **{case: True})
+  elif case in ('inactive', 'invalid'):
+    later = replace(later, **{'active' if case == 'inactive' else 'valid': False})
+  elif case == 'card_age':
+    later = replace(later, state_ns=BASE)
+  elif case == 'source_age':
+    control = replace(control, state_ns=BASE - 30_000_001)
+    collector.ready[-1] = (control, previous)
+  elif case == 'expired_native':
+    control = replace(control, native_expires_ns=BASE + 30_000_000)
+    collector.ready[-1] = (control, previous)
+  elif case == 'old_prior':
+    later = replace(later, apply_ns=BASE + 50_000_001, state_ns=BASE + 50_000_000)
+  elif case == 'codec':
+    later = replace(later, projection=cast(TorqueProjection, later.projection)._replace(can_bus=1))
+  elif case == 'requested':
+    later = replace(later, projection=cast(TorqueProjection, later.projection)._replace(requested_torque=.201))
+  elif case == 'not_late':
+    later = replace(later, apply_ns=BASE + 30_000_000)
+  elif case == 'native_disabled':
+    later = replace(later, native_allowed=False)
+  assert not redundant_late_output(collector, control, later)
+
+
+def test_failed_native_decode_keeps_existing_withdrawal():
+  collector, control, _, later = fixture()
+  collector.adapter.replay.applied_matches.return_value = False
+  collector._join_emission(control, later)
+  assert not collector.ready and collector.last_apply == collector.last_qualified_response == 0
+  collector.live_tracking.interrupt.assert_called_once_with('consumed_source_mismatch', context_changed=False)
+
+
+@pytest.mark.parametrize('tamper', (False, True))
+def test_ioniq6_unchanged_repeat_uses_actual_can_decoder(tmp_path, tamper):
+  # Physical state/current permission are declared test fixtures. CarInterface,
+  # encoder, packets, emission history and decoder are the registered path.
+  from opendbc.car.hyundai.values import CAR
+  from openpilot.starpilot.flm.controller_replay import ControllerReplay
+  from openpilot.starpilot.flm.torque_emission_replay import EmissionHistory
+  from openpilot.starpilot.flm.tests.test_torque_emission_join import interface, cc, settle, emitted, BASE as ENCODER_BASE
+
+  ci = interface(CAR.HYUNDAI_IONIQ_6)
+  command = cc(0.)
+  settle(ci, command)
+  stamp = ENCODER_BASE + ci.CC.frame * 10_000_000
+  source, first = emitted(ci, command, tmp_path, control_ns=stamp)
+  collector, control, _, _ = fixture()
+  control = replace(control, requested=0., control_ns=stamp, state_ns=stamp-1,
+    identity=source.identity, source_epoch=source.source_epoch, baseline_revision=source.baseline_revision,
+    native_required=False)
+  history = EmissionHistory(ci.CP, identity=control.identity, baseline_revision=control.baseline_revision)
+  rows = [first]
+  for _ in range(4):
+    _, output = emitted(ci, command, tmp_path, control_ns=stamp)
+    rows.append(output)
+  if tamper:
+    witness = rows[-1].projection
+    assert witness is not None
+    data = bytes([witness.can_data[0] ^ 0xff]) + witness.can_data[1:]
+    rows[-1] = replace(rows[-1], projection=witness._replace(can_data=data))
+  for output in rows:
+    assert history.record_carrier(output) and history.record(control, output)
+  previous, later = rows[-2:]
+  assert consumed_sources_current(control, previous) and not consumed_sources_current(control, later)
+  replay = ControllerReplay.__new__(ControllerReplay)
+  replay.emissions = history
+  collector.adapter.replay = replay
+  collector.ready.clear()
+  collector.ready.append((control, previous))
+  assert redundant_late_output(collector, control, later) is not tamper

openpilot/starpilot/flm/unchanged_output.py

@@ -0,0 +1,42 @@
+"""Discard an unchanged repeat without admitting a late actuation sample."""
+
+
+def redundant_late_output(collector, control, output):
+  from openpilot.starpilot.flm.fullrate_collector import JOIN_NS
+  from openpilot.starpilot.flm.torque_eligibility import consumed_sources_current
+  from opendbc.car.steering_projection import TorqueProjection, projection_to_dict
+
+  live = collector.live_tracking
+  if (live is None or getattr(getattr(live, 'episodes', None), 'source_tracking', False) is not True
+      or getattr(collector, '_learning_current_native_allowed', False) is not True
+      or control.domain != 'normalized_torque' or output.domain != control.domain or not collector.ready):
+    return False
+  previous_control, previous = collector.ready[-1]
+  if (previous_control is not control or not consumed_sources_current(control, previous)
+      or not control.state_ns <= control.control_ns <= output.state_ns <= output.apply_ns
+      or control.control_ns - control.state_ns > JOIN_NS or output.apply_ns - output.state_ns > JOIN_NS
+      or output.apply_ns - control.control_ns <= JOIN_NS
+      or not previous.apply_ns < output.apply_ns <= previous.apply_ns + JOIN_NS
+      or not control.enabled or not control.active or not control.valid or control.manual or control.override or control.fault
+      or control.saturated or control.model_wobble
+      or any(not row.valid or not row.active or row.override or row.fault or row.limited or not row.native_allowed
+             for row in (previous, output))):
+    return False
+  if control.native_required and (not control.native_allowed or
+      not 0 < control.native_observed_ns <= control.control_ns <= output.apply_ns <= control.native_expires_ns <= control.native_observed_ns + 200_000_000):
+    return False
+  old, new = previous.projection, output.projection
+  if (type(old) is not TorqueProjection or type(new) is not TorqueProjection
+      or projection_to_dict(old) is None or projection_to_dict(new) is None
+      or output.control_ns != control.control_ns or output.applied != previous.applied
+      or new.requested_torque != control.requested or new.reported_torque != output.applied or not new.steer_requested):
+    return False
+  # Current CAN bytes/counter and native prestate are independently verified
+  # below. Every other encoded command/codec field must remain exactly equal.
+  changing = {'frame', 'can_data', 'previous_torque', 'measured_torque'}
+  if any(getattr(old, name) != getattr(new, name) for name in TorqueProjection._fields if name not in changing):
+    return False
+  replay = getattr(collector.adapter, 'replay', None)
+  if replay is None:
+    return False
+  return replay.applied_matches(control, output) is True

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "99155a256dac423974ec48b2ea94b5b7b8cea2412cd207350f5cf259e238fc55", "schema_version": 1}
+{"protected_inventory_sha256": "785627931b899d1f0b0af928e64147556fe0ded7f60d012770fbaf9a6455978b", "schema_version": 1}