fix(qnn): soft-reset temporal history without reconnecting

openpilot/sunnypilot/modeld_v2/external_host/faults.py

@@ -11,6 +11,10 @@ class HistoryResetRequired(HostError):
   """An input was not executed on the required sequence/cadence."""
 
 
+class TransientBackpressure(HostError):
+  """Local admission is full; drop this frame without resetting the transport."""
+
+
 class FatalModelFault(HostError):
   """Model identity, schema, output or protocol integrity was not valid."""
 

openpilot/sunnypilot/modeld_v2/external_host/opif/transport.py

@@ -20,6 +20,7 @@ class Frame:
   action_t: tuple[float, float] = (0., 0.)
   timestamp_ns: int = 0
   context: object = None
+  reset_temporal: bool = False
 
 
 @dataclass
@@ -73,7 +74,7 @@ class Pipeline:
     if len(frame.tensor) != self.session.tensor_bytes:
       raise ValueError("incorrect current6 tensor size")
     prefix = wire.INFERENCE_PREFIX_STRUCT.pack(
-      int(self.session.layout), 0, frame.desire, frame.traffic, int(self.next_id == 1),
+      int(self.session.layout), 0, frame.desire, frame.traffic, int(self.next_id == 1 or frame.reset_temporal),
       *frame.action_t, frame.timestamp_ns, len(frame.tensor), len(frame.tensor))
     request = wire.encode_message(wire.MessageType.INFERENCE_REQUEST, self.session.session_id,
                                   self.next_id, prefix + frame.tensor, self.session.maximum_datagram_bytes)

openpilot/sunnypilot/modeld_v2/external_host/opif_client.py

@@ -112,7 +112,7 @@ class OpifLink:
       self.close()
       raise classify(error) from error
 
-  def submit(self, frame_id, eof, pixels, scalars):
+  def submit(self, frame_id, eof, pixels, scalars, reset_temporal=False):
     if self.error is not None:
       raise self.error
     if not self.ready or self.stopped.is_set():
@@ -140,7 +140,7 @@ class OpifLink:
       self.sent[seq] = (int(frame_id), int(eof), captured)
       if self.stage_enabled:
         self.stages[seq] = {'t_link_submit_ns': time.clock_gettime_ns(time.CLOCK_BOOTTIME)}
-    frame = Frame(bytes(pixels), captured, desire, traffic, tuple(values[10:]), int(eof), (seq, int(frame_id), int(eof)))
+    frame = Frame(bytes(pixels), captured, desire, traffic, tuple(values[10:]), int(eof), (seq, int(frame_id), int(eof)), bool(reset_temporal))
     try:
       self.pending.put_nowait(frame)
     except queue.Full as e:
@@ -180,7 +180,7 @@ class OpifLink:
               'runtime_identity_verified': False, 'result_schema': self.result_schema,
               'opif_session_id': self.pipeline.session.session_id, 'state_epoch': 1,
               'opif_state_epoch': 1, 'request_id': seq, 'opif_request_id': seq,
-              'frame_id': fid, 'eof_ns': eof,
+              'frame_id': fid, 'eof_ns': eof, 'soft_reset': bool(frame.reset_temporal),
               'terminal': 'expired' if time.monotonic() - frame.captured >= self.frame_timeout else 'executed'}
           if self.stage_enabled:
             stats.update(self.stages.pop(seq, {}))

openpilot/sunnypilot/modeld_v2/external_host/qnn_hotstandby.py

@@ -201,8 +201,17 @@ class QnnHotStandby:
       identity = (event.get('opif_session_id'), event.get('state_epoch'))
       if self.history_identity is None:
         self.history_identity = identity
-      if identity != self.history_identity:
-        raise HostError('Remote execution history epoch changed without requalification')
+      elif identity != self.history_identity:
+        if event.get('soft_reset') is not True:
+          raise HostError('Remote execution history epoch changed without requalification')
+        self.history_identity = identity
+      if event.get('soft_reset') is True:
+        # The client-side admission window requested RESET_TEMPORAL on the
+        # existing TCP session. Accept that sanctioned reset while preserving
+        # the transport and restarting only continuity/qualification.
+        self.last_remote_seq = self.last_remote_frame = None
+        self.good_frames = 0
+        self.last_eligible_at = None
       seq, frame = event['seq'], event['frame_id']
       if event['terminal'] not in ('executed', 'expired'):
         raise HostError('Remote input execution was not confirmed')

openpilot/sunnypilot/modeld_v2/external_host/qnn_model.py

@@ -8,7 +8,7 @@ import numpy as np
 
 from .session import AsyncSession
 from .wire import HostError
-from .faults import TransportUnavailable, HistoryResetRequired
+from .faults import TransportUnavailable, HistoryResetRequired, TransientBackpressure
 from .qnn_reconnect import RecoveryWorker
 
 SOURCE_SHA = '404a18cfd86d29637d20c697dfde245bb47c666ae016730ab674c65f4d1e1aa4'
@@ -295,13 +295,10 @@ def create_model(base_class, config_path, cam_w, cam_h, progress=None):
           raise HostError('QNN preprocessing exceeded the capture deadline')
         try:
           self.async_session.submit(meta.frame_id, meta.timestamp_eof, pixels.tobytes(), scalars, context)
-        except HistoryResetRequired:
-          # A full local admission window is transient backpressure, not a
-          # transport failure. Keep the TCP session alive and drop this frame;
-          # the next frame can use the freed sequence slot without making the
-          # phone UI report a disconnect/reconnect cycle.
-          if self.async_session.error is not None or self.async_session.stopped:
-            raise
+        except TransientBackpressure:
+          # Local admission pressure is not a transport fault. The session
+          # keeps its TCP connection and schedules a RESET_TEMPORAL bit on the
+          # first frame submitted after the old in-flight window drains.
           if after_enqueue:
             after_enqueue()
           return None

