From 0022fde1cbf34bdaa871cdce24e4d0e6ad14e16d Mon Sep 17 00:00:00 2001
From: firestar5683 <168790843+firestar5683@users.noreply.github.com>
Date: Fri, 23 Oct 2026 17:36:10 -0500
Subject: [PATCH] Fix Auto Learn collection recovery after evidence gaps

---
 openpilot/starpilot/flm/fullrate_collector.py |  17 +-
 openpilot/starpilot/flm/live_tracking.py      |  47 +++-
 .../starpilot/flm/live_tracking_actor.py      |  13 +-
 openpilot/starpilot/flm/steerlearnerd.py      |  26 +-
 .../flm/tests/test_tracking_study_recovery.py | 247 ++++++++++++++++++
 prebuilt.json                                 |   2 +-
 6 files changed, 342 insertions(+), 10 deletions(-)
 create mode 100644 openpilot/starpilot/flm/tests/test_tracking_study_recovery.py

diff --git a/openpilot/starpilot/flm/fullrate_collector.py b/openpilot/starpilot/flm/fullrate_collector.py
index 5cc52458df..448cc37940 100644
--- a/openpilot/starpilot/flm/fullrate_collector.py
+++ b/openpilot/starpilot/flm/fullrate_collector.py
@@ -327,7 +327,6 @@ 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 now_ns - value.control_ns > MAX_AGE_NS
       or 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))
@@ -411,6 +410,15 @@ 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 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')
+        return False
+      self.split('stale_control')
+      return False
     previous = self.recent_controls.get(value.control_ns)
     if previous is not None:
       if previous != value:
@@ -443,7 +451,6 @@ class FullrateCollector:
       or any(type(n) is not int for n in (value.control_ns, value.state_ns, value.apply_ns))
       or not 0 < value.control_ns <= value.apply_ns <= now_ns
       or not 0 < value.state_ns <= value.apply_ns
-      or now_ns - value.apply_ns > MAX_AGE_NS
       or value.apply_ns - value.control_ns > MAX_AGE_NS
     ):
       self.split('invalid_applied_output')
@@ -471,6 +478,12 @@ class FullrateCollector:
         self.split('actual_carrier_unavailable', reset=True)
         return False
       self.last_carrier_frame = value.carrier_frame
+    if now_ns - value.apply_ns > MAX_AGE_NS:
+      if value.domain != self.adapter.command_domain:
+        self.split('consumed_source_mismatch', reset=True)
+        return False
+      self.split('stale_applied_output')
+      return False
     key = (value.control_ns, value.apply_ns)
     previous = self.recent_applied.get(key)
     if previous is not None:
diff --git a/openpilot/starpilot/flm/live_tracking.py b/openpilot/starpilot/flm/live_tracking.py
index 5ac55b162e..9e1b41f96f 100644
--- a/openpilot/starpilot/flm/live_tracking.py
+++ b/openpilot/starpilot/flm/live_tracking.py
@@ -479,6 +479,7 @@ class LiveTrackingOptimizer:
     self.selector_after_ns = self.last_episode_end_ns = 0
     self.comparison: list[TrackingEpisode] = []
     self.settled_ns: int | None = None
+    self.paused_intent: tuple[TrialState, tuple[TrackingEpisode, ...], tuple[TrackingEpisode, ...]] | None = None
     self.status, self.reason = 'collecting', 'qualified_tracking_episodes_required'
     self.model_status = 'unavailable'
 
@@ -486,6 +487,22 @@ class LiveTrackingOptimizer:
     return full_profile_sha256(self.binding.baseline_revision, values)
 
   def interrupt(self, reason: str, *, context_changed: bool = False) -> None:
