Keep source learning progressing after interrupted trials

openpilot/starpilot/flm/feedback_observation.py

@@ -197,7 +197,7 @@ def card_capture_for(owner, now_ns):
     # ignores it there. Observational capture must check its REAL bits directly.
     service = 'autoSteeringLearning'
     stamp = int(sm.logMonoTime[service])
-    receipt = int(sm.recv_time[service]*1e9)
+    receipt = int(round(sm.recv_time[service]*1e9))
     if (type(now_ns) is not int or not now_ns > 0 or
         not all((sm.seen[service],sm.valid[service],sm.alive[service],sm.freq_ok[service])) or
         not 0 < stamp <= now_ns <= stamp+500_000_000 or
@@ -234,7 +234,7 @@ def capture_controls(owner, cc, *, control_ns, input_limited):
   sm = owner.sm
   header = int(sm.logMonoTime['carOutput'])
   output = sm['carOutput'].actuatorsOutput
-  row = CoConsumption(1,control_ns,header,int(sm.recv_frame['carOutput']),int(sm.recv_time['carOutput']*1e9),
+  row = CoConsumption(1,control_ns,header,int(sm.recv_frame['carOutput']),int(round(sm.recv_time['carOutput']*1e9)),
     bool(sm.seen['carOutput']),bool(sm.valid['carOutput']),co_body(header,bool(sm.valid['carOutput']),output.to_dict()),
     f32(float(output.torque)),f32(float(cc.actuators.torque)),bool(cc.latActive),bool(input_limited),
     bool(owner.steer_limited_by_safety),int(getattr(owner,'learning_feedback_prior_publish_ns',0)),

openpilot/starpilot/flm/global_runtime.py

@@ -153,7 +153,7 @@ class GlobalRuntime:
     try:
       if (type(now_ns) is int and now_ns > 0 and self.descriptor is not None and
           sm.seen[receipt_service] and sm.alive[receipt_service] and sm.valid[receipt_service] and sm.freq_ok[receipt_service] and
-          0 <= now_ns - int(sm.recv_time[receipt_service] * 1e9) <= SOURCE_LEASE_NS):
+          0 <= now_ns - int(round(sm.recv_time[receipt_service] * 1e9)) <= SOURCE_LEASE_NS):
         envelope = sm[receipt_service]
         receipt = envelope.sourceReceipt if live_delivery else envelope.learning.sourceReceipt
         if receipt.version in (2, 3):
@@ -164,7 +164,7 @@ class GlobalRuntime:
             early_source = read_source_receipt(SimpleNamespace(valid=True, autoSteeringLearning=envelope),
               now_ns=now_ns, cp_sha256=self.descriptor.cp_sha256)
       stamp = int(sm.logMonoTime['extrinsicsCalibration'])
-      receipt = int(sm.recv_time['extrinsicsCalibration'] * 1e9)
+      receipt = int(round(sm.recv_time['extrinsicsCalibration'] * 1e9))
       current_calibration = bool(type(now_ns) is int and now_ns > 0 and
         sm.all_checks(['extrinsicsCalibration']) and
         0 < stamp <= now_ns <= stamp + 500_000_000 and
@@ -268,7 +268,7 @@ class GlobalRuntime:
       )
       for service, maximum in age_limits:
         stamp = int(sm.logMonoTime[service])
-        receipt = int(sm.recv_time[service] * 1e9)
+        receipt = int(round(sm.recv_time[service] * 1e9))
         if not (sm.seen[service] and 0 < stamp <= now_ns <= stamp + maximum and 0 < receipt <= now_ns <= receipt + maximum):
           if service == 'extrinsicsCalibration' and hasattr(self, '_calibration_lifecycle'):
             self._calibration_lifecycle.unavailable(now_ns=now_ns)
@@ -277,7 +277,7 @@ class GlobalRuntime:
         raise ValueError('controller_saturation')
       if not (sm.seen[receipt_service] and sm.alive[receipt_service] and sm.valid[receipt_service] and sm.freq_ok[receipt_service]):
         raise ValueError('source_unavailable')
-      received = int(sm.recv_time[receipt_service] * 1e9)
+      received = int(round(sm.recv_time[receipt_service] * 1e9))
       if not 0 <= now_ns - received <= SOURCE_LEASE_NS:
         raise ValueError('stale_receipt')
       envelope = sm[receipt_service]

openpilot/starpilot/flm/learning_source.py

@@ -176,7 +176,7 @@ class AutomaticLearningConsumer:
     try:
       if not (sm.seen[SERVICE] and sm.alive[SERVICE] and sm.valid[SERVICE] and sm.freq_ok[SERVICE]):
         return None
-      received_ns = int(sm.recv_time[SERVICE] * 1e9)
+      received_ns = int(round(sm.recv_time[SERVICE] * 1e9))
       if not 0 <= now_ns - received_ns <= MAX_AGE_NS:
         return None
       state = learning_payload(sm[SERVICE])

openpilot/starpilot/flm/live_tracking.py

@@ -643,8 +643,12 @@ class LiveTrackingOptimizer:
     hint = self._current_selector(now_ns)
     if hint is None:
       return False
-    if episode.start_ns <= self.selector_after_ns or not _matches_demand(hint['demand'], episode):
-      return True
+    if episode.start_ns <= self.selector_after_ns:
+      return self.rmse is not None
+    if not _matches_demand(hint['demand'], episode):
+      # A source recovery hint is a preference, not a collection lock. A
+      # different original maneuver can seed a new independent experiment.
+      return self.rmse is not None
     from openpilot.starpilot.flm.live_tracking_response import validate_response_witness
 
     try:
@@ -658,11 +662,11 @@ class LiveTrackingOptimizer:
     expected = self._fresh_selected_effect((episode,), hint['name'], target)
     if expected is None or expected <= 0.0001:
       self.status, self.reason = 'collecting', 'fresh_selected_native_probe_required'
-      return True
+      return self.rmse is not None
     allocation = self._comparison_allocation(episode, expected, target, minimum_episodes=hint.get('minimum_episodes'))
     if allocation is None:
       self.status, self.reason = 'collecting', 'fresh_recovery_allocation_power_insufficient'
-      return True
+      return self.rmse is not None
     # This fresh source/probe observation is REVALIDATION, never a score row.
     # Old selection provenance/expiry is preserved, not retagged as new proof.
     # The actual baseline later gets ONE unchanged power check; its honest

openpilot/starpilot/flm/native_baseline.py

@@ -57,7 +57,7 @@ def native_source(sm, cp, now_ns):
     name = 'lateralTorqueParameters'
     if type(now_ns) is not int or now_ns <= 0 or not sm.all_checks([name]):
       return None
-    stamp, received = int(sm.logMonoTime[name]), int(sm.recv_time[name] * 1e9)
+    stamp, received = int(sm.logMonoTime[name]), int(round(sm.recv_time[name] * 1e9))
     row = sm[name]
     values = (float(row.latAccelFactorFiltered), float(row.latAccelOffsetFiltered), float(row.frictionCoefficientFiltered))
     base = cp.lateralTuning.torque

openpilot/starpilot/flm/runtime_adapter_registry.py

@@ -149,7 +149,7 @@ def current_authority(sm, cp, *, now_ns, required):
         and sm.alive['pandaStates']
         and sm.valid['pandaStates']
         and sm.freq_ok['pandaStates']
-        and 0 <= now_ns - int(sm.recv_time['pandaStates'] * 1e9) <= 1_000_000_000
+        and 0 <= now_ns - int(round(sm.recv_time['pandaStates'] * 1e9)) <= 1_000_000_000
       )
       pandas = sm['pandaStates']
       if len(pandas) != len(cp.safetyConfigs):
@@ -202,7 +202,7 @@ def current_ford_authority(sm, cp, *, now_ns):
       raise ValueError('Ford downstream role')
     pandas = sm['pandaStates']
     stamp = int(sm.logMonoTime['pandaStates'])
-    receipt = int(sm.recv_time['pandaStates'] * 1e9)
+    receipt = int(round(sm.recv_time['pandaStates'] * 1e9))
     boot = time.clock_gettime_ns(time.CLOCK_BOOTTIME)
     allowed = bool(
       sm.seen['pandaStates']
@@ -622,7 +622,7 @@ def physical_observer(sm, cp, collector, now_ns, *, distinguish_pending=False):
       collector._physical_seen_pose = pose_ns
       evidence = PhysicalEvidence(
         pose_ns,
-        int(sm.recv_time['deviceMotion'] * 1e9),
+        int(round(sm.recv_time['deviceMotion'] * 1e9)),
         generation,
         digest((float(cp.steerRatio), float(cp.wheelbase), float(cp.centerToFront), float(cp.mass), collector._physical_clock, native_epoch, regime)),
         digest((lag,)), lag, 0.0,
@@ -654,7 +654,7 @@ def physical_observer(sm, cp, collector, now_ns, *, distinguish_pending=False):
       # clocks; this is NEVER a new physical sample or a current-fit grant.
       waiting = queue[0] if queue else None
       return PendingPhysical(pose_ns if waiting is None else waiting.mono_ns,
-        int(sm.recv_time['deviceMotion'] * 1e9) if waiting is None else waiting.received_ns, context_changed)
+        int(round(sm.recv_time['deviceMotion'] * 1e9)) if waiting is None else waiting.received_ns, context_changed)
     return None
   except (AttributeError, KeyError, TypeError, ValueError, OverflowError):
     return reject()
@@ -726,7 +726,7 @@ def original_source_pose_observer(sm, cp, collector, now_ns, *, distinguish_pend
     ):
       return reject('pose_source_or_calibration_flags')
     if equivalence is not None:
-      source_lower, source_upper = source_interval(int(dm.timestamp), int(sm.recv_time['deviceMotion'] * 1e9), now_ns, equivalence)
+      source_lower, source_upper = source_interval(int(dm.timestamp), int(round(sm.recv_time['deviceMotion'] * 1e9)), now_ns, equivalence)
     else:
       source_lower = source_upper = pose_ns
     state_ns = int(sm.logMonoTime['carState'])
@@ -792,7 +792,7 @@ def original_source_pose_observer(sm, cp, collector, now_ns, *, distinguish_pend
             bool(panda.controlsAllowed), bool(panda.safetyRxChecksInvalid), bool(panda.heartbeatLost),
             tuple(str(fault) for fault in panda.faults)) for index, panda in enumerate(sm['pandaStates'])),
         )
