Keep FLM trial input tapes current between motion deliveries

openpilot/starpilot/flm/steerlearnerd.py

@@ -599,8 +599,9 @@ def run(
     while True:
       state = (live_actor.ingress.optimizer.offer_state
                if live_actor is not None and live_actor.ingress is not None else None)
-      defer_gc = bool(gc_initially_enabled and state is not None and state['phase'] == 'trial' and
-                      not state.get('accepted', False) and clock() <= state['deadline_ns'])
+      pending_trial = bool(state is not None and state['phase'] == 'trial' and
+                           not state.get('accepted', False) and clock() <= state['deadline_ns'])
+      defer_gc = gc_initially_enabled and pending_trial
       if defer_gc and not trial_gc_deferred:
         # This separate optional process must receive the real 100Hz tape during
         # the existing bounded trial. Reference counting continues normally;
@@ -619,8 +620,10 @@ def run(
         pending_stamp = getattr(pending_poses[0], 'mono_ns', 0)
         pending_now = clock()
         pending_fresh = type(pending_stamp) is int and 0 < pending_stamp <= pending_now <= pending_stamp + MAX_GAP_NS
+      # Keep the genuine tapes current between poses during a bounded trial;
+      # otherwise each motion arrival starts behind another full FIFO batch.
       delivered_pose_header = (source_delivery.next(sm, pending_evidence=(control_socket, applied_socket))
-                               if pending_fresh else source_delivery.next(sm))
+                               if pending_fresh or pending_trial else source_delivery.next(sm))
       source_rejected = False
       now = clock()
       # This is an ORIGINAL header from the real poll or an already consumed

openpilot/starpilot/flm/tests/test_pending_motion_publication.py

@@ -27,7 +27,7 @@ def test_absent_pending_pose_does_not_start_extra_observation():
   assert not steerlearnerd._pending_pose_bracket_ready(S(), S(state_ns=BASE), BASE)
 
 
-@pytest.mark.parametrize('missing', ['state_bracket', 'native_pair', 'delivery_in_batch', 'delivery_received_by_native_refresh'])
+@pytest.mark.parametrize('missing', ['state_bracket', 'native_pair', 'delivery_in_batch', 'delivery_received_by_native_refresh', 'pending_trial_ingest'])
 @pytest.mark.parametrize('retained_header', [False, True])
 def test_actual_callbacks_publish_new_witness_before_remaining_fifo_work(monkeypatch, missing, retained_header):
   now = [BASE-25_000_000]
@@ -94,6 +94,8 @@ def test_actual_callbacks_publish_new_witness_before_remaining_fifo_work(monkeyp
     def tick(self, *args, **kwargs):
       self.calls += 1
       assert self.calls == 1
+      if missing == 'pending_trial_ingest':
+        self.ingress = S(optimizer=S(offer_state={'phase':'trial', 'accepted':False, 'deadline_ns':old.deadline_ns}))
       return old
     def mark_publication(self, value):
       self.published.append(value.sequence)
@@ -104,6 +106,17 @@ def test_actual_callbacks_publish_new_witness_before_remaining_fifo_work(monkeyp
       return True
 
   actor = Actor()
+  control_tape, output_tape = object(), object()
+  original_next = steerlearnerd._SourceDelivery.next
+  def observe_wait(delivery, sm, *, pending_evidence=()):
+    if missing == 'pending_trial_ingest' and actor.calls:
+      assert not collector._original_source_pose_queue
+      assert pending_evidence == (control_tape, output_tape)
+      # The fixture owns the arrival schedule below; actual socket polling is
+      # separately tested. This assertion observes the real daemon's API call.
+      return original_next(delivery, sm)
+    return original_next(delivery, sm, pending_evidence=pending_evidence)
+  monkeypatch.setattr(steerlearnerd._SourceDelivery, 'next', observe_wait)
   fitter = S(cancel=lambda *args:None, close=lambda *args:True)
   history = S(offer=None, requalifier=None, reset=lambda:None, advance=lambda *args, **kwargs:None,
     observe_response_recovery=lambda *args:False)
@@ -158,7 +171,7 @@ def test_actual_callbacks_publish_new_witness_before_remaining_fifo_work(monkeyp
     raise Stop
   monkeypatch.setattr(steerlearnerd, 'drain_pair', drain)
   with pytest.raises(Stop):
-    steerlearnerd.run(S(), SM(), object(), object(), S(send=send), lambda *args, **kwargs:adapter,
+    steerlearnerd.run(S(), SM(), control_tape, output_tape, S(send=send), lambda *args, **kwargs:adapter,
       None, source, lambda *args:('objective',), lambda *args:'objective', None, None, fitter, None,
       clock=lambda:now[0], live_actor=actor)
   fresh_sends = [stamp for stamp, version, sequence in sent if version == 1 and sequence == 2]

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "f0549dd2e30990df108d65ab4321b4671fcf80b9b1564d47ed03b62892cb810b", "schema_version": 1}
+{"protected_inventory_sha256": "c05bcc6b9f084bfe824655bcd33496f3ef67827c22721fb29e70af3c3e151caa", "schema_version": 1}