+    # Retain only inert numerical study intent on missing/stale evidence. The
+    # command still rolls back; a fresh baseline ACK must precede a NEW wire
+    # trial. Completed scores, alpha and original deadline are never reset.
+    transient = not context_changed and reason in (
+      'acknowledged_trial_profile_withdrawn', 'expired_response_witness', 'stale_control', 'stale_applied_output',
+      'current_evidence_delayed', 'learning_evidence_delayed',
+      'fresh_physical_residual_window_required',
+    )
+    if not transient:
+      self.paused_intent = None
+    elif (reason in ('acknowledged_trial_profile_withdrawn', 'expired_response_witness') and self.offer_state is not None
+        and self.offer_state['phase'] == 'trial' and not self.offer_state.get('accepted', False) and self.before):
+      # Profile withdrawal requires a qualified newer source control. Expiry
+      # requires an otherwise valid original owned physical witness. Neither
+      # restores its command lease; both retain only the numerical study.
+      self.paused_intent = (self.offer_state.copy(), self.before, tuple(self.future))
     # Already completed clean geometry survives an ordinary brake/override.
     # The input builder discards interrupted partial episodes independently.
     retain_comparison = not context_changed and self.discovery is not None and (self.offer_state is None or self.offer_state.get('accepted') is True)
@@ -1137,6 +1154,7 @@ class LiveTrackingOptimizer:
     )
 
   def _rollback(self, reason: str) -> None:
+    self.paused_intent = None  # An actual scored rejection is terminal.
     if self.offer_state is None:
       raise ValueError('rollback requires a pending live trial')
     state = self.offer_state.copy()
@@ -1221,6 +1239,21 @@ class LiveTrackingOptimizer:
         self.offer_state = accepted
         if self.selector_pending:
           self.selector_after_ns = max(self.selector_after_ns, control_ns)
+      paused, self.paused_intent = self.paused_intent, None
+      if phase == 'rollback' and paused is not None:
+        intent, before, future = paused
+        if now_ns < intent['deadline_ns'] and intent['baseline'] == self.accepted:
+          self.sequence += 1
+          resumed = intent.copy()
+          resumed['trial_id'] = hashlib.sha256(repr((intent['trial_id'], self.sequence, control_ns)).encode()).hexdigest()
+          resumed['target_sequence'] = self.sequence
+          resumed['published'] = False
+          resumed['prior_accepted'] = self.offer_state
+          self.offer_state = resumed
+          self.before, self.future = before, list(future)
+          self.settled_ns = None
+          self.status, self.reason = 'trial_pending', 'fresh_target_application_after_evidence_gap_required'
+          return True
       self.baseline.clear()
       self.future.clear()
       self.settled_ns = None
@@ -1232,6 +1265,18 @@ class LiveTrackingOptimizer:
     if state is not None and offer.trial_id == state['trial_id'] and offer.target_sequence == state.get('target_sequence', self.sequence):
       state['published'] = True  # Conservative before-send marker, not an ACK.
 
+  def _witness_expired(self, now_ns: int) -> bool:
+    from openpilot.starpilot.flm.live_tracking_response import response_witness_value, validate_response_witness
+
+    try:
+      value = response_witness_value(self.response_witness)
+      # Validate every original field at its original deadline. This tests
+      # expiry only; it does not make the old witness current or authorize TX.
+      validate_response_witness(self.response_witness, self.binding, value.valid_until_ns)
+      return type(now_ns) is int and now_ns > value.valid_until_ns
+    except (AttributeError, TypeError, ValueError):
+      return False
+
   def offer(self, now_ns: int, *, publication=True) -> LiveTrialOffer | None:
     from openpilot.starpilot.flm.live_tracking_response import validate_response_witness
 
@@ -1239,7 +1284,7 @@ class LiveTrackingOptimizer:
       validate_response_witness(self.response_witness, self.binding, now_ns)
     except (TypeError, ValueError):
       if self.offer_state is not None:
-        self.interrupt('fresh_physical_response_witness_unavailable')
+        self.interrupt('expired_response_witness' if self._witness_expired(now_ns) else 'fresh_physical_response_witness_unavailable')
       return None
     state = self.offer_state
     if state is None:
