Qualify temporary Card receive timing on native ARM

Build-only diagnostic branch. Add bounded receive/GIL and Card phase
observations plus tests, and an isolated source-pinned ARM artifact job.
No control-policy change, device deployment, receipt refresh or main
branch update. Artifact adoption requires separate evidence review.

.github/workflows/card_stall_native_artifact.yaml

@@ -0,0 +1,89 @@
+name: Card stall native artifact qualification
+
+on:
+  push:
+    branches: [diagnostics/card-stall-build-20261012]
+    paths:
+      - .github/workflows/card_stall_native_artifact.yaml
+      - tools/ci/card_stall_native_artifact/**
+      - msgq_repo/**
+      - openpilot/selfdrive/car/card.py
+      - openpilot/starpilot/card_stall_timing.py
+      - openpilot/starpilot/aol/tests/test_card_stall_timing.py
+
+permissions:
+  contents: read
+
+concurrency:
+  group: card-stall-native-artifact-${{ github.ref }}
+  cancel-in-progress: false
+
+jobs:
+  native_arm64:
+    if: github.repository == 'firestar5683/StarPilot' && github.ref == 'refs/heads/diagnostics/card-stall-build-20261012'
+    runs-on: ubuntu-24.04-arm
+    timeout-minutes: 25
+    env:
+      PYTHONNOUSERSITE: '1'
+      UV_NO_PROGRESS: '1'
+      ARTIFACT_DIR: ${{ runner.temp }}/card-stall-native-artifact
+    steps:
+      - name: Fetch exact public source with no archive fallback
+        shell: bash
+        env:
+          GIT_TERMINAL_PROMPT: '0'
+          GIT_LFS_SKIP_SMUDGE: '1'
+          GIT_CONFIG_NOSYSTEM: '1'
+          GIT_CONFIG_GLOBAL: /dev/null
+        run: |
+          set -euo pipefail
+          test "$GITHUB_REPOSITORY" = firestar5683/StarPilot
+          test "$GITHUB_REF" = refs/heads/diagnostics/card-stall-build-20261012
+          [[ "$GITHUB_SHA" =~ ^[0-9a-f]{40}$ ]]
+          python3 -c 'import re, subprocess; v = subprocess.check_output(["git", "version"], text=True); m = re.search(r"(\d+)\.(\d+)\.(\d+)", v); assert m and tuple(map(int, m.groups())) >= (2, 28, 0), v; print(v.strip())'
+          test ! -e .git
+          git init
+          git remote add origin https://github.com/firestar5683/StarPilot.git
+          git config remote.origin.promisor true
+          git config remote.origin.partialclonefilter blob:none
+          git sparse-checkout init --no-cone
+          printf '%s\n' '/msgq_repo/' '/openpilot/selfdrive/car/card.py' '/openpilot/starpilot/card_stall_timing.py' '/openpilot/starpilot/aol/tests/test_card_stall_timing.py' '/tools/ci/card_stall_native_artifact/' '/uv.lock' '/pyproject.toml' '/.github/workflows/card_stall_native_artifact.yaml' | git sparse-checkout set --no-cone --stdin
+          git -c protocol.version=2 fetch --filter=blob:none --depth=2 origin "$GITHUB_SHA"
+          test "$(git rev-parse FETCH_HEAD)" = "$GITHUB_SHA"
+          test "$(git rev-parse FETCH_HEAD^)" = e8389b0b7529bad402f1f4e5fbd672ff13d9d382
+          git checkout --detach "$GITHUB_SHA"
+          test ! -e openpilot/selfdrive/modeld/models
+      - name: Verify frozen source and baseline artifacts
+        run: python3 tools/ci/card_stall_native_artifact/verify_source.py "$ARTIFACT_DIR/provenance"
+      - name: Install bounded native build prerequisites
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y --no-install-recommends gcc-13 g++-13 binutils libzmq5
+      - name: Install pinned Python environment manager
+        uses: astral-sh/setup-uv@e92bafb6253dcd438e0484186d7669ea7a8ca1cc # v6.4.3
+        with:
+          version: '0.12.5'
+          enable-cache: false
+      - name: Qualify source, build and execute native ARM tests
+        shell: bash
+        env:
+          EXPECTED_BASE_COMMIT: e8389b0b7529bad402f1f4e5fbd672ff13d9d382
+          EXPECTED_SOURCE_COMMIT: ${{ github.sha }}
+          EXPECTED_GCC_VERSION: '13.3.0'
+          GCC: /usr/bin/gcc-13
+          GXX: /usr/bin/g++-13
+        run: |
+          uv python install 3.12.13
+          uv venv --python 3.12.13 "$RUNNER_TEMP/card-stall-native-venv"
+          export PYTHON="$RUNNER_TEMP/card-stall-native-venv/bin/python"
+          uv pip install --python "$PYTHON" --require-hashes -r tools/ci/card_stall_native_artifact/requirements.lock
+          bash tools/ci/card_stall_native_artifact/build_native.sh "$PWD" "$ARTIFACT_DIR/provenance/baseline-ipc_pyx.so" "$ARTIFACT_DIR/build"
+      - name: Retain artifact and complete qualification evidence
+        if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+        with:
+          name: card-stall-native-arm64-${{ github.sha }}
+          path: ${{ env.ARTIFACT_DIR }}/
+          if-no-files-found: error
+          retention-days: 7
+          compression-level: 6

msgq_repo/msgq/ipc_pyx.pyx

@@ -7,9 +7,49 @@ from libcpp.vector cimport vector
 from libcpp cimport bool
 from libc cimport errno
 from libc.string cimport strerror
+from libc.stdint cimport int64_t, uint64_t
 from cython.operator import dereference
 
 
+cdef extern from "Python.h":
+  unsigned long PyThread_get_thread_ident() noexcept
+
+
+cdef extern from *:
+  """
+  #include <errno.h>
+  #include <stdint.h>
+  #include <time.h>
+
+  // Diagnostics must neither raise nor disturb the backend's errno.
+  static inline void msgq_receive_clock_pair(int64_t *wall, int64_t *cpu) {
+    const int saved_errno = errno;
+    struct timespec ts;
+    *wall = clock_gettime(CLOCK_MONOTONIC, &ts) == 0
+      ? (int64_t)ts.tv_sec * 1000000000LL + ts.tv_nsec : -1;
+    *cpu = clock_gettime(CLOCK_THREAD_CPUTIME_ID, &ts) == 0
+      ? (int64_t)ts.tv_sec * 1000000000LL + ts.tv_nsec : -1;
+    errno = saved_errno;
+  }
+  """
+  void msgq_receive_clock_pair(int64_t *wall, int64_t *cpu) noexcept nogil
+
+
+cdef struct ReceiveTiming:
+  uint64_t sequence
+  bool non_blocking
+  bool returned_message
+  size_t byte_length
+  int64_t before_wall
+  int64_t before_cpu
+  int64_t backend_return_wall
+  int64_t backend_return_cpu
+  int64_t gil_reacquired_wall
+  int64_t gil_reacquired_cpu
+  int64_t python_return_wall
+  int64_t python_return_cpu
+
+
 from .ipc cimport Context as cppContext
 from .ipc cimport SubSocket as cppSubSocket
 from .ipc cimport PubSocket as cppPubSocket
@@ -161,6 +201,14 @@ cdef class Poller:
 cdef class SubSocket:
   cdef cppSubSocket * socket
   cdef bool is_owner
+  cdef bool receive_timing_enabled
+  cdef unsigned long receive_timing_owner
+  cdef ReceiveTiming receive_timing_ring[64]
+  cdef unsigned int receive_timing_next
+  cdef unsigned int receive_timing_count
+  cdef uint64_t receive_timing_sequence
+  cdef uint64_t receive_timing_omitted
+  cdef uint64_t receive_timing_generation
 
   def __cinit__(self):
     with nogil:
@@ -198,8 +246,87 @@ cdef class SubSocket:
     with nogil:
       self.socket.setTimeout(timeout)
 
+  cdef check_receive_timing_owner(self):
+    if self.receive_timing_owner != 0 and self.receive_timing_owner != PyThread_get_thread_ident():
+      raise RuntimeError("receive timing must be managed by the thread that enabled it")
+
+  def enable_receive_timing(self, bool enabled):
+    """Opt in on the receiving thread; disabling preserves previous records."""
+    self.check_receive_timing_owner()
+    if enabled:
+      self.receive_timing_owner = PyThread_get_thread_ident()
+    self.receive_timing_enabled = enabled
+
+  def reset_receive_timing(self):
+    """Explicitly clear records and counts, without changing enablement."""
+    self.check_receive_timing_owner()
+    self.receive_timing_next = 0
+    self.receive_timing_count = 0
+    self.receive_timing_sequence = 0
+    self.receive_timing_omitted = 0
+    self.receive_timing_generation += 1
+
+  def take_receive_timing(self):
+    """Return (oldest-to-newest records, overwritten count), without clearing.
+
+    Each record is (sequence, non_blocking, returned_message, byte_length,
+    before_mono_ns, before_thread_cpu_ns, backend_return_mono_ns,
+    backend_return_thread_cpu_ns, gil_reacquired_mono_ns,
+    gil_reacquired_thread_cpu_ns, python_return_mono_ns,
+    python_return_thread_cpu_ns). Failed clocks are -1. The final interval
+    includes the bytes copy and message cleanup; gil_handoff includes scheduling.
+    """
+    self.check_receive_timing_owner()
+    cdef unsigned int i
+    cdef unsigned int start = (self.receive_timing_next + 64 - self.receive_timing_count) % 64
+    cdef ReceiveTiming *record
+    records = []
+    for i in range(self.receive_timing_count):
+      record = &self.receive_timing_ring[(start + i) % 64]
+      records.append((record.sequence, record.non_blocking, record.returned_message, record.byte_length,
+                      record.before_wall, record.before_cpu, record.backend_return_wall, record.backend_return_cpu,
+                      record.gil_reacquired_wall, record.gil_reacquired_cpu, record.python_return_wall, record.python_return_cpu))
+    return tuple(records), self.receive_timing_omitted
+
+  cdef receive_timed(self, bool non_blocking):
+    cdef cppMessage *msg
+    cdef ReceiveTiming record
+    cdef uint64_t generation = self.receive_timing_generation
+    record.non_blocking = non_blocking
+    with nogil:
+      msgq_receive_clock_pair(&record.before_wall, &record.before_cpu)
+      msg = self.socket.receive(non_blocking)
+      msgq_receive_clock_pair(&record.backend_return_wall, &record.backend_return_cpu)
+    msgq_receive_clock_pair(&record.gil_reacquired_wall, &record.gil_reacquired_cpu)
+
+    record.returned_message = msg != NULL
+    record.byte_length = 0
+    if msg == NULL:
+      m = None
+    else:
+      record.byte_length = msg.getSize()
+      m = msg.getData()[:record.byte_length]
+      with nogil:
+        del msg
+
+    msgq_receive_clock_pair(&record.python_return_wall, &record.python_return_cpu)
+    # All shared ring access holds the GIL. Only this stack-local record is
+    # written while nogil; no Python object is created for the receipt here.
+    if generation == self.receive_timing_generation:
+      self.receive_timing_sequence += 1
+      record.sequence = self.receive_timing_sequence
+      self.receive_timing_ring[self.receive_timing_next] = record
+      self.receive_timing_next = (self.receive_timing_next + 1) % 64
+      if self.receive_timing_count < 64:
+        self.receive_timing_count += 1
+      else:
+        self.receive_timing_omitted += 1
+    return m
+
   def receive(self, bool non_blocking=False):
     cdef cppMessage *msg
+    if self.receive_timing_enabled and self.receive_timing_owner == PyThread_get_thread_ident():
+      return self.receive_timed(non_blocking)
     with nogil:
       msg = self.socket.receive(non_blocking)
 

msgq_repo/msgq/tests/test_receive_timing.py

@@ -0,0 +1,131 @@
+import ctypes
+import threading
+import time
+import uuid
+
+import msgq
+import pytest
+
+
+def sockets(timeout=10):
+  endpoint = f"receive_timing_{uuid.uuid4().hex}"
+  pub = msgq.pub_sock(endpoint)
+  sub = msgq.sub_sock(endpoint, timeout=timeout)
+  return pub, sub
+
+
+def assert_clocks(record):
+  assert len(record) == 12
+  assert all(t >= 0 for t in record[4:])
+  assert list(record[4::2]) == sorted(record[4::2])
+  assert list(record[5::2]) == sorted(record[5::2])
+
+
+def test_receive_timing_opt_in_and_preserves_messages():
+  pub, sub = sockets()
+  for payload in (b"x", b"unchanged\x00message"):
+    pub.send(payload)
+    assert sub.receive() == payload
+  assert sub.receive(non_blocking=True) is None
+  assert sub.take_receive_timing() == ((), 0)
+
+  sub.enable_receive_timing(True)
+  for payload in (b"x", b"unchanged\x00message"):
+    pub.send(payload)
+    assert sub.receive() == payload
+  assert sub.receive(non_blocking=True) is None
+  assert sub.receive() is None
+  records, omitted = sub.take_receive_timing()
+  assert omitted == 0
+  assert [r[:4] for r in records] == [(1, False, True, 1), (2, False, True, 17),
+                                     (3, True, False, 0), (4, False, False, 0)]
+  for record in records:
+    assert_clocks(record)
+  assert sub.take_receive_timing() == (records, omitted)
+
+  sub.enable_receive_timing(False)
+  pub.send(b"disabled again")
+  assert sub.receive() == b"disabled again"
+  assert sub.receive(non_blocking=True) is None
+  assert sub.take_receive_timing() == (records, omitted)
+  sub.reset_receive_timing()
+  assert sub.take_receive_timing() == ((), 0)
+
+
+def test_receive_timing_bounded_ring_and_explicit_reset():
+  _, sub = sockets()
+  sub.enable_receive_timing(True)
+  for _ in range(150):
+    assert sub.receive(non_blocking=True) is None
+  records, omitted = sub.take_receive_timing()
+  assert len(records) == 64
+  assert omitted == 86
+  assert [r[0] for r in records] == list(range(87, 151))
+  sub.reset_receive_timing()
+  assert sub.take_receive_timing() == ((), 0)
+  assert sub.receive(non_blocking=True) is None
+  assert sub.take_receive_timing()[0][0][0] == 1
+
+
+def test_receive_timing_thread_owner_and_socket_isolation():
+  _, sub = sockets()
+  _, other_sub = sockets()
+  sub.enable_receive_timing(True)
+  failures = []
+
+  def other_thread():
+    try:
+      # The receive itself remains valid and uninstrumented on other threads.
+      assert sub.receive(non_blocking=True) is None
+      for op in (sub.take_receive_timing, sub.reset_receive_timing,
+                 lambda: sub.enable_receive_timing(False), lambda: sub.enable_receive_timing(True)):
+        with pytest.raises(RuntimeError, match="thread that enabled"):
+          op()
+    except BaseException as exc:
+      failures.append(exc)
+
+  worker = threading.Thread(target=other_thread)
+  worker.start()
+  worker.join(timeout=3)
+  assert not worker.is_alive()
+  assert not failures
+  assert sub.take_receive_timing() == ((), 0)
+  assert sub.receive(non_blocking=True) is None
+  assert len(sub.take_receive_timing()[0]) == 1
+  assert other_sub.receive(non_blocking=True) is None
+  assert other_sub.take_receive_timing() == ((), 0)
+
+
+def test_receive_timing_exposes_gil_handoff_delay():
+  # PyDLL deliberately retains the main thread's GIL while libc sleeps.
+  # The receiving thread's native timeout can expire independently of it.
+  libc = ctypes.PyDLL(None)
+  libc.usleep.argtypes = [ctypes.c_uint]
+  libc.usleep.restype = ctypes.c_int
+  ready = threading.Event()
+  results = []
+  failures = []
+
+  def receiver():
+    try:
+      _, sub = sockets(timeout=40)
+      sub.enable_receive_timing(True)
+      ready.set()
+      assert sub.receive() is None
+      results.append(sub.take_receive_timing()[0][0])
+    except BaseException as exc:
+      failures.append(exc)
+
+  worker = threading.Thread(target=receiver)
+  worker.start()
+  assert ready.wait(timeout=3)
+  time.sleep(0.01)
+  assert libc.usleep(150_000) == 0
+  worker.join(timeout=3)
+  assert not worker.is_alive()
+  assert not failures
+  record = results[0]
+  assert_clocks(record)
+  assert record[6] - record[4] >= 30_000_000
+  assert record[8] - record[6] >= 50_000_000
+  assert record[9] - record[7] < 20_000_000

openpilot/selfdrive/car/card.py

@@ -13,6 +13,7 @@ from opendbc.car.structs import car
 
 from openpilot.common.params import Params
 from openpilot.starpilot.card_loop_timing import CardLoopTiming
+from openpilot.starpilot.card_stall_timing import CardStallTiming
 from openpilot.common.realtime import config_realtime_process, drop_realtime, set_core_affinity, Priority, Ratekeeper
 from openpilot.common.swaglog import cloudlog, ForwardingHandler
 
@@ -442,6 +443,12 @@ class Car:
         tracker.claim(gesture, now_ns, key)
 
   def timing_mark(self, name):
+    try:
+      stall = getattr(self, 'stall_timing', None)
+      if stall is not None:
+        stall.mark(name)
+    except Exception:
+      pass  # Observations must never interrupt control or its fail-closed paths.
     timing = getattr(self, 'loop_timing', None)
     if timing is not None and getattr(timing, 'detailed', True):
       timing.mark(name)
@@ -455,6 +462,12 @@ class Car:
     can_received_ns = time.monotonic_ns()
     can_list = can_capnp_to_list(can_strs)
     self.timing_mark('can_decode_end')
+    try:
+      stall = getattr(self, 'stall_timing', None)
+      if stall is not None:
+        stall.observe_can(can_strs, can_list)
+    except Exception:
+      pass
 
     # Update carState from CAN
     CS = self.CI.update(can_list)
@@ -920,6 +933,12 @@ class Car:
     self.timing_mark('car_state_send_start')
     self.pm.send('carState', cs_send)
     self.timing_mark('car_state_send_end')
+    try:
+      stall = getattr(self, 'stall_timing', None)
+      if stall is not None:
+        stall.published(int(cs_send.logMonoTime), self.slc_producer_session)
+    except Exception:
+      pass
     self.timing_source_car_state_ns = int(cs_send.logMonoTime)
     timing = getattr(self, 'loop_timing', None)
     if timing is not None:
@@ -1402,6 +1421,13 @@ class Car:
     set_core_affinity([0, 1, 2, 3])
     next_cruise_read = 0.0
     while not evt.is_set():
+      try:
+        stall = getattr(self, 'stall_timing', None)
+        if stall is not None:
+          stall.worker_mark('settings_start')
+          stall.flush(cloudlog.event)
+      except Exception:
+        pass
       timing = getattr(self, 'loop_timing', None)
       report = timing.take_report() if timing is not None else None
       if report is not None:
@@ -1418,9 +1444,20 @@ class Car:
         self.slc_requested_configuration = self.read_slc_configuration()
         self.v_cruise_helper.intervals = read_cruise_intervals(self.params, pcm_cruise=self.CP.pcmCruise)
         next_cruise_read = now + 1.0
+      try:
+        if stall is not None:
+          stall.worker_mark('settings_end')
+      except Exception:
+        pass
       time.sleep(0.1)
 
   def card_thread(self):
+    self.stall_timing = None
+    try:
+      if self.aol_replay and not REPLAY and ioniq6_long_eligible(self.CP):
+        self.stall_timing = CardStallTiming(self.can_sock)
+    except Exception:
+      pass
     self.loop_timing = (CardLoopTiming(detailed=os.getenv('CARD_PHASE_TIMING') == '1')
                         if self.aol_replay and not REPLAY else None)
     self.timing_thread_id = threading.get_native_id()
@@ -1429,6 +1466,11 @@ class Car:
     try:
       t.start()
       while True:
+        try:
+          if self.stall_timing is not None:
+            self.stall_timing.begin()
+        except Exception:
+          pass
         timing = self.loop_timing
         if timing is not None and timing.detailed:
           timing.begin()
@@ -1436,6 +1478,11 @@ class Car:
         self.timing_mark('startup_check_end')
         self.step()
         self.rk.monitor_time()
+        try:
+          if self.stall_timing is not None:
+            self.stall_timing.finish()
+        except Exception:
+          pass
         if timing is not None and timing.detailed:
           timing.finish({'source_car_state_ns': getattr(self, 'timing_source_car_state_ns', 0),
                          'source_control_ns': int(self.sm.logMonoTime['carControl']),

openpilot/starpilot/aol/tests/test_card_stall_timing.py

@@ -0,0 +1,446 @@
+"""Bounded, exception-isolated Card observations never grant authority."""
+from contextlib import ExitStack
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from openpilot.starpilot.card_stall_timing import CardStallTiming
+
+BASE = 10_000_000_000
+MS = 1_000_000
+
+
+class Clock:
+  def __init__(self):
+    self.wall = BASE
+    self.cpu = 0
+    self.calls = 0
+
+  def __call__(self):
+    self.calls += 1
+    return self.wall, self.cpu
+
+  def advance(self, delta, cpu=0):
+    self.wall += delta
+    self.cpu += cpu
+
+
+class Socket:
+  def __init__(self):
+    self.enabled = False
+    self.takes = 0
+
+  def enable_receive_timing(self, enabled):
+    self.enabled = enabled
+
+  def take_receive_timing(self):
+    self.takes += 1
+    return (), 0
+
+
+def recorder():
+  clock = Clock()
+  return CardStallTiming(Socket(), clock=clock), clock
+
+
+def frame(timing, clock, delta, source=None, session='producer'):
+  timing.begin()
+  clock.advance(delta)
+  timing.mark('car_state_send_end')
+  timing.published(clock.wall if source is None else source, session)
+  timing.finish()
+
+
+def report(timing):
+  result = []
+  timing.last_attempt = None
+  timing.flush(lambda *args, **kwargs: result.append(kwargs))
+  assert len(result) == 1
+  return result[0]
+
+
+def test_cross_loop_tail_and_head_gap_is_captured_before_loop_end():
+  timing, clock = recorder()
+  timing.begin()
+  clock.advance(MS)
+  timing.published(clock.wall, 'producer')
+  clock.advance(19 * MS)
+  timing.mark('controller_apply_end')
+  timing.finish()
+  timing.begin()
+  clock.advance(18 * MS, cpu=2 * MS)
+  timing.mark('submaster_end')
+  timing.published(clock.wall, 'producer')
+  result = report(timing)
+  assert result['source_gap_ns'] == result['send_gap_ns'] == 37 * MS
+  assert result['previous_publication_frame_retained'] and not result['current_frame_complete']
+  assert result['previous_frame'][-1][0] == 'loop_end'
+  assert result['current_frame'][-1][0] == 'submaster_end'
+  assert result['current_frame'][-1][2] - result['current_frame'][0][2] == 2 * MS
+  saved = result['current_frame']
+  for _ in range(4):
+    frame(timing, clock, MS)
+  assert result['current_frame'] == saved and timing.socket.takes == 1
+
+
+@pytest.mark.parametrize('extra, emitted', [(0, False), (1, True)])
+def test_strict_threshold(extra, emitted):
+  timing, clock = recorder()
+  frame(timing, clock, MS)
+  frame(timing, clock, 30 * MS + extra)
+  assert bool(timing.pending) == emitted
+
+
+@pytest.mark.parametrize('source_gap, send_gap', [(37 * MS, 21 * MS), (14 * MS, 33 * MS)])
+def test_source_and_send_clocks_remain_separate(source_gap, send_gap):
+  timing, clock = recorder()
+  frame(timing, clock, MS, source=BASE)
+  frame(timing, clock, send_gap, source=BASE + source_gap)
+  result = report(timing)
+  assert result['source_gap_ns'] == source_gap and result['send_gap_ns'] == send_gap
+  assert result['source_clock'] == 'CLOCK_MONOTONIC' and result['clock'] == 'CLOCK_MONOTONIC'
+  assert len(result['clock_mapping_mono_boot_mono']) == 3
+
+
+def test_sessions_missing_sources_and_clock_reversal_reset_baseline():
+  timing, clock = recorder()
+  frame(timing, clock, MS)
+  frame(timing, clock, 40 * MS, session='replacement')
+  frame(timing, clock, 40 * MS, source=0, session='replacement')
+  frame(timing, clock, 40 * MS, session='replacement')
+  clock.wall -= 100 * MS
+  frame(timing, clock, MS, session='replacement')
+  assert not timing.pending
+  frame(timing, clock, 31 * MS, session='replacement')
+  assert report(timing)['source_gap_ns'] == 31 * MS
+
+
+def test_bounds_throttle_queue_full_and_quiet_loss_summary():
+  timing, clock = recorder()
+  frame(timing, clock, MS)
+  frame(timing, clock, 31 * MS)
+  frame(timing, clock, 31 * MS)
+  frame(timing, clock, timing.INTERVAL_NS)
+  assert len(timing.pending) == 1
+  assert timing.late_count == 3 and timing.throttled_count == 1 and timing.queue_full_count == 1
+  for _ in range(100):
+    timing.mark('can_receive_end')
+  assert max(timing.lengths) == timing.MAX_MARKS and timing.omitted_marks > 0
+  result = report(timing)
+  assert result['counters']['queue_full_count'] == 1
+  timing.errors += 1
+  result = report(timing)
+  assert 'source_gap_ns' not in result and result['counters']['capture_errors'] == 1
+  assert len(timing.frames) == 2 and all(len(f) == timing.MAX_MARKS for f in timing.frames)
+
+
+def test_nonpublishing_frames_and_worker_snapshot_are_explicit():
+  timing, clock = recorder()
+  frame(timing, clock, MS)
+  timing.worker_mark('settings_start')
+  for _ in range(3):
+    timing.begin()
+    clock.advance(10 * MS)
+    timing.finish()
+  frame(timing, clock, MS)
+  old = timing.worker
+  timing.worker_mark('settings_end')
+  result = report(timing)
+  assert result['nonpublishing_frames'] == 3 and not result['previous_publication_frame_retained']
+  assert result['worker_progress'] == old
+
+
+def test_missing_native_api_is_reported_and_python_capture_remains_honest():
+  clock = Clock()
+  timing = CardStallTiming(object(), clock=clock)
+  assert not timing.native_available
+  result = report(timing)
+  assert not result['native_available'] and result['error']
+  frame(timing, clock, MS)
+  frame(timing, clock, 31 * MS)
+  result = report(timing)
+  assert result['native'] is None and not result['native_available']
+
+
+def test_clock_snapshot_and_logging_failures_do_not_escape(monkeypatch):
+  timing, clock = recorder()
+  def fail(*args, **kwargs):
+    raise RuntimeError('observer failure')
+  timing.clock = fail
+  timing.begin()
+  timing.mark('can_receive_end')
+  timing.finish()
+  timing.published(BASE, 'producer')
+  timing.worker_mark('settings_start')
+  assert timing.errors == 5
+  timing.clock = clock
+  frame(timing, clock, MS)
+  timing.socket.take_receive_timing = fail
+  frame(timing, clock, 31 * MS)
+  assert timing.errors == 6
+  monkeypatch.setattr('openpilot.starpilot.card_stall_timing.time.monotonic_ns', lambda: clock.wall)
+  for _ in range(100):
+    timing.flush(fail)
+  assert timing.log_errors == 1
+  clock.advance(timing.INTERVAL_NS)
+  timing.flush(fail)
+  assert timing.log_errors == 2
+
+
+def test_unselected_marks_do_not_read_clocks():
+  timing, clock = recorder()
+  before = clock.calls
+  for _ in range(100):
+    timing.mark('unselected')
+  assert clock.calls == before
+
+
+@pytest.mark.parametrize('full_op', [False, True])
+def test_actual_card_publication_bytes_and_order_are_unchanged(full_op):
+  from openpilot.common.prefix import OpenpilotPrefix
+  from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup, OpenpilotPrefix():
+    cleanup.callback(case.doCleanups)
+    sd, _, step, daemon, _ = case._ioniq_availability_callers(full_op=full_op)
+    axis, _, command = step(2)
+    assert axis.lateralActive and command.latActive and axis.longitudinalActive == full_op
+    state = fixture.car_state()
+    daemon.car_params_published = True
+    daemon.wheel_commands = ()
+    daemon.initialized_prev = True
+    daemon.timing_thread_id = 7
+    initial_sequence = daemon.aol_sequence
+    results = []
+    for active in (False, True):
+      timing, clock = recorder()
+      clock.wall = BASE + 20 * MS
+      daemon.stall_timing = timing if active else None
+      daemon.aol_sequence = initial_sequence
+      output = []
+      daemon.pm = SimpleNamespace(send=lambda name, msg, output=output: output.append((name, msg.to_bytes())))
+      timing.begin()
+      daemon.state_publish(state, None)
+      timing.finish()
+      results.append(output)
+      if active:
+        assert timing.last_publication[1] == BASE + 20 * MS
+    assert results[0] == results[1]
+    services = [name for name, _ in results[1]]
+    assert services.index('carOutput') < services.index('aolIntentWire') < services.index('carState')
+    assert sd.aol_axis_decision.lateral_active and sd.aol_axis_decision.longitudinal_active == full_op
+
+
+def test_failed_send_does_not_advance_successful_publication_baseline():
+  from openpilot.common.prefix import OpenpilotPrefix
+  from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup, OpenpilotPrefix():
+    cleanup.callback(case.doCleanups)
+    _, _, _, daemon, _ = case._ioniq_availability_callers(full_op=False)
+    timing, clock = recorder()
+    daemon.stall_timing = timing
+    daemon.car_params_published = True
+    daemon.timing_thread_id = 7
+    timing.begin()
+    def send(name, unused):
+      if name == 'carState':
+        raise RuntimeError('send failed')
+    daemon.pm = SimpleNamespace(send=send)
+    with pytest.raises(RuntimeError, match='send failed'):
+      daemon.state_publish(fixture.car_state(), None)
+    assert timing.last_publication is None and not timing.pending
+
+
+
+
+@pytest.mark.parametrize('replay, aol, long_profile, expected', [
+  (False, True, True, True), (True, True, True, False),
+  (False, False, True, False), (False, True, False, False),
+])
+def test_actual_card_thread_gate_uses_configured_profile_not_active_long(replay, aol, long_profile, expected, monkeypatch):
+  import threading
+  from openpilot.selfdrive.car import card as module
+  from openpilot.selfdrive.car.tests.test_hyundai_aol import candidate
+  class StopLoop(Exception):
+    pass
+  class Worker:
+    def __init__(self, **unused):
+      pass
+    def start(self):
+      pass
+    def join(self):
+      pass
+  _, cp = candidate(True)
+  cp.safetyConfigs[-1].safetyParam |= 0x0800
+  if not long_profile:
+    cp.openpilotLongitudinalControl = False
+  daemon: Any = module.Car.__new__(module.Car)
+  daemon.CP, daemon.aol_replay = cp, aol
+  daemon.can_sock = Socket()
+  daemon.ioniq6_keepalive = None
+  daemon.vehicle_startup = SimpleNamespace(check=lambda: None)
+  def stop():
+    raise StopLoop
+  daemon.step = stop
+  monkeypatch.setattr(module, 'REPLAY', replay)
+  monkeypatch.setattr(module, 'threading', SimpleNamespace(Thread=Worker, Event=threading.Event, get_native_id=lambda: 7))
+  with pytest.raises(StopLoop):
+    daemon.card_thread()
+  assert (daemon.stall_timing is not None) == expected
+
+
+def test_existing_card_worker_continues_settings_after_diagnostic_logger_error(monkeypatch):
+  import threading
+  from openpilot.selfdrive.car import card as module
+  timing, clock = recorder()
+  frame(timing, clock, MS)
+  frame(timing, clock, 31 * MS)
+  daemon: Any = module.Car.__new__(module.Car)
+  daemon.stall_timing, daemon.loop_timing = timing, None
+  daemon.CP = SimpleNamespace(openpilotLongitudinalControl=True, pcmCruise=False)
+  daemon.aol_card_intent = SimpleNamespace(settings=None)
+  daemon.v_cruise_helper = SimpleNamespace(intervals=None)
+  reads = []
+  daemon.params = SimpleNamespace(get_bool=lambda key: reads.append(key) or False)
+  daemon.read_slc_configuration = lambda: 'configuration'
+  stopped = threading.Event()
+  def fail(*unused, **kwargs):
+    raise RuntimeError('diagnostic logger unavailable')
+  monkeypatch.setattr(module, 'cloudlog', SimpleNamespace(event=fail))
+  monkeypatch.setattr(module, 'drop_realtime', lambda: None)
+  monkeypatch.setattr(module, 'set_core_affinity', lambda cores: None)
+  monkeypatch.setattr(module, 'read_settings', lambda params: 'aol')
+  monkeypatch.setattr(module, 'read_cruise_intervals', lambda *args, **kwargs: 'intervals')
+  monkeypatch.setattr(module, 'time', SimpleNamespace(monotonic=lambda: 10., sleep=lambda duration: stopped.set()))
+  daemon.params_thread(stopped)
+  assert reads == ['IsMetric', 'ExperimentalMode']
+  assert daemon.aol_card_intent.settings == 'aol' and daemon.v_cruise_helper.intervals == 'intervals'
+  assert timing.log_errors == 1 and timing.worker[0] == 'settings_end'
+
+
+
+
+@pytest.mark.parametrize('hook', ['mark', 'published'])
+def test_added_publication_hooks_cannot_stop_original_outputs(hook):
+  from openpilot.common.prefix import OpenpilotPrefix
+  from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  def fail(*args, **kwargs):
+    raise RuntimeError('injected observer failure')
+  with ExitStack() as cleanup, OpenpilotPrefix():
+    cleanup.callback(case.doCleanups)
+    _, _, _, daemon, _ = case._ioniq_availability_callers(full_op=False)
+    daemon.car_params_published = True
+    daemon.wheel_commands = ()
+    daemon.stall_timing = SimpleNamespace(mark=lambda name: None, published=lambda *args: None)
+    setattr(daemon.stall_timing, hook, fail)
+    output = []
+    daemon.pm = SimpleNamespace(send=lambda name, msg: output.append(name))
+    daemon.state_publish(fixture.car_state(), None)
+    assert 'carState' in output and 'carOutput' in output
+
+
+@pytest.mark.parametrize('hook', ['begin', 'finish', 'constructor'])
+def test_added_loop_hooks_cannot_interrupt_control_order(hook, monkeypatch):
+  import threading
+  from openpilot.selfdrive.car import card as module
+  class StopLoop(Exception):
+    pass
+  class Worker:
+    def __init__(self, **kwargs):
+      pass
+    def start(self):
+      pass
+    def join(self):
+      pass
+  trace = []
+  daemon: Any = module.Car.__new__(module.Car)
+  daemon.aol_replay = True
+  daemon.ioniq6_keepalive = None
+  daemon.CP = object()
+  daemon.can_sock = Socket()
+  daemon.vehicle_startup = SimpleNamespace(check=lambda: trace.append('check'))
+  daemon.step = lambda: trace.append('step')
+  def monitor():
+    trace.append('monitor')
+    if trace.count('monitor') == 2:
+      raise StopLoop
+  daemon.rk = SimpleNamespace(monitor_time=monitor)
+  def fail(*args):
+    raise RuntimeError('observer failure')
+  timing = SimpleNamespace(begin=lambda: None, finish=lambda: None, mark=lambda name: None)
+  if hook != 'constructor':
+    setattr(timing, hook, fail)
+  monkeypatch.setattr(module, 'REPLAY', False)
+  monkeypatch.setattr(module, 'ioniq6_long_eligible', lambda cp: True)
+  monkeypatch.setattr(module, 'CardStallTiming', fail if hook == 'constructor' else lambda socket: timing)
+  monkeypatch.setattr(module, 'CardLoopTiming', lambda **kwargs: SimpleNamespace(detailed=False))
+  monkeypatch.setattr(module, 'threading', SimpleNamespace(Thread=Worker, Event=threading.Event, get_native_id=lambda: 7))
+  with pytest.raises(StopLoop):
+    daemon.card_thread()
+  assert trace == ['check', 'step', 'monitor'] * 2
+
+
+@pytest.mark.parametrize('hook', ['worker_mark', 'flush'])
+def test_worker_hooks_cannot_stop_settings(hook, monkeypatch):
+  import threading
+  from openpilot.selfdrive.car import card as module
+  daemon: Any = module.Car.__new__(module.Car)
+  def fail(*args):
+    raise RuntimeError('observer failure')
+  daemon.stall_timing = SimpleNamespace(worker_mark=lambda *args: None, flush=lambda *args: None)
+  setattr(daemon.stall_timing, hook, fail)
+  daemon.loop_timing = None
+  daemon.CP = SimpleNamespace(openpilotLongitudinalControl=False, pcmCruise=False)
+  daemon.aol_card_intent = None
+  daemon.v_cruise_helper = SimpleNamespace(intervals=None)
+  reads = []
+  daemon.params = SimpleNamespace(get_bool=lambda key: reads.append(key) or False)
+  daemon.read_slc_configuration = lambda: 'configuration'
+  stopped = threading.Event()
+  monkeypatch.setattr(module, 'drop_realtime', lambda: None)
+  monkeypatch.setattr(module, 'set_core_affinity', lambda cores: None)
+  monkeypatch.setattr(module, 'read_cruise_intervals', lambda *args, **kwargs: 'intervals')
+  monkeypatch.setattr(module, 'time', SimpleNamespace(monotonic=lambda: 10., sleep=lambda duration: stopped.set()))
+  daemon.params_thread(stopped)
+  assert reads == ['IsMetric', 'ExperimentalMode'] and daemon.v_cruise_helper.intervals == 'intervals'
+
+
+def test_can_summary_is_bounded_and_detached_without_payload_copy():
+  timing, clock = recorder()
+  timing.begin()
+  raw = [b'can'] * 100
+  decoded = [(i + BASE, [(1, b'abc', 0)] * 3) for i in range(100)]
+  timing.observe_can(raw, decoded)
+  assert timing.can_summaries[timing.index] == (100, 64, 192, 192, BASE, BASE + 63)
+  timing.published(BASE, 'producer')
+  timing.finish()
+  frame(timing, clock, 31 * MS)
+  result = report(timing)
+  raw.clear()
+  decoded.clear()
+  assert result['previous_can'] == (100, 64, 192, 192, BASE, BASE + 63)
+  assert result['can_source_clock'] == 'CLOCK_BOOTTIME'
+
+
+def test_actual_receive_observer_failure_preserves_can_and_original_interface_error(monkeypatch):
+  from openpilot.selfdrive.car import card as module
+  daemon: Any = module.Car.__new__(module.Car)
+  daemon.can_sock = object()
+  raw = [b'raw']
+  decoded = [(BASE, [(1, b'data', 0)])]
+  seen = []
+  def observer_fail(*args):
+    raise RuntimeError('observer failure')
+  def original_update(batch):
+    seen.append(batch)
+    raise ValueError('original interface exception')
+  daemon.stall_timing = SimpleNamespace(mark=lambda name: None, observe_can=observer_fail)
+  daemon.CI = SimpleNamespace(update=original_update)
+  monkeypatch.setattr(module.messaging, 'drain_sock_raw', lambda socket, **kwargs: raw)
+  monkeypatch.setattr(module, 'can_capnp_to_list', lambda batch: decoded if batch is raw else None)
+  with pytest.raises(ValueError, match='original interface exception'):
+    daemon.state_update()
+  assert seen == [decoded] and seen[0] is decoded

openpilot/starpilot/card_stall_timing.py

@@ -0,0 +1,185 @@
+"""Temporary, bounded observations of Card publication gaps.
+
+This observer never supplies control data. MONOTONIC and thread CPU are paired
+sequential samples, not atomic observations. Backend-return to GIL-reacquired
+also includes scheduler delay; it does not prove contention on Python's lock.
+"""
+from collections import deque
+import threading
+import time
+
+
+class CardStallTiming:
+  LIMIT_NS = 30_000_000
+  INTERVAL_NS = 5_000_000_000
+  MAX_MARKS = 32
+  MARKS = frozenset(('can_receive_start', 'can_receive_end', 'can_decode_end', 'interface_update_end',
+                    'radar_update_start', 'radar_update_end', 'submaster_end', 'permission_and_cruise_end',
+                    'conditional_params_end', 'wheel_params_start', 'wheel_params_end', 'state_update_end',
+                    'startup_maintain_end', 'car_state_send_start', 'car_state_send_end', 'state_publish_end',
+                    'controller_apply_start', 'controller_apply_end', 'sendcan_start', 'sendcan_end',
+                    'controls_update_end'))
+
+  def __init__(self, socket, *, clock=None):
+    self.clock = clock or self._clock
+    self.socket = socket
+    self.frames = [[None] * self.MAX_MARKS for _ in range(2)]
+    self.lengths = [0, 0]
+    self.can_summaries = [None, None]
+    self.index = 0
+    self.generation = 0
+    self.previous = False
+    self.last_publication = None
+    self.last_capture = None
+    self.last_attempt = None
+    self.pending = deque(maxlen=1)
+    self.late_count = 0
+    self.throttled_count = 0
+    self.queue_full_count = 0
+    self.omitted_marks = 0
+    self.errors = 0
+    self.log_errors = 0
+    self.worker = None
+    self.thread_id = threading.get_native_id()
+    self.native_available = False
+    try:
+      socket.enable_receive_timing(True)
+      self.native_available = True
+    except Exception:
+      self.errors += 1
+    self.status_pending = True
+    self.last_counters = None
+
+  @staticmethod
+  def _clock():
+    return time.monotonic_ns(), time.thread_time_ns()
+
+  def begin(self):
+    try:
+      self.generation += 1
+      self.index = 1 - self.index
+      self.previous = self.lengths[1 - self.index] > 0
+      self.lengths[self.index] = 0
+      self.can_summaries[self.index] = None
+      self._mark('loop_start')
+    except Exception:
+      self.errors += 1
+
+  def _mark(self, name):
+    length = self.lengths[self.index]
+    if length == self.MAX_MARKS:
+      self.omitted_marks += 1
+      return
+    self.frames[self.index][length] = (name, *self.clock())
+    self.lengths[self.index] = length + 1
+
+  def mark(self, name):
+    try:
+      if name in self.MARKS:
+        self._mark(name)
+    except Exception:
+      self.errors += 1
+
+  def observe_can(self, raw, decoded):
+    try:
+      # Bound observer work even if the ordinary drain returned an unusually
+      # large batch. No additional envelope decode or CAN payload copying.
+      count = min(len(raw), len(decoded), 64)
+      frames = byte_count = 0
+      first = last = None
+      for i in range(count):
+        stamp, messages = decoded[i]
+        frames += len(messages)
+        byte_count += len(raw[i])
+        first = stamp if first is None else min(first, stamp)
+        last = stamp if last is None else max(last, stamp)
+      self.can_summaries[self.index] = (len(raw), count, frames, byte_count, first, last)
+    except Exception:
+      self.errors += 1
+
+  def finish(self):
+    try:
+      self._mark('loop_end')
+    except Exception:
+      self.errors += 1
+
+  def published(self, source_ns, session):
+    try:
+      now, cpu = self.clock()
+      last = self.last_publication
+      self.last_publication = (source_ns, now, session, self.generation)
+      if source_ns <= 0 or last is None or last[0] <= 0 or last[2] != session or source_ns <= last[0] or now <= last[1]:
+        return
+      source_gap, send_gap = source_ns - last[0], now - last[1]
+      if max(source_gap, send_gap) <= self.LIMIT_NS:
+        return
+      self.late_count += 1
+      if self.last_capture is not None and now - self.last_capture < self.INTERVAL_NS:
+        self.throttled_count += 1
+        return
+      if self.pending:
+        self.queue_full_count += 1
+        return
+      self.last_capture = now
+      # Snapshot only on a bounded late event. No formatting, logging, Params,
+      # file I/O or locks on Card; the sole consumer runs in params_thread.
+      previous = tuple(self.frames[1 - self.index][:self.lengths[1 - self.index]]) if self.previous else ()
+      current = tuple(self.frames[self.index][:self.lengths[self.index]])
+      native = self.socket.take_receive_timing() if self.native_available else None
+      clock_mapping = (time.monotonic_ns(), time.clock_gettime_ns(time.CLOCK_BOOTTIME), time.monotonic_ns())
+      self.pending.append((source_ns, now, cpu, session, last, source_gap, send_gap,
+                           previous, current, native, self.worker, self.generation, clock_mapping,
+                           self.can_summaries[1 - self.index], self.can_summaries[self.index]))
+    except Exception:
+      self.errors += 1
+
+  def worker_mark(self, phase):
+    try:
+      self.worker = (phase, threading.get_native_id(), *self.clock())
+    except Exception:
+      self.errors += 1
+
+  def flush(self, emit):
+    """Called only by the existing non-realtime worker, at most once per 5s.
+
+    This Python work still uses the GIL. Native ring/phase sizes and report
+    frequency bound it; formatting is not evidence of zero observer effect.
+    """
+    try:
+      now = time.monotonic_ns()
+      if self.last_attempt is not None and now - self.last_attempt < self.INTERVAL_NS:
+        return
+      counter_values = (self.late_count, self.throttled_count, self.queue_full_count, self.omitted_marks, self.errors, self.log_errors)
+      if not self.pending and not self.status_pending and counter_values == self.last_counters:
+        return
+      self.last_attempt = now
+      self.last_counters = counter_values
+      status = self.status_pending
+      self.status_pending = False
+      sample = self.pending.popleft() if self.pending else None
+      counters = {'late_count': self.late_count, 'throttled_count': self.throttled_count,
+                  'queue_full_count': self.queue_full_count, 'omitted_marks': self.omitted_marks,
+                  'capture_errors': self.errors, 'log_errors': self.log_errors}
+      report = {'version': 1, 'native_available': self.native_available, 'thread_id': self.thread_id,
+                'clock': 'CLOCK_MONOTONIC', 'source_clock': 'CLOCK_MONOTONIC', 'cpu_clock': 'CLOCK_THREAD_CPUTIME_ID',
+                'native_fields': ('sequence', 'non_blocking', 'returned_message', 'byte_length', 'before_wall', 'before_cpu',
+                                  'backend_return_wall', 'backend_return_cpu', 'gil_reacquired_wall',
+                                  'gil_reacquired_cpu', 'python_return_wall', 'python_return_cpu'),
+                'counters': counters, 'status': status}
+      if sample is not None:
+        source, send, cpu, session, last, source_gap, send_gap, previous, current, native, worker, generation, mapping, prev_can, current_can = sample
+        report.update(source_ns=source, send_return_observed_ns=send, send_thread_cpu_ns=cpu,
+                      producer_session=session, previous_publication=last,
+                      source_gap_ns=source_gap, send_gap_ns=send_gap, previous_frame=previous,
+                      current_frame=current, current_frame_complete=False, native=native,
+                      native_capacity=64, native_omitted_is_cumulative=True, generation=generation,
+                      previous_publication_frame_retained=last[3] == generation - 1,
+                      nonpublishing_frames=max(0, generation - last[3] - 1),
+                      clock_mapping_mono_boot_mono=mapping,
+                      can_summary_fields=('total_envelopes', 'summarized_envelopes', 'summarized_frames',
+                                          'summarized_bytes', 'min_source_ns', 'max_source_ns'),
+                      can_source_clock='CLOCK_BOOTTIME', previous_can=prev_can, current_can=current_can,
+                      worker_progress=worker)
+      emit('card.stall_timing', error=sample is not None or not self.native_available, **report)
+    except Exception:
+      self.log_errors += 1

tools/ci/card_stall_native_artifact/build_native.py

@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+"""Isolated native msgq build. Never modifies source checkout or build receipts."""
+import argparse
+import hashlib
+import importlib.metadata
+import json
+import os
+from pathlib import Path
+import platform
+import shutil
+import subprocess
+import sys
+import sysconfig
+import tempfile
+
+sys.dont_write_bytecode = True
+
+from validate_elf import BASELINE_SHA256
+
+HERE = Path(__file__).resolve().parent
+PINNED = {'Cython': '3.3.0', 'SCons': '4.11.1', 'pytest': '8.4.2', 'iniconfig': '2.3.0',
+          'packaging': '26.3', 'pluggy': '1.6.0', 'Pygments': '2.21.0'}
+
+
+def check(condition, message):
+  if not condition:
+    raise RuntimeError(message)
+
+
+def capture(*args, cwd=None):
+  return subprocess.check_output(args, cwd=cwd, text=True).strip()
+
+
+def sha(path):
+  return hashlib.sha256(Path(path).read_bytes()).hexdigest()
+
+
+def manifest(root):
+  return {str(p.relative_to(root)): {'size': p.stat().st_size, 'sha256': sha(p)}
+          for p in sorted(root.rglob('*')) if p.is_file() and '__pycache__' not in p.parts}
+
+
+def bind_committed_sources(source, revision, build, source_manifest):
+  tree = subprocess.check_output(['git', 'ls-tree', '-rz', revision, '--', 'msgq_repo'], cwd=source)
+  committed = {}
+  for row in tree.split(b'\0'):
+    if row:
+      metadata, path = row.split(b'\t', 1)
+      mode, kind, blob = metadata.decode().split()
+      committed[path.decode()] = (mode, kind, blob)
+  for relative, info in source_manifest.items():
+    original = 'msgq_repo/'+relative
+    check(original in committed, f'Uncommitted build input: {original}')
+    mode, kind, blob = committed[original]
+    check(kind == 'blob' and mode in ('100644', '100755'), f'Unsupported build input type: {original}')
+    copied = build/relative
+    check(not copied.is_symlink(), f'Symlink build input is not permitted: {original}')
+    check(('100755' if copied.stat().st_mode & 0o111 else '100644') == mode, f'Build input mode differs from commit: {original}')
+    data = subprocess.check_output(['git', 'cat-file', 'blob', blob], cwd=source)
+    check(copied.read_bytes() == data, f'Build input bytes differ from committed blob: {original}')
+    info.update(git_path=original, git_mode=mode, git_blob=blob)
+
+
+def main():
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('source_root', type=Path)
+  parser.add_argument('baseline_so', type=Path)
+  parser.add_argument('output_dir', type=Path)
+  args = parser.parse_args()
+  source, baseline, output = (p.resolve() for p in (args.source_root, args.baseline_so, args.output_dir))
+  check(platform.machine() == 'aarch64' and platform.system() == 'Linux', 'Actual native Linux AArch64 required')
+  check(sys.implementation.name == 'cpython' and sys.version_info[:3] == (3, 12, 13), 'Actual CPython 3.12.13 required')
+  check(os.environ.get('GITHUB_ACTIONS') == 'true' and os.environ.get('CI') == 'true', 'This helper is scoped to the explicitly approved CI job')
+  check(os.environ.get('RUNNER_ARCH') == 'ARM64' and sys.flags.optimize == 0, 'Native ARM64 runner and non-optimized Python required')
+  check(not Path('/AGNOS').exists(), 'Do not run this helper on a device')
+  check(not output.is_relative_to(source) and not source.is_relative_to(output), 'Output and source must be separate trees')
+  check(not output.exists() or not any(output.iterdir()), 'Output directory must be new or empty')
+  check(sha(baseline) == BASELINE_SHA256, 'Wrong reviewed baseline library')
+  source_commit = capture('git', 'rev-parse', 'HEAD', cwd=source)
+  check(source_commit == os.environ['EXPECTED_SOURCE_COMMIT'], 'Source commit differs from approved SHA')
+  base_commit = os.environ['EXPECTED_BASE_COMMIT']
+  check(base_commit == 'e8389b0b7529bad402f1f4e5fbd672ff13d9d382', 'Unreviewed baseline commit')
+  check(len(base_commit) == 40 and capture('git', 'rev-parse', base_commit+'^{commit}', cwd=source) == base_commit, 'Full baseline commit must be available locally')
+  subprocess.run(['git', 'merge-base', '--is-ancestor', base_commit, source_commit], cwd=source, check=True)
+  check(not capture('git', 'status', '--porcelain', '--untracked-files=all', cwd=source), 'Source checkout must be clean, including untracked paths')
+  versions = {name: importlib.metadata.version(name) for name in PINNED}
+  check(versions == PINNED, f'Build/test packages differ from reviewed uv.lock pins: {versions}')
+  compiler = {key: Path(os.environ.get(key, '/usr/bin/'+name+'-13')).resolve() for key, name in [('GCC', 'gcc'), ('GXX', 'g++')]}
+  expected_gcc = os.environ['EXPECTED_GCC_VERSION']
+  check(expected_gcc.split('.')[0] == '13', 'Only explicitly reviewed GCC 13 is allowed')
+  for path in compiler.values():
+    check(capture(str(path), '-dumpfullversion') == expected_gcc, f'Unapproved compiler version: {path}')
+    check(capture(str(path), '-dumpmachine') == 'aarch64-linux-gnu', f'Compiler is not native generic AArch64: {path}')
+  output.mkdir(parents=True, exist_ok=True)
+  artifacts = output/'artifacts'
+  artifacts.mkdir()
+  toolchain = {
+    'qualification': 'CI ARM host artifact; not deployed or device-qualified',
+    'source_commit': source_commit, 'base_commit': base_commit, 'baseline_sha256': sha(baseline),
+    'python': sys.version, 'python_executable': str(Path(sys.executable).resolve()), 'python_sha256': sha(Path(sys.executable).resolve()),
+    'python_sysconfig': {key: sysconfig.get_config_var(key) for key in ('SOABI', 'MULTIARCH', 'CONFIG_ARGS', 'CC', 'CFLAGS', 'LDFLAGS')},
+    'packages': versions, 'platform': platform.platform(), 'os_release': Path('/etc/os-release').read_text(),
+    'compilers': {key: {'path': str(path), 'sha256': sha(path), 'version': capture(str(path), '--version')} for key, path in compiler.items()},
+    'compiler_packages': capture('dpkg-query', '-W', '-f=${Package}\t${Version}\t${Architecture}\n', 'gcc-13', 'g++-13', 'libstdc++6', 'libc6', 'binutils'),
+    'binutils': capture('readelf', '--version'), 'isa_flags': ['-march=armv8-a', '-mtune=generic'],
+    'cxx_compatibility_flags': ['-fno-sized-deallocation'],
+    'requirements_sha256': sha(HERE/'requirements.lock'), 'uv_lock_sha256': sha(source/'uv.lock'),
+  }
+  libstdcxx = Path(capture(str(compiler['GXX']), '-print-file-name=libstdc++.so')).resolve()
+  toolchain['compiler_libstdcxx'] = {'path': str(libstdcxx), 'sha256': sha(libstdcxx)}
+  (artifacts/'toolchain.json').write_text(json.dumps(toolchain, indent=2)+'\n')
+  build = output/'build'
+  shutil.copytree(source/'msgq_repo', build, ignore=shutil.ignore_patterns('.git', '*.so', '*.o', '*.a', '*.pyc', '__pycache__', '.sconsign*', '*_pyx.cpp'))
+  source_manifest = manifest(build)
+  bind_committed_sources(source, source_commit, build, source_manifest)
+  (artifacts/'source_files.json').write_text(json.dumps(source_manifest, indent=2)+'\n')
+  baseline_root = output/'baseline'
+  (baseline_root/'msgq/visionipc').mkdir(parents=True)
+  (baseline_root/'msgq/tests').mkdir()
+  baseline_files = ['msgq/__init__.py', 'msgq/visionipc/__init__.py', 'msgq/visionipc/visionipc_pyx.so',
+                    'msgq/tests/__init__.py', 'msgq/tests/test_fake.py', 'msgq/tests/test_messaging.py', 'msgq/tests/test_poller.py']
+  for relative in baseline_files:
+    data = subprocess.check_output(['git', 'show', base_commit+':msgq_repo/'+relative], cwd=source)
+    (baseline_root/relative).write_bytes(data)
+  tracked_baseline = subprocess.check_output(['git', 'show', base_commit+':msgq_repo/msgq/ipc_pyx.so'], cwd=source)
+  check(hashlib.sha256(tracked_baseline).hexdigest() == BASELINE_SHA256, 'Baseline commit does not contain reviewed ipc_pyx bytes')
+  shutil.copy2(baseline, baseline_root/'msgq/ipc_pyx.so')
+  companion_relative = Path('msgq/visionipc/visionipc_pyx.so')
+  companion_sha = sha(baseline_root/companion_relative)
+  check(companion_sha == 'd9aee015a9ca9d6d8f4dc54c5bca6a8f3ab24ad9a42477561ea7cc32eebff5a4', 'Unreviewed companion baseline')
+  check(sha(source/'msgq_repo'/companion_relative) == companion_sha, 'Source checkout changed shipped visionipc companion')
+  shutil.copy2(baseline_root/companion_relative, build/companion_relative)
+  (artifacts/'unchanged_companion.json').write_text(json.dumps({'git_blob': capture('git', 'rev-parse', base_commit+':msgq_repo/'+str(companion_relative), cwd=source), 'sha256': companion_sha}, indent=2)+'\n')
+  wrappers = output/'compiler-bin'
+  wrappers.mkdir()
+  for name in ('cc', 'gcc', 'c++', 'g++'):
+    (wrappers/name).symlink_to(HERE/'compiler_wrapper.py')
+  env = dict(os.environ, PATH=str(wrappers)+os.pathsep+str(Path(sys.executable).parent)+os.pathsep+os.environ['PATH'],
+             PYTHONNOUSERSITE='1', PYTHONDONTWRITEBYTECODE='1', PYTEST_DISABLE_PLUGIN_AUTOLOAD='1', PYTHONPATH=str(build),
+             ARM_CI_GCC=str(compiler['GCC']), ARM_CI_GXX=str(compiler['GXX']),
+             ARM_CI_COMPILER_LOG=str(artifacts/'compiler_commands.jsonl'),
+             ARM_CI_BASELINE_ROOT=str(baseline_root), ARM_CI_CANDIDATE_ROOT=str(build),
+             ARM_CI_COMPANION_SHA256=companion_sha, ARM_CI_PAIR_REPORT=str(artifacts/'api_and_companion.json'))
+  env.pop('CEREAL_FAKE', None)
+  env.pop('SCONSFLAGS', None)
+  env.pop('LD_PRELOAD', None)
+  env.pop('PYTHONOPTIMIZE', None)
+
+  def run_logged(name, command, cwd, command_env=env):
+    with (artifacts/(name+'.log')).open('w') as log:
+      log.write(json.dumps({'cwd': str(cwd), 'argv': command})+'\n')
+      log.flush()
+      subprocess.run(command, cwd=cwd, env=command_env, stdout=log, stderr=subprocess.STDOUT, check=True, timeout=300)
+
+  run_logged('build', [sys.executable, '-m', 'SCons', '--minimal', '--cache-disable', '-j2', 'msgq/ipc_pyx.so'], build)
+  candidate = build/'msgq/ipc_pyx.so'
+  run_logged('static_abi', [sys.executable, str(HERE/'validate_elf.py'), '--baseline', str(baseline), '--candidate', str(candidate), '--report', str(artifacts/'elf_comparison.json')], output)
+  for name, package in [('baseline', baseline_root), ('candidate', build)]:
+    with tempfile.TemporaryDirectory(prefix='msgq_armci_', dir='/dev/shm') as namespace:
+      test_env = dict(env, PYTHONPATH=str(package), OPENPILOT_PREFIX=Path(namespace).name.removeprefix('msgq_'))
+      run_logged(name+'_tests', [sys.executable, '-m', 'pytest', '--noconftest', '--rootdir', str(output), '-p', 'no:cacheprovider', '-o', 'addopts=', '-q', '--junitxml', str(artifacts/(name+'_tests.xml')), str(package/'msgq/tests')], output, test_env)
+  run_logged('mixed_version_tests', [sys.executable, '-m', 'pytest', '--noconftest', '--rootdir', str(output), '-p', 'no:cacheprovider', '-o', 'addopts=', '-q', '--junitxml', str(artifacts/'mixed_version_tests.xml'), str(HERE/'test_native_pair.py')], output)
+  generated = {str(p.relative_to(build)): {'size': p.stat().st_size, 'sha256': sha(p)} for p in sorted(build.rglob('*'))
+               if p.is_file() and (p.suffix in ('.o', '.a', '.so') or p.name == 'ipc_pyx.cpp')}
+  (artifacts/'generated_files.json').write_text(json.dumps(generated, indent=2)+'\n')
+  shutil.copy2(candidate, artifacts/'ipc_pyx.so')
+  shutil.copy2(build/'msgq/ipc_pyx.cpp', artifacts/'ipc_pyx.cpp')
+  check(all(sha(build/p) == info['sha256'] for p, info in source_manifest.items()), 'Build modified an input source file')
+  check(not capture('git', 'status', '--porcelain', '--untracked-files=all', cwd=source), 'Build changed checkout, including untracked paths')
+  (artifacts/'RESULT.json').write_text(json.dumps({'status': 'native_arm_ci_checks_passed', 'candidate_sha256': sha(candidate),
+                                                 'device_qualification': 'NOT ESTABLISHED; explicit artifact/toolchain/runtime review required',
+                                                 'receipts_modified': False, 'deployed': False}, indent=2)+'\n')
+  print(artifacts)
+
+
+if __name__ == '__main__':
+  main()

tools/ci/card_stall_native_artifact/build_native.sh

@@ -0,0 +1,9 @@
+#!/usr/bin/env bash
+# Artifact-only preparation. Run only after explicit CI execution approval.
+set -euo pipefail
+export PYTHONDONTWRITEBYTECODE=1
+: "${PYTHON:?Set the absolute native ARM CPython 3.12.13 interpreter}"
+: "${EXPECTED_SOURCE_COMMIT:?Set the exact approved diagnostics commit SHA}"
+: "${EXPECTED_BASE_COMMIT:?Set the exact reviewed e838 baseline commit SHA}"
+: "${EXPECTED_GCC_VERSION:?Set the explicitly reviewed GCC 13 full version}"
+exec "$PYTHON" "$(dirname "$0")/build_native.py" "$@"

tools/ci/card_stall_native_artifact/compiler_wrapper.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+"""Use only the preflight-approved GCC 13, recording effective ISA flags."""
+import json
+import os
+from pathlib import Path
+import sys
+
+name = Path(sys.argv[0]).name
+compiler = os.environ['ARM_CI_GXX' if name in ('g++', 'c++') else 'ARM_CI_GCC']
+compatibility_flags = ['-fno-sized-deallocation'] if name in ('g++', 'c++') else []
+command = [compiler, '-march=armv8-a', '-mtune=generic', *compatibility_flags, *sys.argv[1:]]
+if any(flag == '-fsized-deallocation' or flag.startswith('-mcpu=') or (flag.startswith('-march=') and flag != '-march=armv8-a') or
+       (flag.startswith('-mtune=') and flag != '-mtune=generic') for flag in sys.argv[1:]):
+  raise SystemExit('CPU-specific compiler flags are prohibited for this artifact')
+fd = os.open(os.environ['ARM_CI_COMPILER_LOG'], os.O_APPEND | os.O_CREAT | os.O_WRONLY, 0o600)
+try:
+  os.write(fd, (json.dumps({'cwd': os.getcwd(), 'argv': command})+'\n').encode())
+finally:
+  os.close(fd)
+os.execv(compiler, command)

tools/ci/card_stall_native_artifact/frozen-source.json

@@ -0,0 +1,58 @@
+{
+  "parent_commit": "e8389b0b7529bad402f1f4e5fbd672ff13d9d382",
+  "expected_branch": "diagnostics/card-stall-build-20261012",
+  "source_whitelist": [
+    {
+      "path": "openpilot/selfdrive/car/card.py",
+      "sha256": "41694a8ca800b0c850cafd18cf902f83a7155d4c15405d0a44bbde3efd8269a2",
+      "git_mode": "100755",
+      "baseline_sha256": "14e2dcc0aa4e4a945677a41b35dd599ea6d9f8bf7c6a449e74b362ef9203045f"
+    },
+    {
+      "path": "openpilot/starpilot/card_stall_timing.py",
+      "sha256": "268d67c8ebdf07a83c254fe4cde04bb3fb56f5cd5bc793da57b6e8c2370a8063",
+      "git_mode": "100644",
+      "baseline_sha256": null
+    },
+    {
+      "path": "openpilot/starpilot/aol/tests/test_card_stall_timing.py",
+      "sha256": "dc56cd86346da6cd790475d055c9f873663ce5342d46cbabda196aa358efccf6",
+      "git_mode": "100644",
+      "baseline_sha256": null
+    },
+    {
+      "path": "msgq_repo/msgq/ipc_pyx.pyx",
+      "sha256": "344491d95253e677b1190d7d9440edbb75d2067bb2c41e1919a21a244566d4ef",
+      "git_mode": "100644",
+      "baseline_sha256": "02cf21d0232967eed7bd68ed0209e55f5f1e3edef3675671bc6524ca64976fa4"
+    },
+    {
+      "path": "msgq_repo/msgq/tests/test_receive_timing.py",
+      "sha256": "88b7f736c7cbd3d0d0c29634df35112f9eca8f10236a3561afce6683e88f7006",
+      "git_mode": "100644",
+      "baseline_sha256": null
+    }
+  ],
+  "baseline_native": {
+    "path": "msgq_repo/msgq/ipc_pyx.so",
+    "git_blob": "3453001a9e283d60ae4489607beafaa6776acd1e",
+    "sha256": "5a50085f727e3b22dcf596cb42be351066b0dcf7db7866fe04be40e69fe57133",
+    "git_mode": "100755"
+  },
+  "python_version": "3.12.13",
+  "cython_version": "3.3.0",
+  "compiler_family_major": "GCC 13",
+  "generic_isa": "armv8-a",
+  "max_symbol_versions": {
+    "GLIBC": "2.34",
+    "GLIBCXX": "3.4.32",
+    "CXXABI": "1.3",
+    "GCC": "3.0"
+  },
+  "baseline_companion": {
+    "path": "msgq_repo/msgq/visionipc/visionipc_pyx.so",
+    "git_blob": "b4f95da4ec84c15a515d56dea0f028c37674d083",
+    "sha256": "d9aee015a9ca9d6d8f4dc54c5bca6a8f3ab24ad9a42477561ea7cc32eebff5a4",
+    "git_mode": "100755"
+  }
+}

tools/ci/card_stall_native_artifact/requirements.lock

@@ -0,0 +1,10 @@
+# Minimal ARM build/test closure pinned to repository uv.lock.
+# Install with: python -m pip install --require-hashes -r requirements.lock
+# No project extras, model dependencies, or editable workspace installation.
+cython==3.3.0 --hash=sha256:7e321ae700995a16dc3055ada06ffb8d61e1a7434e5d0e811547a45ac1015ebd --hash=sha256:9b24b5c8cd536946b62086fcafee6d5509d3f549f72d553d2336af87ffbe0da1
+scons==4.11.1 --hash=sha256:454cef364348053422696e3d2ecb4fa593c96a624f955842eaaea64f95c8d11d
+pytest==8.4.2 --hash=sha256:872f880de3fc3a5bdc88a11b39c9710c3497a547cfa9320bc3c5e62fbf272e79
+iniconfig==2.3.0 --hash=sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12
+packaging==26.3 --hash=sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c
+pluggy==1.6.0 --hash=sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746
+pygments==2.21.0 --hash=sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9

tools/ci/card_stall_native_artifact/test_native_pair.py

@@ -0,0 +1,147 @@
+"""Run on native ARM CPython 3.12.13; never substitute a rebuilt companion."""
+import hashlib
+import json
+import os
+from pathlib import Path
+import platform
+import select
+import subprocess
+import sys
+import tempfile
+import uuid
+
+import pytest
+
+PAYLOADS = [b'x', b'unchanged\x00binary', bytes(range(256)), b'z'*4096] * 32
+
+
+def exact_runtime():
+  assert platform.machine() == 'aarch64', platform.machine()
+  assert sys.implementation.name == 'cpython' and sys.version_info[:3] == (3, 12, 13), sys.version
+
+
+def child_env(root, prefix):
+  env = dict(os.environ, PYTHONPATH=str(root), PYTHONNOUSERSITE='1', OPENPILOT_PREFIX=prefix)
+  env.pop('CEREAL_FAKE', None)
+  return env
+
+
+def read_line(process, timeout=15):
+  if not select.select([process.stdout], [], [], timeout)[0]:
+    raise TimeoutError('native IPC subprocess did not produce its expected line')
+  line = process.stdout.readline()
+  if not line:
+    raise RuntimeError(f'native IPC subprocess exited before its expected line: {process.poll()}')
+  return line.decode().rstrip('\n')
+
+
+def roots():
+  return {name: Path(os.environ['ARM_CI_'+name.upper()+'_ROOT']) for name in ('baseline', 'candidate')}
+
+
+def test_public_api_and_unchanged_companion():
+  exact_runtime()
+  results = {}
+  for name, root in roots().items():
+    result = subprocess.run([sys.executable, __file__, 'api'], env=child_env(root, 'api_'+uuid.uuid4().hex),
+                            check=True, text=True, capture_output=True, timeout=20)
+    results[name] = json.loads(result.stdout)
+    assert results[name]['companion_sha256'] == os.environ['ARM_CI_COMPANION_SHA256']
+    assert Path(results[name]['ipc_path']).resolve() == (root/'msgq/ipc_pyx.so').resolve()
+    assert Path(results[name]['companion_path']).resolve() == (root/'msgq/visionipc/visionipc_pyx.so').resolve()
+  baseline, candidate = results['baseline'], results['candidate']
+  for name, methods in baseline['api'].items():
+    assert set(methods) <= set(candidate['api'][name]), name
+  assert {'enable_receive_timing', 'reset_receive_timing', 'take_receive_timing'} <= set(candidate['api']['SubSocket'])
+  Path(os.environ['ARM_CI_PAIR_REPORT']).write_text(json.dumps(results, indent=2)+'\n')
+
+
+@pytest.mark.parametrize('publisher,subscriber,timing', [
+  ('baseline', 'baseline', False), ('baseline', 'candidate', False),
+  ('baseline', 'candidate', True), ('candidate', 'baseline', False), ('candidate', 'candidate', True),
+])
+def test_interprocess_payload_equivalence(publisher, subscriber, timing):
+  exact_runtime()
+  packages = roots()
+  with tempfile.TemporaryDirectory(prefix='msgq_armci_', dir='/dev/shm') as namespace:
+    prefix = Path(namespace).name.removeprefix('msgq_')
+    endpoint = 'abi_'+uuid.uuid4().hex
+    processes = []
+    try:
+      def start(role, root, enabled):
+        process = subprocess.Popen([sys.executable, __file__, role, endpoint, str(int(enabled))],
+                                   env=child_env(root, prefix), stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                                   stderr=subprocess.PIPE, bufsize=0)
+        processes.append(process)
+        assert read_line(process) == 'ready'
+        return process
+      pub = start('publish', packages[publisher], False)
+      sub = start('subscribe', packages[subscriber], timing)
+      pub.stdin.write(b'go\n')
+      pub.stdin.flush()
+      result = json.loads(read_line(sub))
+      assert result['sha256'] == hashlib.sha256(b''.join(PAYLOADS)).hexdigest()
+      assert result['count'] == len(PAYLOADS)
+      assert result['bytes'] == sum(map(len, PAYLOADS))
+      assert sub.wait(timeout=10) == 0
+      pub.stdin.write(b'done\n')
+      pub.stdin.flush()
+      assert pub.wait(timeout=10) == 0
+    finally:
+      for process in processes:
+        if process.poll() is None:
+          process.kill()
+        process.wait(timeout=10)
+        stderr = process.stderr.read().decode(errors='replace')
+        if stderr:
+          print(stderr, file=sys.stderr)
+
+
+def child():
+  exact_runtime()
+  import msgq
+  import msgq.ipc_pyx
+  role = sys.argv[1]
+  if role == 'api':
+    # A missing dependency is a blocking import error, never a reason to build
+    # or swap this companion, whose reviewed bytes must remain unchanged.
+    import msgq.visionipc.visionipc_pyx as companion
+    names = ('Context', 'SubSocket', 'PubSocket', 'Poller', 'Event', 'SocketEventHandle')
+    print(json.dumps({'api': {n: sorted(x for x in dir(getattr(msgq.ipc_pyx, n)) if not x.startswith('_')) for n in names},
+                      'ipc_path': msgq.ipc_pyx.__file__, 'companion_path': companion.__file__,
+                      'companion_sha256': hashlib.sha256(Path(companion.__file__).read_bytes()).hexdigest(),
+                      'python': sys.version, 'machine': platform.machine()}), flush=True)
+    return
+  endpoint, timing = sys.argv[2], bool(int(sys.argv[3]))
+  if role == 'publish':
+    sock = msgq.pub_sock(endpoint)
+    print('ready', flush=True)
+    assert input() == 'go'
+    for payload in PAYLOADS:
+      sock.send(payload)
+    assert input() == 'done'
+  elif role == 'subscribe':
+    sock = msgq.sub_sock(endpoint, timeout=2000)
+    if timing:
+      sock.enable_receive_timing(True)
+    print('ready', flush=True)
+    received = [sock.receive() for _ in PAYLOADS]
+    assert received == PAYLOADS
+    assert sock.receive(non_blocking=True) is None
+    if hasattr(sock, 'take_receive_timing'):
+      records, omitted = sock.take_receive_timing()
+      if timing:
+        assert len(records) == 64 and omitted == len(PAYLOADS)+1-64
+        assert records[-1][:4] == (len(PAYLOADS)+1, True, False, 0)
+        assert all(len(r) == 12 and min(r[4:]) >= 0 for r in records)
+        assert all(list(r[4::2]) == sorted(r[4::2]) and list(r[5::2]) == sorted(r[5::2]) for r in records)
+      else:
+        assert (records, omitted) == ((), 0)
+    print(json.dumps({'count': len(received), 'bytes': sum(map(len, received)),
+                      'sha256': hashlib.sha256(b''.join(received)).hexdigest()}), flush=True)
+  else:
+    raise ValueError(role)
+
+
+if __name__ == '__main__':
+  child()

tools/ci/card_stall_native_artifact/validate_elf.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+"""Static, fail-closed comparison. Does not execute either shared library."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import re
+import struct
+import subprocess
+
+BASELINE_SHA256 = '5a50085f727e3b22dcf596cb42be351066b0dcf7db7866fe04be40e69fe57133'
+
+
+def command(*args):
+  return subprocess.check_output(args, text=True)
+
+
+def inspect(path):
+  data = path.read_bytes()
+  if len(data) < 64 or data[:6] != b'\x7fELF\x02\x01' or struct.unpack_from('<HH', data, 16) != (3, 183):
+    raise ValueError(f'{path}: requires ELF64 little-endian AArch64 shared object')
+  dynamic = command('readelf', '--wide', '-d', str(path))
+  versions = command('readelf', '--wide', '--version-info', str(path))
+  symbols = {}
+  for line in command('nm', '-D', '--defined-only', '--format=posix', str(path)).splitlines():
+    fields = line.split()
+    if len(fields) >= 2:
+      symbols[fields[0]] = fields[1]
+  names = sorted(symbols)
+  demangled = subprocess.check_output(['c++filt'], input='\n'.join(names)+'\n', text=True).splitlines()
+  undefined = command('nm', '-D', '--undefined-only', '--format=posix', str(path))
+  version_names = sorted(set(re.findall(r'Name: ([A-Za-z0-9_.]+)', versions)))
+  return {
+    'path': str(path), 'sha256': hashlib.sha256(data).hexdigest(), 'size': len(data),
+    'needed': sorted(re.findall(r'\(NEEDED\).*?\[(.*?)\]', dynamic)),
+    'rpaths': re.findall(r'\((?:RUNPATH|RPATH)\).*?\[(.*?)\]', dynamic),
+    'versions': version_names, 'defined_symbols': dict(zip(names, demangled, strict=True)),
+    'undefined_symbols': undefined.splitlines(),
+  }
+
+
+def version_max(info, namespace):
+  values = [tuple(map(int, name[len(namespace)+1:].split('.'))) for name in info['versions'] if re.fullmatch(re.escape(namespace)+r'_[0-9]+(?:\.[0-9]+)*', name)]
+  return max(values, default=(0,))
+
+
+def project_symbol(name):
+  classes = r'(?:Context|Message|SubSocket|PubSocket|Poller|MSGQSubSocket|MSGQPoller|FakePoller|FakeSubSocket|SocketEventHandle|Event)'
+  return bool(re.match(r'^(?:PyInit_ipc_pyx$|msgq_|messaging_use_fake\(|sigusr2_handler\(|event_state_shm_mmap\(|'
+                       + classes+r'(?:::|<)|(?:typeinfo(?: name)?|vtable) for '+classes+r'(?:$|<))', name))
+
+
+def main():
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('--baseline', type=Path, required=True)
+  parser.add_argument('--candidate', type=Path, required=True)
+  parser.add_argument('--report', type=Path, required=True)
+  args = parser.parse_args()
+  baseline, candidate = inspect(args.baseline), inspect(args.candidate)
+  errors = []
+  if baseline['sha256'] != BASELINE_SHA256:
+    errors.append('baseline hash does not match the reviewed e838 tracked artifact')
+  unexpected = sorted(set(candidate['needed']) - set(baseline['needed']))
+  if unexpected:
+    errors.append(f'new dynamic dependencies: {unexpected}')
+  # Standalone linking needs no runtime search path, so reject all rather than
+  # maintaining a fragile allow-list of possible build-host paths.
+  if candidate['rpaths']:
+    errors.append(f'candidate must not embed RPATH/RUNPATH: {candidate["rpaths"]}')
+  for namespace, ceiling in {'GLIBC': (2, 34), 'GLIBCXX': (3, 4, 32), 'CXXABI': version_max(baseline, 'CXXABI'), 'GCC': version_max(baseline, 'GCC')}.items():
+    ceiling = min(ceiling, version_max(baseline, namespace))
+    if version_max(candidate, namespace) > ceiling:
+      errors.append(f'{namespace} requirement exceeds {ceiling}')
+  nonnumeric_versions = [v for v in candidate['versions'] if not re.fullmatch(r'[A-Z][A-Z0-9]*_[0-9]+(?:\.[0-9]+)*', v)]
+  if nonnumeric_versions:
+    errors.append(f'unreviewed nonnumeric symbol versions: {nonnumeric_versions}')
+  unexpected_versions = {v.split('_', 1)[0] for v in candidate['versions']} - {v.split('_', 1)[0] for v in baseline['versions']}
+  if unexpected_versions:
+    errors.append(f'new symbol-version namespace: {sorted(unexpected_versions)}')
+  bsyms, csyms = baseline['defined_symbols'], candidate['defined_symbols']
+  missing_project = {n: d for n, d in bsyms.items() if project_symbol(d) and n not in csyms}
+  if missing_project:
+    errors.append(f'missing existing msgq/extension exported symbols: {missing_project}')
+  report = {
+    'qualification': 'Static ABI boundary checks only; native tests and deployment review remain separate.',
+    'baseline': baseline, 'candidate': candidate, 'errors': errors,
+    'removed_exports': {n: bsyms[n] for n in sorted(set(bsyms)-set(csyms))},
+    'added_exports': {n: csyms[n] for n in sorted(set(csyms)-set(bsyms))},
+  }
+  args.report.write_text(json.dumps(report, indent=2, sort_keys=True)+'\n')
+  if errors:
+    raise SystemExit('\n'.join(errors))
+  print('AArch64 ELF, dependency, search-path, symbol-version and existing msgq-export checks passed.')
+
+
+if __name__ == '__main__':
+  main()

tools/ci/card_stall_native_artifact/verify_source.py

@@ -0,0 +1,57 @@
+#!/usr/bin/env python3
+"""Read-only source/manifest gate for the isolated native-artifact branch."""
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+root = Path.cwd().resolve()
+output = Path(sys.argv[1]).resolve()
+output.mkdir(parents=True, exist_ok=False)
+manifest = json.loads(Path(__file__).with_name('frozen-source.json').read_text())
+
+def git(*args):
+  return subprocess.check_output(['git', '-c', 'core.hooksPath=/dev/null', *args], cwd=root)
+
+def digest(raw):
+  return hashlib.sha256(raw).hexdigest()
+
+head = git('rev-parse', 'HEAD').decode().strip()
+assert head == os.environ['GITHUB_SHA'], 'checkout is not the exact event commit'
+assert os.environ['GITHUB_REF'] == 'refs/heads/' + manifest['expected_branch']
+assert os.environ['GITHUB_REPOSITORY'] == 'firestar5683/StarPilot'
+base = manifest['parent_commit']
+git('cat-file', '-e', base + '^{commit}')
+changed = git('diff', '--name-only', base, head).decode().splitlines()
+source_paths = {row['path'] for row in manifest['source_whitelist']}
+allowed = lambda p: p in source_paths or p == '.github/workflows/card_stall_native_artifact.yaml' or p.startswith('tools/ci/card_stall_native_artifact/')
+assert all(allowed(p) for p in changed), changed
+assert source_paths <= set(changed)
+rows = []
+for entry in manifest['source_whitelist']:
+  path = entry['path']
+  data = (root / path).read_bytes()
+  assert digest(data) == entry['sha256'], path
+  tree = git('ls-tree', head, '--', path).decode().strip().split()
+  assert tree[0] == entry['git_mode'], (path, tree)
+  assert digest(git('show', head + ':' + path)) == entry['sha256']
+  rows.append({'path': path, 'sha256': digest(data), 'git_mode': tree[0], 'git_blob': tree[2]})
+for key in ('baseline_native', 'baseline_companion'):
+  entry = manifest[key]
+  data = git('show', base + ':' + entry['path'])
+  tree = git('ls-tree', base, '--', entry['path']).decode().strip().split()
+  assert tree[0] == entry['git_mode'] and tree[2] == entry['git_blob']
+  assert digest(data) == entry['sha256']
+  assert (root / entry['path']).read_bytes() == data, 'unexpected already-replaced native artifact'
+  (output / ('baseline-ipc_pyx.so' if key == 'baseline_native' else 'baseline-visionipc_pyx.so')).write_bytes(data)
+assert not (root / 'openpilot/selfdrive/modeld/models').exists(), 'model payload checkout is outside this job'
+source_tree = git('ls-tree', '-rz', '--full-tree', head)
+(output / 'source-tree.git.bin').write_bytes(source_tree)
+report = {'head': head, 'base': base, 'tree': git('rev-parse', 'HEAD^{tree}').decode().strip(),
+          'changed_paths': changed, 'source_tree_sha256': digest(source_tree), 'frozen_sources': rows,
+          'baseline_native': manifest['baseline_native'], 'baseline_companion': manifest['baseline_companion'],
+          'workflow': '.github/workflows/card_stall_native_artifact.yaml', 'source_only_sparse_checkout': True}
+(output / 'source.json').write_text(json.dumps(report, indent=2) + '\n')
+print(json.dumps(report, indent=2))