Merge latest steering learning changes for Toyota publication

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