-        ordinary_safety = ('ordinary-panda-current-v1', cp_sha, int(sm.recv_time['pandaStates'] * 1e9),
+        ordinary_safety = ('ordinary-panda-current-v1', cp_sha, int(round(sm.recv_time['pandaStates'] * 1e9)),
           int(sm.logMonoTime['pandaStates']), configuration)
       native = bool(
         ordinary_allowed and
@@ -850,7 +850,7 @@ def original_source_pose_observer(sm, cp, collector, now_ns, *, distinguish_pend
       pose = calibrator.build_calibrated_pose(Pose.from_device_motion(dm))
       collector._original_source_seen_pose = pose_ns
       evidence = OriginalSourcePose(
-        mono_ns=pose_ns, received_ns=int(sm.recv_time['deviceMotion'] * 1e9), observed_ns=now_ns,
+        mono_ns=pose_ns, received_ns=int(round(sm.recv_time['deviceMotion'] * 1e9)), observed_ns=now_ns,
         source_boot_ns=int(dm.timestamp) if contract.source_domain == 'boottime' else 0, clock_offset_ns=clock_offset,
         source_timestamp_ns=int(dm.timestamp), source_clock_domain=contract.source_domain,
         source_clock_contract=(contract.source_domain, contract.producer_revision, contract.proof_sha256, contract.scope),
@@ -905,7 +905,7 @@ def original_source_pose_observer(sm, cp, collector, now_ns, *, distinguish_pend
       # clocks; this is NEVER a new physical sample or a current-fit grant.
       waiting = queue[0] if queue else None
       return PendingPhysical(pose_ns if waiting is None else waiting.mono_ns,
-        int(sm.recv_time['deviceMotion'] * 1e9) if waiting is None else waiting.received_ns, context_changed)
+        int(round(sm.recv_time['deviceMotion'] * 1e9)) if waiting is None else waiting.received_ns, context_changed)
     return None
   except (AttributeError, KeyError, TypeError, ValueError, OverflowError):
     return reject('source_observation_exception')

openpilot/starpilot/flm/steerlearnerd.py

@@ -101,7 +101,7 @@ def _configuration_calibration(sm, calibration_lifecycle, now):
   # deliveries. Sample freshness and current native authority are checked
   # independently when admitting physical observations and applying offers.
   stamp = int(sm.logMonoTime['extrinsicsCalibration'])
-  receipt = int(sm.recv_time['extrinsicsCalibration'] * 1e9)
+  receipt = int(round(sm.recv_time['extrinsicsCalibration'] * 1e9))
   if not (sm.seen['extrinsicsCalibration'] and sm.valid['extrinsicsCalibration'] and
           0 < stamp <= now and 0 < receipt <= now):
     return calibration_lifecycle.observe(None, current=False)

openpilot/starpilot/flm/tests/test_source_recovery_progress.py

@@ -0,0 +1,81 @@
+"""Numerical state-machine regressions; no source/native/road authority."""
+from dataclasses import replace
+import unittest
+from openpilot.starpilot.flm.tests.test_tracking_source_context import owner, refresh
+from openpilot.starpilot.flm.tests.test_live_tracking import episode
+from openpilot.starpilot.flm.live_tracking import _demand
+
+
+def pending_recovery():
+  o=owner()
+  old=replace(episode(o,0,'ff_gain_left'),normalization_accel=0.5)
+  o.selector_hint={'group':'ff.left','name':'ff_gain_left','delta':.005,
+    'demand':_demand(old),'binding':o.binding,'profile_sha256':o._hash(o.accepted),
+    'model_revision':o.certificate.model_revision,'controller_domain':o.domain,
+    'selected_ns':old.end_ns,'expires_ns':old.end_ns+600_000_000_000,
+    'source_episode_ids':(old.episode_id,),'quota':1,'minimum_episodes':1}
+  o.selector_pending=True;o.selector_after_ns=old.end_ns
+  return o,old
+
+
+class SourceRecoveryProgress(unittest.TestCase):
+  def test_old_left_hint_does_not_swallow_new_right_windows(self):
+    o,old=pending_recovery()
+    discovery=set()
+    for i in range(2,7):
+      e=replace(episode(o,i,'ff_gain_right'),normalization_accel=.5)
+      refresh(o,e.end_ns);o.observe_episode(e,now_ns=e.end_ns)
+      if i<4:discovery.add(e.episode_id)
+    assert o.offer_state is not None and o.offer_state['names']==('ff_gain_right',)
+    assert len(o.before)==3 and discovery.isdisjoint(e.episode_id for e in o.before)
+    assert not any(o.accepted.values())
+
+  def test_missing_old_native_effect_does_not_hide_another_hook(self):
+    o,old=pending_recovery()
+    for i in range(2,7):
+      e=replace(episode(o,i,'center_taper_max',kind='large'),normalization_accel=.5)
+      refresh(o,e.end_ns);o.observe_episode(e,now_ns=e.end_ns)
+    assert o.offer_state is not None and o.offer_state['names']==('center_taper_max',)
+    assert len(o.before)==3 and not any(o.accepted.values())
+
+  def test_missing_response_still_cannot_resume_or_launch(self):
+    o,old=pending_recovery()
+    e=replace(episode(o,2,'ff_gain_left'),normalization_accel=.5)
+    o.response_witness=None
+    assert o._revalidate_selector(e,e.end_ns)
+    assert o.discovery is None and o.offer_state is None
+    assert o.reason=='fresh_selector_response_required'
+
+class ReceiptClockConversion(unittest.TestCase):
+  def capture(self,publication,receipt,control):
+    from types import SimpleNamespace as NS
+    from openpilot.starpilot.flm.feedback_observation import capture_controls
+    class SM(dict): pass
+    output=NS(torque=0.,to_dict=lambda:{'torque':0.})
+    sm=SM(carOutput=NS(actuatorsOutput=output));sm.logMonoTime={'carOutput':publication}
+    sm.recv_frame={'carOutput':1};sm.recv_time={'carOutput':receipt/1e9}
+    sm.seen={'carOutput':True};sm.valid={'carOutput':True}
+    source=NS(enabled=True,manual=False,observed_ns=control-1,expires_ns=control+100_000_000,source_epoch='a'*64,cp_sha256='b'*64)
+    obj=NS(sm=sm,CP=NS(steerControlType='torque'),global_learning=NS(eligible=True,source=source),steer_limited_by_safety=False)
+    cc=NS(actuators=NS(torque=0.),latActive=True)
+    return capture_controls(obj,cc,control_ns=control,input_limited=False)
+
+  def test_actual_capture_preserves_receipt_order_across_uptime_boundary(self):
+    stamp=1_024_000_000_052
+    assert int(stamp/1e9*1e9)==stamp-1  # Demonstrated original producer defect.
+    row=self.capture(stamp,stamp,stamp+1_000_000)
+    assert row.receipt_ns==stamp and row.publication_ns==stamp
+
+  def test_stale_and_future_receipts_are_still_rejected(self):
+    from openpilot.starpilot.flm.feedback_observation import MAX_AGE_NS
+    stamp=1_024_000_000_052
+    with self.assertRaises(ValueError): self.capture(stamp,stamp,stamp+MAX_AGE_NS+1)
+    with self.assertRaises(ValueError): self.capture(stamp,stamp+2_000_000,stamp+1_000_000)
+
+  def test_recovery_ack_barrier_does_not_hide_new_source_candidates(self):
+    o,old=pending_recovery();o.selector_after_ns=500_000_000_000
+    for i in range(2,7):
+      e=replace(episode(o,i,'ff_gain_right'),normalization_accel=.5)
+      refresh(o,e.end_ns);o.observe_episode(e,now_ns=e.end_ns)
+    assert o.offer_state is not None and o.offer_state['names']==('ff_gain_right',)
+    assert len(o.before)==3 and not any(o.accepted.values())

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "dc8a5c4cba775e8d0b49423febde862ece25381b3d28f46fd91017d8059d48ca", "schema_version": 1}
+{"protected_inventory_sha256": "87c01c16455ae6686d4574153d26e1071ce0b5246b814ee5486d34f839c19ecb", "schema_version": 1}