diff --git a/openpilot/starpilot/flm/live_tracking_actor.py b/openpilot/starpilot/flm/live_tracking_actor.py
index d9281b02d5..cb2c979226 100644
--- a/openpilot/starpilot/flm/live_tracking_actor.py
+++ b/openpilot/starpilot/flm/live_tracking_actor.py
@@ -708,7 +708,7 @@ class LiveTrackingActor:
       == (self.seed.cp_sha256, self.seed.source_epoch, self.seed.reset_generation, self.seed.calibration_generation, self.seed.persistent_source_epoch)
     )
 
-  def tick(self, collector, source, *, now_ns, current_native, native_key=(), hardware_inventory=(), allow_bootstrap=True):
+  def tick(self, collector, source, *, now_ns, current_native, native_key=(), hardware_inventory=(), allow_bootstrap=True, evidence_delay_only=False):
     self.native_key, self.hardware_inventory = native_key, hardware_inventory
     snapshot_result = self.snapshot_process.poll(token=self.token if self._same_bootstrap_source(source, now_ns) else None, now_ns=now_ns)
     if self.probe_process is not None and self.ingress is None:
@@ -748,7 +748,8 @@ class LiveTrackingActor:
       self.probe_queue.clear()
       self.probe_queue_key = None
       if self.ingress is not None:
-        self.ingress.interrupt('current_source_or_native_permission_unavailable', preserve_completed=retain_probe)
+        self.ingress.interrupt('current_evidence_delayed' if evidence_delay_only else 'current_source_or_native_permission_unavailable',
+          preserve_completed=retain_probe)
       return None
     try:
       seed = self.binding_for(collector, source)
@@ -862,7 +863,13 @@ class LiveTrackingActor:
       offer.validate()
       old = response_witness_value(offer.response_witness)
       optimizer = self.ingress.optimizer
-      current = validate_response_witness(optimizer.response_witness, self.certificate.binding, now_ns)
+      try:
+        current = validate_response_witness(optimizer.response_witness, self.certificate.binding, now_ns)
+      except (TypeError, ValueError):
+        self.ingress.interrupt('expired_response_witness' if optimizer._witness_expired(now_ns)
+          and offer.binding == self.certificate.binding and old.certificate == self.certificate
+          else 'fresh_response_witness_unavailable')
+        return None
       state = optimizer.offer_state
       if offer.binding != self.certificate.binding or old.certificate != self.certificate or state is None:
         raise ValueError('trial publication identity changed')
diff --git a/openpilot/starpilot/flm/steerlearnerd.py b/openpilot/starpilot/flm/steerlearnerd.py
index 1c6e259dcc..afd77fece1 100644
--- a/openpilot/starpilot/flm/steerlearnerd.py
+++ b/openpilot/starpilot/flm/steerlearnerd.py
@@ -86,6 +86,15 @@ def _collect_source_control(collector, prepared, sm, cp, now_ns):
   return collector.control(prepared, now_ns)
 
 
+def _collect_applied_output(collector, applied, sm, cp, now_ns):
+  """Each emission rechecks current native health for its original consumed CC."""
+  from openpilot.starpilot.flm.native_health import current_native_health
+
+  control = collector.controls.get(applied.control_ns, collector.recent_controls.get(applied.control_ns))
+  collector._learning_current_native_allowed = bool(control is not None and current_native_health(sm, cp, control, now_ns))
+  return collector.output(applied, now_ns)
+
+
 def _source_config_binding(source):
   return (source.cp_sha256, source.source_epoch, source.reset, source.calibration_generation,
           source.enabled, source.manual, source.persistent_source_epoch, tuple(source.manual_baseline))