openpilot/sunnypilot/modeld_v2/external_host/session.py

@@ -5,7 +5,7 @@ import time
 from collections import deque
 from .wire import HostError
 from .priority import background_thread
-from .faults import classify, HistoryResetRequired
+from .faults import classify, HistoryResetRequired, TransientBackpressure
 
 class AsyncSession:
   """Frame identities and their capture-time context stay paired until publication."""
@@ -28,6 +28,8 @@ class AsyncSession:
     self.terminal_events = deque(maxlen=256)
     self.terminal_history = deque(maxlen=256)
     self.state_identity = None
+    self.soft_reset_pending = False
+    self.awaiting_soft_reset_epoch = False
     self.threads=[threading.Thread(target=self._send,daemon=True),threading.Thread(target=self._receive,daemon=True)]
     for thread in self.threads:thread.start()
 
@@ -52,16 +54,33 @@ class AsyncSession:
 
   def submit(self, frame_id,eof,pixels,scalars,context):
     self.check()
+    reset_temporal = False
     with self.condition:
-      if len(self.pending)>=self.capacity:
-        error_type=HistoryResetRequired if self.soft_expiry else HostError
-        raise error_type(f'External model has {self.capacity} outstanding frames')
+      if self.soft_reset_pending:
+        # Let every old request complete before the client-side temporal reset
+        # bit is sent. This keeps one TCP session alive through backpressure.
+        if self.pending or not self.outgoing.empty():
+          raise TransientBackpressure('External model is draining for a soft history reset')
+        self.soft_reset_pending = False
+        self.awaiting_soft_reset_epoch = True
+        reset_temporal = True
+      elif len(self.pending)>=self.capacity or self.outgoing.full():
+        if self.soft_expiry:
+          self.soft_reset_pending = True
+          raise TransientBackpressure(f'External model has {self.capacity} outstanding frames')
+        raise HostError('External result exceeded the local admission window')
       seq=self.next_seq;self.next_seq+=1
       self.pending[seq]=(time.monotonic(),context,int(frame_id),int(eof))
-    try:self.outgoing.put_nowait((seq,frame_id,eof,pixels,scalars))
-    except queue.Full as error:
-      error_type=HistoryResetRequired if self.soft_expiry else HostError
-      raise error_type('External send queue is full') from error
+      try:
+        item=(seq,frame_id,eof,pixels,scalars,reset_temporal) if self.soft_expiry else (seq,frame_id,eof,pixels,scalars)
+        self.outgoing.put_nowait(item)
+      except queue.Full as error:
+        self.pending.pop(seq,None)
+        self.next_seq -= 1
+        if self.soft_expiry:
+          self.soft_reset_pending = True
+          raise TransientBackpressure('External send queue is full') from error
+        raise HostError('External send queue is full') from error
 
   def poll(self,timeout=0):
     self.check()
@@ -90,7 +109,14 @@ class AsyncSession:
       identity=(stats.get('opif_session_id'),stats.get('state_epoch'))
       if any(type(value) is not int or value<=0 for value in identity):raise HostError('Missing OPIF state identity')
       if self.state_identity is None:self.state_identity=identity
-      if identity!=self.state_identity:raise HostError('OPIF state epoch changed inside an active session')
+      if self.state_identity is None:
+        self.state_identity=identity
+      elif identity!=self.state_identity:
+        if not self.awaiting_soft_reset_epoch or stats.get('soft_reset') is not True:
+          raise HostError('OPIF state epoch changed inside an active session')
+        self.state_identity=identity
+      if stats.get('soft_reset') is True:
+        self.awaiting_soft_reset_epoch=False
       if stats.get('request_id')!=seq:raise HostError('OPIF request identity differs from session sequence')
       expired=source_age>=self.maximum_age or stats.get('terminal')=='expired'
       terminal='expired' if expired else 'executed'
@@ -98,6 +124,7 @@ class AsyncSession:
       received_age = (received_ns-eof)/1e9 if type(received_ns) is int else None
       event={'seq':seq,'request_id':seq,'frame_id':fid,'eof_ns':eof,'context':context,
              'opif_session_id':identity[0],'state_epoch':identity[1],
+             'soft_reset':bool(stats.get('soft_reset',False)),
              'opif_state_epoch':identity[1],'opif_request_id':seq,
              'terminal':terminal,'received_age_seconds':received_age,'consumed_age_seconds':source_age}
       if len(self.terminal_events)>=256:raise HostError('Unconsumed OPIF terminal event queue is full')
@@ -125,9 +152,17 @@ class AsyncSession:
     background_thread()
     try:
       while not self.stopped:
-        try:seq,fid,eof,pixels,scalars=self.outgoing.get(timeout=.05)
+        try:item=self.outgoing.get(timeout=.05)
         except queue.Empty:continue
-        actual=self.link.submit(fid,eof,pixels,scalars)
+        if len(item)==6:
+          seq,fid,eof,pixels,scalars,reset_temporal=item
+        else:
+          seq,fid,eof,pixels,scalars=item
+          reset_temporal=False
+        if reset_temporal:
+          actual=self.link.submit(fid,eof,pixels,scalars,reset_temporal=True)
+        else:
+          actual=self.link.submit(fid,eof,pixels,scalars)
         if actual!=seq:raise HostError('Sender sequence diverged')
         with self.condition:self.condition.notify_all()
     except Exception as error:self._fail(error)