@@ -334,6 +343,7 @@ def run(
   live_offer = None
   previous_tracking_pose_header = 0
   next_tracking_status_ns = 0
+  study_health_control = None
   saved_live_targets = set()
   live_history = _LiveHistoryRestore(confirmed_history)
   import gc
@@ -341,6 +351,7 @@ def run(
   trial_gc_deferred = False
 
   def clear(reason):
+    nonlocal study_health_control
     nonlocal collector, raw_context, active_key, pending_key, pending_token, unavailable_reason, retained_event, pending_event, live_offer
     unavailable_reason = reason
     retained_event = pending_event = None
@@ -348,6 +359,7 @@ def run(
       if reporter is not None:
         reporter.reset(reason)
       collector.split(reason, reset=True)
+    study_health_control = None
     collector = None
     cached_reports.clear()
     prior_attached.clear()
@@ -419,6 +431,9 @@ def run(
     source_rejected = False
     return True, now
 
+  def study_native_current(now_ns):
+    return study_health_control is not None and current_native_health(sm, CP, study_health_control, now_ns)
+
   class Sink:
     def split(self, reason, **kwargs):
       if collector is not None:
@@ -462,6 +477,8 @@ def run(
         # Geometry arrives at100Hz independently of the actual native emission
         # cadence. This Boolean is current owner health, never a fabricated CAN.
         previous = collector.ready[-1][0] if collector.ready else None
+        if previous is not None:
+          study_health_control = previous
         collector._learning_current_native_allowed = bool(source_valid and previous is not None and
           current_native_health(sm, CP, previous, now))
 
@@ -472,7 +489,9 @@ def run(
           return
         if event.which() == 'starpilotLateralState' and event.starpilotLateralState.learningEvidence.version == 0:
           if collector is not None:
-            collector.split('no_current_learning_evidence')
+            collector.split('learning_evidence_delayed' if event.starpilotLateralState.active and
+              study_native_current(received_now)
+              else 'no_current_learning_evidence')
           return
         evidence = read_control(event)
         if (
@@ -568,7 +587,7 @@ def run(
           if event.which() == 'starpilotCarState' and event.starpilotCarState.appliedEvidence.version == 0:
             collector.split('no_current_applied_evidence')
             return
-          collector.output(read_applied(event), received_now)
+          _collect_applied_output(collector, read_applied(event), 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)
@@ -688,7 +707,8 @@ def run(
             current_native=current_native, native_key=native_key, hardware_inventory=hardware_inventory)
         latest = live_actor.tick(collector, source if source_valid else None, now_ns=now,
           current_native=current_native, native_key=native_key, hardware_inventory=hardware_inventory,
-          allow_bootstrap=live_history.requalifier is None)
+          allow_bootstrap=live_history.requalifier is None,
+          evidence_delay_only=bool(source_valid and study_native_current(now)))
         if live_history.observe_response_recovery(live_actor):
           # Clear cached trial/restore publication in this SAME parent tick.
           # No old history or accepted target is silently promoted afterward.
diff --git a/openpilot/starpilot/flm/tests/test_tracking_study_recovery.py b/openpilot/starpilot/flm/tests/test_tracking_study_recovery.py
new file mode 100644
index 0000000000..d5a774465e
--- /dev/null
+++ b/openpilot/starpilot/flm/tests/test_tracking_study_recovery.py
@@ -0,0 +1,247 @@
+"""Numerical intent only; actual command/native leases are never restored."""
+import pytest
+from openpilot.starpilot.flm.tests.test_live_tracking import optimizer, ready, refresh, acknowledge, episode
+
+
+def active(owner=None, error=.1):
+  owner = optimizer() if owner is None else owner
+  state, now = ready(owner)
+  assert state is not None
+  pending = owner.offer(now)
+  assert pending is not None and acknowledge(owner, pending)
+  row = episode(owner, 20, 'ff_gain_left', profile=owner._hash(state['target']), error=error, start=now+2_000_000_000)
+  refresh(owner, row.end_ns)
+  owner.observe_episode(row, now_ns=row.end_ns)
+  assert len(owner.future) == 1
+  return owner, row.end_ns, state.copy(), owner.before, tuple(owner.future)
+
+
+def rollback_ack(owner, now):
+  refresh(owner, now)
+  offer = owner.offer(now)
+  assert offer is not None and offer.phase == 'rollback'
+  assert acknowledge(owner, offer, stamp=now+10_000_000)
+
+
+@pytest.mark.parametrize('error', (.1, .201))
+def test_evidence_gap_retains_study_but_requires_fresh_baseline_ack_and_new_wire_trial(error):
+  owner, now, old, before, future = active(error=error)
+  owner.interrupt('acknowledged_trial_profile_withdrawn')
+  assert owner.offer_state['phase'] == 'rollback' and owner.settled_ns is None
+  assert owner.before == () and owner.future == []
+  rollback_ack(owner, now+100_000_000)
+  resumed = owner.offer_state
+  assert resumed['phase'] == 'trial' and resumed['trial_id'] != old['trial_id']
+  assert resumed['target_sequence'] > old['target_sequence'] and resumed['published'] is False
+  assert owner.before == before and tuple(owner.future) == future and owner.settled_ns is None
+  for field in ('target', 'baseline', 'started_ns', 'deadline_ns', 'alpha', 'required_episodes', 'look_quotas'):
+    assert resumed[field] == old[field]
+  assert owner.paused_intent is None
+
+
+@pytest.mark.parametrize('reason', ('manual_profile_override', 'episode actual native join unqualified', 'persistent_physical_response_drift',
+  'invalid_control', 'ineligible_applied_output', 'current_source_or_native_permission_unavailable'))
+def test_driver_fault_or_model_exclusions_cancel_inert_intent(reason):
+  owner, now, _old, _before, _future = active()
+  owner.interrupt('acknowledged_trial_profile_withdrawn')
+  assert owner.paused_intent is not None
+  owner.interrupt(reason)
+  assert owner.paused_intent is None
+  rollback_ack(owner, now+100_000_000)
+  assert owner.offer_state['phase'] == 'rollback' and owner.offer_state['accepted']
+
+
+def test_identity_change_and_original_deadline_never_resume_intent():
+  owner, now, _old, _before, _future = active()
+  owner.interrupt('acknowledged_trial_profile_withdrawn')
+  owner.interrupt('source_epoch_changed', context_changed=True)
+  assert owner.paused_intent is None
+  owner, now, old, _before, _future = active()
+  owner.interrupt('acknowledged_trial_profile_withdrawn')
+  rollback_ack(owner, old['deadline_ns']+100_000_000)
+  assert owner.offer_state['phase'] == 'rollback' and owner.offer_state['accepted']
+  assert owner.paused_intent is None
+
+
+def test_stale_baseline_ack_does_not_resume_inert_intent():
+  owner, now, _old, _before, _future = active()
+  owner.interrupt('acknowledged_trial_profile_withdrawn')
+  pending = owner.offer(now)
+  assert pending is not None
+  assert not owner.acknowledge_application(trial_id=pending.trial_id, target_sequence=pending.target_sequence,
+    phase='rollback', target_sha256=pending.target_sha256, applied_full30=pending.target,
+    control_ns=now+200_000_000, binding=pending.binding, now_ns=now+200_000_000, status='applied')
+  assert owner.offer_state['phase'] == 'rollback' and owner.paused_intent is not None
+
+
+"""Input rejection and inert study recovery; no mocked native authority."""
+from dataclasses import replace
+import pytest
+from openpilot.starpilot.flm.tests.test_fullrate_pipeline import snapshots, SyntheticAdapter, BASE
+from openpilot.starpilot.flm.fullrate_collector import FullrateCollector
+from openpilot.starpilot.flm.live_tracking_response import response_witness_value
+
+
+@pytest.mark.parametrize('error', (.1, .201))
+def test_real_original_witness_expiry_recovers_complete_scores_only_after_new_baseline_ack(error):
+  owner, _now, old, before, future = active(error=error)
+  witness = response_witness_value(owner.response_witness)
+  now = witness.valid_until_ns + 1
+  assert owner._witness_expired(now)
+  assert owner.offer(now) is None
+  assert owner.paused_intent is not None and owner.offer_state['phase'] == 'rollback'
+  assert owner.before == () and owner.future == [] and owner.settled_ns is None
+  for reason in ('learning_evidence_delayed', 'stale_control', 'stale_applied_output', 'current_evidence_delayed',
+                 'fresh_physical_residual_window_required'):
+    owner.interrupt(reason)
+    assert owner.paused_intent is not None
+  rollback_ack(owner, now+100_000_000)
+  resumed = owner.offer_state
+  assert resumed['phase'] == 'trial' and resumed['trial_id'] != old['trial_id']
+  assert resumed['target_sequence'] > old['target_sequence'] and not resumed['published']
+  assert owner.before == before and tuple(owner.future) == future and owner.settled_ns is None
+  for name in ('target', 'baseline', 'started_ns', 'deadline_ns', 'alpha', 'required_episodes', 'look_quotas'):
+    assert resumed[name] == old[name]
+
+
+@pytest.mark.parametrize('field,value', [('independent_poses', 0), ('binding', None), ('last_pose_ns', -1)])
+def test_malformed_original_witness_never_classified_as_expiry(field, value):
+  owner, _now, _old, _before, _future = active()
+  original = response_witness_value(owner.response_witness)
+  owner.response_witness = replace(original, **{field: value})
+  assert not owner._witness_expired(original.valid_until_ns+1)
+  assert owner.offer(original.valid_until_ns+1) is None
+  assert owner.paused_intent is None
+
+
+class Interruptions:
+  def __init__(self):
+    self.rows = []
+
+  def interrupt(self, reason, *, context_changed=False):
+    self.rows.append((reason, context_changed))
+
+
+def collector():
+  result = FullrateCollector(SyntheticAdapter())
+  events = Interruptions()
+  result.live_tracking = events
+  return result, events
+
+
+@pytest.mark.parametrize('field,value,reason', [
+  ('fault', False, 'stale_control'), ('fault', True, 'ineligible_control'),
+  ('manual', True, 'ineligible_control'), ('override', True, 'ineligible_control'),
+  ('valid', False, 'ineligible_control'), ('identity', 'bad', 'invalid_control'),
+  ('state_ns', BASE+1, 'invalid_control'), ('requested', float('nan'), 'invalid_control'),
+])
+def test_old_control_fault_and_structural_exclusions_take_precedence(field, value, reason):
+  c, events = collector()
+  control, _output, _pose = snapshots(BASE)
+  assert not c.control(replace(control, **{field: value}), BASE+150_000_000)
+  assert c.reason == reason and events.rows[-1][0] == reason
+  assert not c.controls and not c.ready
+
+
+@pytest.mark.parametrize('field,value,reason', [
+  ('fault', False, 'stale_applied_output'), ('fault', True, 'ineligible_applied_output'),
+  ('override', True, 'ineligible_applied_output'), ('active', False, 'ineligible_applied_output'),
+  ('applied', float('nan'), 'invalid_applied_output'), ('domain', 'wrong', 'consumed_source_mismatch'),
+])
+def test_old_output_fault_and_structural_exclusions_take_precedence(field, value, reason):
+  c, events = collector()
+  _control, output, _pose = snapshots(BASE)
+  assert not c.output(replace(output, **{field: value}), BASE+150_000_000)
+  assert c.reason == reason and events.rows[-1][0] == reason
+  assert not c.applied and not c.ready
+
+
+@pytest.mark.parametrize('field,value', [('independent_poses', 0), ('binding', None), ('last_pose_ns', -1)])
+def test_malformed_witness_after_a_pause_cancels_the_inert_study(field, value):
+  owner, _now, _old, _before, _future = active()
+  original = response_witness_value(owner.response_witness)
+  owner.offer(original.valid_until_ns+1)
+  assert owner.paused_intent is not None
+  owner.response_witness = replace(original, **{field: value})
+  assert owner.offer(original.valid_until_ns+2) is None
+  assert owner.paused_intent is None
+
+
+def test_explicit_native_refusal_is_not_classified_as_input_delay():
+  c, events = collector()
+  control, _output, _pose = snapshots(BASE)
+  control = replace(control, native_required=True, native_allowed=False, native_identity='4'*64,
+    native_observed_ns=BASE-50_000_000, native_expires_ns=BASE+150_000_000)
+  assert not c.control(control, BASE+150_000_000)
+  assert c.reason == 'original_native_admission_unavailable'
+  assert events.rows[-1][0] == c.reason and not c.ready
+
+
+"""Real current-authority parser, declared original CC and output delivery seam."""
+from types import SimpleNamespace
+
+import pytest
+from openpilot.starpilot.flm import steerlearnerd
+from openpilot.starpilot.flm.tests.test_native_health import toyota_cp, typed_state
+from openpilot.starpilot.flm.runtime_adapter_registry import current_authority
+
+
+# Original timestamps retained from the full-cycle native-join refusal. The
+# current wire/CP are declared unit inputs, not reconstruction of a lost wire.
+OBSERVED, CONTROL, APPLY, EXPIRES, RECEIVED = 8549444345139, 8549483378931, 8549484378931, 8549644345139, 8549490482223
+
+
+def original_control(identity):
+  return SimpleNamespace(native_required=True, native_allowed=True, native_identity=identity,
+    native_observed_ns=OBSERVED, control_ns=CONTROL, native_expires_ns=EXPIRES)
+
+
+class Delivery:
+  def __init__(self, control, storage):
+    self.controls = {} if storage == 'recent' else {control.control_ns: control}
+    self.recent_controls = {control.control_ns: control} if storage == 'recent' else {}
+    self._learning_current_native_allowed = False  # Real reset/empty-ready condition.
+    self.calls = []
+
+  def output(self, value, now_ns):
+    self.calls.append((value, now_ns, self._learning_current_native_allowed))
+    return self._learning_current_native_allowed
+
+
+@pytest.mark.parametrize('storage', ('pending', 'recent'))
+@pytest.mark.parametrize('fault', ('none', 'current_denied', 'current_stale', 'record_denied', 'record_stale', 'identity'))
+def test_each_output_rechecks_its_actual_consumed_control_against_real_current_native(storage, fault):
+  cp = toyota_cp()
+  healthy = typed_state(cp, RECEIVED-1_000_000)
+  allowed, _native, identity = current_authority(healthy, cp, now_ns=RECEIVED, required=True)
+  assert allowed
+  control = original_control(identity)
+  if fault == 'record_denied':
+    control.native_allowed = False
+  elif fault == 'record_stale':
+    control.native_expires_ns = CONTROL-1
+  elif fault == 'identity':
+    control.native_identity = '0'*64
+  sm = typed_state(cp, RECEIVED-1_000_000, denied=True) if fault == 'current_denied' else healthy
+  collector = Delivery(control, storage)
+  output = SimpleNamespace(control_ns=control.control_ns, apply_ns=APPLY)
+  now = RECEIVED+300_000_000 if fault == 'current_stale' else RECEIVED
+  # Old direct delivery inherits the false verdict even with a healthy actual CC.
+  assert collector.output(output, now) is False
+  result = steerlearnerd._collect_applied_output(collector, output, sm, cp, now)
+  assert result is (fault == 'none')
+  assert collector.calls[-1] == (output, now, fault == 'none')
+  assert control.control_ns == output.control_ns  # No timestamp is rewritten.
+
+
+def test_missing_consumed_control_cannot_inherit_another_frames_healthy_native_verdict():
+  cp = toyota_cp()
+  sm = typed_state(cp, RECEIVED-1_000_000)
+  allowed, _native, identity = current_authority(sm, cp, now_ns=RECEIVED, required=True)
+  assert allowed
+  control = original_control(identity)
+  collector = Delivery(control, 'recent')
+  collector._learning_current_native_allowed = True
+  output = SimpleNamespace(control_ns=control.control_ns+1)
+  assert steerlearnerd._collect_applied_output(collector, output, sm, cp, RECEIVED) is False
+  assert collector.calls[-1] == (output, RECEIVED, False)
diff --git a/prebuilt.json b/prebuilt.json
index caae5ee738..558af752bc 100644
--- a/prebuilt.json
+++ b/prebuilt.json
@@ -1 +1 @@
-{"protected_inventory_sha256": "41166c6e38cd33eae4c4daedfaa19b5714d0d99ebbc6bec425931304c8667c0e", "schema_version": 1}
+{"protected_inventory_sha256": "f29a00c7be8a21d34adb6a9262fe9786de0f9648a9fa30e91d880b036c77035d", "schema_version": 1}