Add Heatdeath Controls Benchmark

heatdeath

@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+set -euo pipefail
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PYTHON_BIN="${HEATDEATH_PYTHON:-python3}"
+if [[ -z "${HEATDEATH_PYTHON:-}" && -x /usr/local/venv/bin/python3 ]]; then
+  PYTHON_BIN=/usr/local/venv/bin/python3
+fi
+exec "$PYTHON_BIN" -B "$ROOT_DIR/tools/heatdeath/heatdeath.py" "$@"

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "9742943ea33307b48b47dbdcba8a6390aeba9077c821c543e398374cd193e6bb", "schema_version": 1}
+{"protected_inventory_sha256": "ac7c85b608097912bf473e5e66eaf4006cd3642ba874d7b5a5a1f978faeb8cf9", "schema_version": 1}

tools/STARPILOT_DEVELOPMENT.md

@@ -20,6 +20,7 @@ cache. The UI remains the custom large and compact Raylib UI.
 | `./dev pytest <arguments>` | pytest, including its xdist and mock plugins |
 | `./dev shell` | Shell with host imports and private settings |
 | `./dev sync [shared\|cabana]` | Refresh source and dependencies without launching a tool |
+| `./heatdeath --help` | Joined Controls replay benchmark and CPU-placement experiments; [setup](heatdeath/README.md) |
 | `./test` | Run lint, packaging, startup, control, model, UI and selected vehicle regressions |
 | `./test --full` | Run the complete Linux pre-deployment suite, including recorded routes and replay |
 

tools/ci/tests/test_heatdeath.py

@@ -0,0 +1,204 @@
+"""Portable qualification of the exported benchmark control plane and reporting."""
+
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import struct
+import subprocess
+import sys
+import tempfile
+import types
+import unittest
+import time
+import uuid
+from unittest.mock import patch
+
+HERE = Path(__file__).resolve().parents[2] / 'heatdeath'
+# Load as a distinct package so this suite does not leak generic helper module
+# names or change sys.path for the rest of the full-repository test process.
+package_spec = importlib.util.spec_from_file_location('heatdeath_bench_tests_package', HERE / '__init__.py', submodule_search_locations=[str(HERE)])
+package = importlib.util.module_from_spec(package_spec)
+sys.modules[package_spec.name] = package
+package_spec.loader.exec_module(package)
+from heatdeath_bench_tests_package import actor, heatdeath as cli, placement, report
+
+install = actor.install
+apply, load_plan, parked = placement.apply, placement.load_plan, placement.parked
+phase, summarize = report.phase, report.summarize
+
+
+class HeatdeathTests(unittest.TestCase):
+  def test_cli_help_needs_no_device_or_native_import(self):
+    for command in ([], ['prepare'], ['build'], ['run'], ['report']):
+      result = subprocess.run([sys.executable, '-B', str(HERE / 'heatdeath.py'), *command, '--help'], capture_output=True, text=True, timeout=5)
+      self.assertEqual(result.returncode, 0, result.stderr)
+
+  def test_placement_rejects_malformed_or_foreign_actors(self):
+    for value in ('[]', '{"manager":[1]}', '{"card":[]}', '{"card":[true]}', '{"card":[8]}', '{"card":[-1]}', '{"card":[1,1]}', '{"card":"6"}'):
+      with self.subTest(value=value), self.assertRaises(ValueError):
+        load_plan(value)
+    self.assertEqual(load_plan('{"controlsd":[6],"peer":[0,1]}'), {'controlsd': [6], 'peer': [0, 1]})
+
+  def test_affinity_override_after_both_production_placement_calls(self):
+    seen = []
+    module = types.SimpleNamespace(config_realtime_process=lambda *args: seen.append(args), set_core_affinity=lambda value: seen.append(value))
+    with patch.object(actor, 'apply') as requested:
+      install(module, 'controlsd')
+      module.config_realtime_process(4, 53)
+      module.set_core_affinity([6])
+      self.assertEqual(seen, [(4, 53), [6]])
+      self.assertEqual(requested.call_count, 2)
+      requested.assert_called_with('controlsd')
+
+  def test_apply_only_changes_own_process_and_verifies(self):
+    with (
+      patch.dict(os.environ, HEATDEATH_PLACEMENT='{"card":[4]}'),
+      patch.object(placement.os, 'sched_setaffinity', create=True) as setter,
+      patch.object(placement.os, 'sched_getaffinity', return_value={4}, create=True),
+    ):
+      self.assertEqual(apply('card'), [4])
+      setter.assert_called_once_with(0, {4})
+      setter.reset_mock()
+      apply('peer')
+      setter.assert_not_called()
+
+  def test_offroad_and_engaged_are_both_required(self):
+    for offroad, engaged, accepted in ((b'1', b'0', True), (b'0', b'0', False), (b'1', b'1', False)):
+      with self.subTest(offroad=offroad, engaged=engaged), patch.object(Path, 'read_bytes', side_effect=[offroad, engaged]):
+        if accepted:
+          parked()
+        else:
+          with self.assertRaises(RuntimeError):
+            parked()
+
+  def test_phase_reports_actual_duration_and_gap(self):
+    rows = [(i, 10, i, stamp, stamp + 2_000_000, 0, 1_000_000, 0) for i, stamp in enumerate((0, 10_000_000, 30_000_000))]
+    result = phase(rows, 0, 1_000_000_000)
+    self.assertEqual(result['observedCallsPerSecond'], 3)
+    self.assertEqual(result['startGapMsMax'], 20)
+    self.assertEqual(result['wallMsP99'], 2)
+    self.assertEqual(result['threadCpuMsP95'], 1)
+
+  def test_report_checks_hash_and_ring_overflow(self):
+    with tempfile.TemporaryDirectory() as temp:
+      out = Path(temp)
+      (out / 'release.bin').write_bytes(struct.pack('<Q', 1))
+      result = {
+        'sourcePins': {'sourceHead': 'abc'},
+        'sourceDurationNs': 1_000_000_000,
+        'failure': None,
+        'everAuthorityReady': False,
+        'activeObserverSamples': 0,
+        'placement': {},
+        'syntheticReserve': None,
+        'limits': [],
+      }
+      (out / 'result.json').write_text(json.dumps(result))
+      raw = struct.pack('<' + 'q' * 20, 0, 1, 0, 1, 2, 0, 1, 0, *([0] * 12))
+      (out / 'timing.bin').write_bytes(raw)
+      entry = {'path': '/old/path/timing.bin', 'sha256': hashlib.sha256(raw).hexdigest(), 'rowsDropped': 0, 'rowsRetained': 1}
+      for actor in ('card', 'controlsd', 'selfdrived'):
+        (out / f'{actor}-timing.json').write_text(json.dumps({'timing': {'step': entry}, 'sampleColumns': {}}))
+      report = summarize(out, 0)
+      self.assertEqual(report['actors']['controlsd']['step']['calls'], 1)
+      self.assertFalse(report['qualifiedActiveLatencyBench'])
+      self.assertFalse(report['everAuthorityReady'])
+      with self.assertRaises(ValueError):
+        summarize(out, 1)
+      (out / 'timing.bin').write_bytes(b'wrong')
+      with self.assertRaisesRegex(ValueError, 'hash'):
+        summarize(out, 0)
+      (out / 'timing.bin').write_bytes(raw)
+      entry['rowsDropped'] = 1
+      (out / 'card-timing.json').write_text(json.dumps({'timing': {'step': entry}, 'sampleColumns': {}}))
+      with self.assertRaisesRegex(ValueError, 'lost rows'):
+        summarize(out, 0)
+
+  def test_bundle_refuses_modified_helper_or_changed_source(self):
+    with tempfile.TemporaryDirectory() as temp:
+      bundle = Path(temp)
+      (bundle / 'packet').mkdir()
+      (bundle / 'owned.py').write_text('original')
+      (bundle / 'packet/joined-source-pins.json').write_text('{"sources":{}}')
+      info = {'repo': temp, 'sourceHead': 'abc', 'files': {'owned.py': cli.sha(bundle / 'owned.py')}}
+      (bundle / 'bundle.json').write_text(json.dumps(info))
+      with patch.object(cli, 'clean', return_value='abc'):
+        cli.verify(bundle)
+        (bundle / 'owned.py').write_text('changed')
+        with self.assertRaisesRegex(RuntimeError, 'Bundle changed'):
+          cli.verify(bundle)
+      with patch.object(cli, 'clean', return_value='other'):
+        with self.assertRaisesRegex(RuntimeError, 'revision'):
+          cli.verify(bundle)
+
+  def test_all_local_imports_resolve_without_private_audit_packet(self):
+    import ast
+
+    for source in HERE.glob('*.py'):
+      tree = ast.parse(source.read_text())
+      for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom) and node.module and node.module.startswith(('driver_', 'peer_', 'owned_', 'bounded_', 'phase_')):
+          self.assertTrue((HERE / (node.module + '.py')).is_file(), node.module)
+
+  @unittest.skipUnless(sys.platform == 'linux' and os.geteuid() == 0, 'Root Linux process-identity check')
+  def test_actual_owned_child_reaped_without_touching_foreign_child(self):
+    token = uuid.uuid4().hex
+    owned = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'], env=dict(os.environ, SP_PRIVATE_IRQ_LEASE_TOKEN=token))
+    foreign = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'])
+    try:
+      self.assertIn(owned.pid, cli.token_processes(token))
+      cli.reap_token(token)
+      owned.wait(timeout=3)
+      self.assertIsNone(foreign.poll())
+      self.assertEqual(cli.token_processes(token), [])
+    finally:
+      for child in (owned, foreign):
+        if child.poll() is None:
+          child.kill()
+        child.wait(timeout=3)
+
+  @unittest.skipUnless(sys.platform == 'linux', 'Linux synthetic workload')
+  def test_actual_bounded_synthetic_reserve(self):
+    with tempfile.TemporaryDirectory() as temp:
+      root = Path(temp)
+      release, output = root / 'release.bin', root / 'reserve.json'
+      release.write_bytes(struct.pack('<Q', time.monotonic_ns() + 100_000_000))
+      subprocess.run(
+        [
+          sys.executable,
+          '-B',
+          str(HERE / 'reserve.py'),
+          '--core',
+          str(min(os.sched_getaffinity(0))),
+          '--cpu-ms',
+          '1',
+          '--period-ms',
+          '10',
+          '--duration-ns',
+          '40000000',
+          '--release',
+          str(release),
+          '--output',
+          str(output),
+        ],
+        check=True,
+        timeout=5,
+      )
+      result = json.loads(output.read_text())
+      self.assertTrue(result['completed'])
+      self.assertEqual(len(result['rows']), 4)
+      self.assertEqual(result['kind'], 'synthetic-cfs-cpu-reserve')
+      self.assertTrue(all(row['cpuNs'] >= 0 for row in result['rows']))
+
+  def test_output_refuses_source_or_existing_paths(self):
+    with tempfile.TemporaryDirectory() as temp:
+      repo = Path(temp)
+      for output in (repo / 'new', repo):
+        with self.assertRaises(ValueError):
+          cli.external(output, repo)
+
+
+if __name__ == '__main__':
+  unittest.main()

tools/heatdeath/README.md

@@ -0,0 +1,63 @@
+# Heatdeath
+
+`./heatdeath` exports the joined replay fixture used during StarPilot Controls CPU optimization. It runs actual Card, SelfdriveD, Controls, RadarD and planner processes, feeds recorded CAN and peer messages at their recorded cadence, and captures bounded timing and native-safety evidence. It changes only owned benchmark processes' affinity. It does not change production scheduling, IRQ affinity, CPU governors, online cores, installed preferences or source files.
+
+This first export retains the existing **active-long Ioniq 6 steady-state fixture**. It is a CPU-placement benchmark, not a general vehicle replay test. It explicitly restores the recorded remembered cruise speed and post-ECU-handoff initialization phase; it never restores an armed latch, command, native permission or healthy receipt. Discovery, physical ECU handoff and startup performance are excluded. Native permission must be earned by fresh replay inputs.
+
+## Requirements
+
+- A parked, offroad Linux ARM64 comma bench device with the normal compiled StarPilot dependencies and development compiler installed. Build the checkout normally before preparation. Execution requires root for the production realtime setup and private namespaces.
+- A clean source checkout. A bundle is pinned to that exact revision and source hashes. Prepare/build a separate bundle when comparing a different revision.
+- Local full rlogs, including startup CarParams, from a compatible Ioniq 6 active-long drive. Recordings and compiled binaries are not bundled in Git.
+- An explicit JSON preference snapshot, mapping parameter names to base64 bytes. Provide the actual intended benchmark settings, including `OpenpilotEnabledToggle=1` and `IsOffroad=0` for the **private** Params namespace. Production readiness/CarParams keys cannot be supplied. No physical device preferences are changed.
+- Explicit source clock context: the source boot-minus-monotonic offset and whether `deviceMotion` timestamps use monotonic or boottime. Do not infer this from your current device clock.
+
+The root command selects `/usr/local/venv/bin/python3` when installed; override with `HEATDEATH_PYTHON`. Run it directly, outside `./dev`, which can enable simulation environment variables. `./heatdeath --help` works without native dependencies or root.
+
+## Prepare and build
+
+Use a 60-second window with two clean, stable cruise-speed samples immediately before its start. Times below are examples; choose the timestamps from your recording. Include startup and any adjacent rlogs needed to cover the window.
+
+```sh
+./heatdeath prepare \
+  --rlog /data/bench/startup/rlog.zst \
+  --rlog /data/bench/window/rlog.zst \
+  --start-mono-ns 220000000000 --duration 60 \
+  --source-boot-minus-mono-ns 0 --device-motion-clock monotonic \
+  --params-json /data/bench/params.json \
+  --output /data/bench/heatdeath-bundle
+
+sudo ./heatdeath build --bundle /data/bench/heatdeath-bundle
+```
+
+Use zero boot offset only when verified appropriate for the recording. The bundle retains input hashes, actual CarParams, source/helper hashes and the explicit checkpoint. Build compiles a virtual transport around current production pandad and its safety test library. USB/SPI implementations are not linked; it cannot send commands to the vehicle. Build output remains in the external bundle, with a receipt and logs.
+
+## Compare CPU placements
+
+First capture the production placement baseline, then change only the actor placements being compared. Run outputs must be new directories outside the checkout and bundle.
+
+```sh
+sudo ./heatdeath run --bundle /data/bench/heatdeath-bundle \
+  --output /data/bench/baseline-1
+
+sudo ./heatdeath run --bundle /data/bench/heatdeath-bundle \
+  --placement '{"card":[4],"controlsd":[6],"selfdrived":[5],"radard":[3,5],"plannerd":[3,5],"peer":[1],"observer":[0,1]}' \
+  --native-core2 --reserve-core 6 --reserve-ms 8 --reserve-period-ms 50 \
+  --output /data/bench/placement-1
+
+./heatdeath report --output /data/bench/placement-1 --skip-seconds 10
+```
+
+Actor names in `--placement` are optional: omitted actors retain their original fixture/production placement. Priority remains the current production policy. `--native-core2` changes only the virtual native endpoint's placement after validating its actual startup thread state; it does not move physical pandad. CPU choices must already be online and allowed. No cores are enabled automatically.
+
+The reserve is an **optional synthetic CFS CPU workload**, with measured dose and missed-pulse reporting. It is not inference, soundd or Android Auto. A clean baseline and a loaded run answer different questions. Use repeated baseline/candidate/candidate/baseline runs with identical inputs, reserve dose, peer/native placement, thermals and power state to compare distributions. Do not infer a scheduler improvement from one favorable sample.
+
+## Interpret results
+
+`report` shows phase call rate, wall-time p50/p95/p99/max, thread CPU p95, largest start gap, rolling lag flags, native receipt and active-authority observations. Raw binary capture, hashes, CPU/scheduler records, source-feed lateness, child logs and timing rings stay in the run directory. Raw observations can also be decoded using the bundled driver's `decode-observations` command.
+
+A completed command is not a headroom or steering pass. Check native RX/TX admission, sustained actual active output, source-feed completeness/lateness, source ages, lag episodes and scheduling records together. The fixture refuses unsupported/mixed CarParams and stale native/authority evidence; do not disable those checks to obtain prettier timings. Timing rings that overflow are rejected by the report.
+
+Recorded model and driver-monitoring messages are inputs. Actual model inference, camera decoding, audio and Android Auto are not launched by this harness. This benchmark cannot alone establish real co-load headroom, ECU acceptance, physical firmware-heartbeat equivalence, or road behavior. The bounded observer and virtual endpoint themselves consume CPU and RAM; include their cost.
+
+Runs stop if the physical device leaves offroad/disengaged state and have a 180-second supervisor bound. Cleanup targets only actors carrying that run's unique token and only the journaled private message namespace. A per-device lock prevents concurrent Heatdeath runs/builds. It does not coordinate other developers' unrelated bench harnesses; keep the test device exclusively assigned while measuring. Interrupt normally with Ctrl-C; do not delete another active run's files.

tools/heatdeath/__init__.py

@@ -0,0 +1 @@
+"""Offroad joined Controls benchmark and portable report helpers."""

tools/heatdeath/actor.py

@@ -0,0 +1,41 @@
+"""Run a normal daemon, optionally placing only its own realtime entrypoint."""
+
+import argparse
+import importlib
+import sys
+
+if __package__:
+  from .placement import apply
+else:
+  from placement import apply
+
+
+def install(module, name):
+  for attribute in ('config_realtime_process', 'set_core_affinity'):
+    if not hasattr(module, attribute):
+      continue
+    original = getattr(module, attribute)
+
+    def placed(*args, _original=original, **kwargs):
+      result = _original(*args, **kwargs)
+      apply(name)
+      return result
+
+    setattr(module, attribute, placed)
+
+
+def main():
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('name', choices=('radard', 'plannerd'))
+  args = parser.parse_args()
+  module = importlib.import_module('openpilot.selfdrive.controls.' + args.name)
+  if hasattr(module, 'config_realtime_process'):
+    install(module, args.name)
+  else:
+    apply(args.name)
+  module.main()
+
+
+if __name__ == '__main__':
+  sys.dont_write_bytecode = True
+  main()

tools/heatdeath/bounded_capture_v1.py

@@ -0,0 +1,36 @@
+"""Private bounded RAM capture; no disk writes while producers are timed."""
+
+from pathlib import Path
+import time
+
+
+class BoundedCapture:
+  def __init__(self, path, maximum_bytes=128 * 1024 * 1024):
+    self.path = Path(path)
+    assert not self.path.exists() and 1 <= maximum_bytes <= 256 * 1024 * 1024
+    self.storage = bytearray(maximum_bytes)
+    self.used = 0
+    self.export_ns = None
+
+  def __enter__(self):
+    return self
+
+  def __exit__(self, *_):
+    return False
+
+  def write(self, value):
+    size = len(value)
+    assert self.used + size <= len(self.storage), 'Bounded raw capture full; no drop/truncation permitted'
+    self.storage[self.used : self.used + size] = value
+    self.used += size
+
+  def export(self):
+    # Caller must reap all producers first. A slow export invalidates wrapper
+    # completion, but cannot delay CAN/peer producers or timed IPC observations.
+    start = time.monotonic_ns()
+    with self.path.open('xb') as stream:
+      view = memoryview(self.storage)
+      for offset in range(0, self.used, 1024 * 1024):
+        stream.write(view[offset : min(offset + 1024 * 1024, self.used)])
+    self.export_ns = time.monotonic_ns() - start
+    return {'bytes': self.used, 'capacityBytes': len(self.storage), 'postReapExportNs': self.export_ns, 'timedDiskWrites': False, 'drops': 0}

tools/heatdeath/checkpoint_card_v2_cruise.py

@@ -0,0 +1,275 @@
+#!/usr/bin/env python3
+"""Explicit private Ioniq6 checkpoint including pinned remembered cruise speed.
+
+No daemon method, clock, socket, native admission, or grant is replaced. This
+restores only the declared post-handoff phase. Fresh parser/panda/session/lease
+checks run normally and any subsequent authority loss remains sticky.
+"""
+
+import argparse
+import base64
+import hashlib
+import json
+import math
+import os
+from pathlib import Path
+import subprocess
+import sys
+import time
+
+
+def sha(path):
+  return hashlib.sha256(Path(path).read_bytes()).hexdigest()
+
+
+def verify_cruise_evidence(args, repo, *, verify_source):
+  from openpilot.cereal import messaging
+  from openpilot.selfdrive.car.cruise import V_CRUISE_MAX
+
+  assert sha(args.cruise_checkpoint) == args.cruise_checkpoint_sha256
+  evidence = json.loads(args.cruise_checkpoint.read_text())
+  assert evidence['kind'] == 'private-recorded-cruise-speed-checkpoint' and evidence['startupExcluded'] is True
+  assert evidence['cpSha256'] == args.car_params_sha256
+  assert evidence['carSchemaSha256'] == args.car_schema_sha256
+  head = subprocess.check_output(['git', '-c', 'safe.directory=' + str(repo), '-C', str(repo), 'rev-parse', 'HEAD'], text=True).strip()
+  assert evidence['sourceHead'] == head
+  source = Path(evidence['sourceRlog'])
+  assert source.is_absolute() and source.is_file() and sha(source) == evidence['sourceRlogSha256']
+  start = evidence['sourceStartMonoNs']
+  assert type(start) is int and start > 0 and len(evidence['samples']) == 2
+  values = evidence['values']
+  fields = {'v_cruise_kph', 'v_cruise_cluster_kph', 'v_cruise_kph_last'}
+  assert set(values) == fields
+  assert all(type(value) in (int, float) and math.isfinite(value) and 0 < value <= V_CRUISE_MAX for value in values.values())
+  raw_samples, times = [], []
+  for sample in evidence['samples']:
+    raw = base64.b64decode(sample['rawEventBase64'], validate=True)
+    event = messaging.log_from_bytes(raw)
+    assert event.which() == 'carState' and event.valid
+    cs, stamp = event.carState, int(event.logMonoTime)
+    assert stamp == sample['monoNs'] and 0 < start - stamp <= 30_000_000
+    assert cs.canValid and not cs.canTimeout and cs.cruiseState.available and not cs.cruiseState.enabled
+    assert not len(cs.buttonEvents) and not cs.buttonEnable
+    assert sample['canValid'] is True and sample['canTimeout'] is False
+    assert sample['cruiseAvailable'] is True and sample['cruiseEnabled'] is False and sample['buttonCount'] == 0
+    assert float(cs.vCruise) == sample['vCruiseKph'] == values['v_cruise_kph'] == values['v_cruise_kph_last']
+    assert float(cs.vCruiseCluster) == sample['vCruiseClusterKph'] == values['v_cruise_cluster_kph']
+    raw_samples.append(raw)
+    times.append(stamp)
+  assert 0 < times[1] - times[0] <= 30_000_000
+  # Independently prove these are the last two actual source samples before
+  # this suffix, rather than trusting scalar claims beside an unrelated rlog.
+  if verify_source:
+    from openpilot.tools.lib.logreader import LogReader
+
+    recent = []
+    for event in LogReader(str(source)):
+      if event.which() == 'carState':
+        if int(event.logMonoTime) >= start:
+          break
+        recent = (recent + [event.as_builder().to_bytes()])[-2:]
+    assert recent == raw_samples, 'Cruise memory does not match the last two source carState events'
+  return {
+    'checkpointSha256': args.cruise_checkpoint_sha256,
+    'sourceStartMonoNs': start,
+    'sourceRlog': str(source),
+    'sourceRlogSha256': evidence['sourceRlogSha256'],
+    'cpSha256': evidence['cpSha256'],
+    'carSchemaSha256': evidence['carSchemaSha256'],
+    'sourceHead': head,
+    'sampleMonoNs': times,
+    'rawSampleSha256': [hashlib.sha256(raw).hexdigest() for raw in raw_samples],
+    'values': values,
+  }
+
+
+def cruise_memory(args, repo, helper):
+  from openpilot.selfdrive.car.cruise import V_CRUISE_UNSET
+
+  checked = verify_cruise_evidence(args, repo, verify_source=False)
+  verification = args.cruise_verification.resolve()
+  assert verification.parent == args.receipt.resolve().parent
+  assert sha(verification) == args.cruise_verification_sha256
+  proof = json.loads(verification.read_text())
+  assert proof['phase'] == 'before-owned-actors' and proof['sourceMembershipVerified'] is True
+  assert proof['prefix'] == os.environ['OPENPILOT_PREFIX'] and proof['paramsRoot'] == os.environ['PARAMS_ROOT']
+  assert proof['evidence'] == checked and 0 < proof['verifiedMonoNs'] <= time.monotonic_ns()
+  values = checked['values']
+  fields = set(values)
+  before = {name: getattr(helper, name) for name in fields}
+  assert before == {'v_cruise_kph': V_CRUISE_UNSET, 'v_cruise_cluster_kph': V_CRUISE_UNSET, 'v_cruise_kph_last': 0}
+  untouched = {name: value for name, value in vars(helper).items() if name not in fields}
+  for name, value in values.items():
+    setattr(helper, name, float(value))
+  assert untouched == {name: value for name, value in vars(helper).items() if name not in fields}
+  return {
+    **checked,
+    'priorSetSpeedInitializationExcluded': True,
+    'sourceVerificationSha256': args.cruise_verification_sha256,
+    'restoredFieldsBefore': before,
+    'restoredFieldsAfter': {name: getattr(helper, name) for name in fields},
+    'otherCruiseFieldsUnchanged': True,
+    'restoredAuthority': False,
+  }
+
+
+def main():
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('--repo', type=Path, required=True)
+  parser.add_argument('--car-params', type=Path, required=True)
+  parser.add_argument('--car-params-sha256', required=True)
+  parser.add_argument('--route-car-params', type=Path, required=True)
+  parser.add_argument('--route-car-params-sha256', required=True)
+  parser.add_argument('--car-schema-sha256', required=True)
+  parser.add_argument('--cruise-checkpoint', type=Path, required=True)
+  parser.add_argument('--cruise-checkpoint-sha256', required=True)
+  parser.add_argument('--cruise-verification', type=Path, required=True)
+  parser.add_argument('--cruise-verification-sha256', required=True)
+  parser.add_argument('--receipt', type=Path, required=True)
+  args = parser.parse_args()
+  repo = args.repo.resolve()
+  assert sys.dont_write_bytecode and sys.platform == 'linux'
+  assert os.environ.get('SP_PRIVATE_NATIVE_BENCH') == '1'
+  assert os.environ.get('SP_STEADY_STATE_CHECKPOINT') == '1'
+  assert os.environ.get('OPENPILOT_PREFIX', '').startswith('replay-bench-')
+  assert Path(os.environ['PARAMS_ROOT']).resolve() == args.receipt.resolve().parent / 'params'
+  assert not args.receipt.exists() and not args.receipt.resolve().is_relative_to(repo)
+  assert all(name not in os.environ for name in ('REPLAY', 'SIMULATION', 'CEREAL_FAKE', 'FAKESEND', 'STARTED'))
+  assert sha(args.car_params) == args.car_params_sha256
+  assert sha(args.route_car_params) == args.route_car_params_sha256
+  assert sha(repo / 'opendbc_repo/opendbc/car/car.capnp') == args.car_schema_sha256
+  pinned = args.car_params.read_bytes()
+  assert pinned == args.route_car_params.read_bytes(), 'Live and recorded final CP must be byte-identical under current schema'
+  sys.path[:0] = [str(repo / path) for path in ('', 'opendbc_repo', 'msgq_repo', 'rednose_repo', 'tinygrad_repo')]
+  from opendbc.car.structs import car
+  from opendbc.car.car_helpers import interfaces
+  from openpilot.common.params import Params
+  from openpilot.common.realtime import Priority, config_realtime_process
+  from openpilot.selfdrive.car.card import Car
+  from openpilot.starpilot.longitudinal.ioniq6_start import eligible
+
+  config_realtime_process(4, Priority.CTRL_HIGH)
+  from placement import apply
+
+  apply('card')
+  with car.CarParams.from_bytes(pinned) as source:
+    cp = source.as_builder()
+  assert eligible(cp), 'Only exact current Ioniq6 active-LONG CP checkpoint is supported'
+  interface = interfaces[cp.carFingerprint]
+  selected = Car(CI=interface(cp), RI=interface.RadarInterface(cp))
+  try:
+    assert selected.CP.to_bytes() == pinned, 'Card finalized CP bytes differ from verified live/route checkpoint'
+    params = Params()
+    assert params.get('CarParams') == pinned
+    assert selected.vehicle_startup.owner is None
+    assert not selected.ioniq6_host_warmed and not selected.ioniq6_panda_armed
+    assert not selected.ioniq6_long_lost
+    assert selected.ioniq6_keepalive is None and selected.ioniq6_restore_callbacks is None
+    restored_fields = {'ci_initialized': True, 'ioniq6_long_selected': True, 'ioniq6_long_prearmed': True, 'ioniq6_long_pending': False}
+    before = {name: getattr(selected, name) for name in restored_fields}
+    # Explicit root-authorized phase checkpoint only. No accepted command,
+    # driver latch, parser sample, native status, proof, clock or session is copied.
+    for name, value in restored_fields.items():
+      setattr(selected, name, value)
+    ready_before = {name: params.get_bool(name) for name in ('FirmwareQueryDone', 'ControlsReady')}
+    for name in ready_before:
+      params.put_bool(name, True, block=True)
+    memory = cruise_memory(args, repo, selected.v_cruise_helper)
+    receipt = {
+      'kind': 'private-already-running-card-checkpoint',
+      'startupExcluded': True,
+      'excluded': [
+        'vehicle discovery',
+        'ECU handoff/ownership transaction',
+        'CI.init',
+        'physical native firmware startup',
+        'prior set-speed initialization',
+        'road qualification',
+      ],
+      'restoredFieldsBefore': before,
+      'restoredFieldsAfter': restored_fields,
+      'restoredPrivateParamsBefore': ready_before,
+      'restoredPrivateParamsAfter': {'FirmwareQueryDone': True, 'ControlsReady': True},
+      'cruiseMemory': memory,
+      'freshGatesRetained': {
+        'ioniq6_host_warmed': selected.ioniq6_host_warmed,
+        'ioniq6_panda_armed': selected.ioniq6_panda_armed,
+        'ioniq6_long_lost': selected.ioniq6_long_lost,
+      },
+      'noRestoredAuthority': [
+        'intent latch',
+        'native grant',
+        'panda inventory',
+        'RX health',
+        'session',
+        'lease',
+        'source timestamp',
+        'axis proof',
+        'active CarControl',
+      ],
+      'cpSha256': sha(args.car_params),
+      'routeCpSha256': sha(args.route_car_params),
+      'carSchemaSha256': args.car_schema_sha256,
+      'cardSourceSha256': sha(repo / 'openpilot/selfdrive/car/card.py'),
+      'pid': os.getpid(),
+      'startedMonoNs': time.monotonic_ns(),
+      'paramsRoot': str(Path(os.environ['PARAMS_ROOT']).resolve()),
+      'prefix': os.environ['OPENPILOT_PREFIX'],
+      'normalScheduler': {'policy': os.sched_getscheduler(0), 'priority': os.sched_getparam(0).sched_priority, 'affinity': sorted(os.sched_getaffinity(0))},
+    }
+    temporary_receipt = args.receipt.with_suffix('.tmp')
+    assert not temporary_receipt.exists()
+    temporary_receipt.write_text(json.dumps(receipt, indent=2) + '\n')
+    temporary_receipt.replace(args.receipt)
+    from phase_timing import PhaseCapture
+
+    capture = PhaseCapture(args.receipt.parent / 'card-timing.json', 16000, 'private-current-card-phases')
+    for name in ('step', 'state_update', 'controls_update', 'state_publish'):
+      capture.add(Car, name, 'Car.' + name, cycle=name == 'step')
+    if os.environ.get('SP_PRIVATE_CARD_MACRO') in ('1', '2'):
+      from opendbc.can.parser import CANParser
+      from opendbc.car import steering_projection
+      from openpilot.starpilot.flm import feedback_observation, physical_input_observation
+
+      capture.add(CANParser, 'update', 'CANParser.update')
+      capture.add(feedback_observation.CardFeedbackCapture, 'publication', 'Feedback.publication')
+      capture.add(feedback_observation, 'applied_digest', 'Feedback.applied_digest')
+      capture.add(physical_input_observation, 'body_digest', 'Physical.body_digest')
+      capture.add(steering_projection, 'projection_to_dict', 'Projection.to_dict')
+      if os.environ.get('SP_PRIVATE_CARD_MACRO') == '2':
+        from openpilot.selfdrive.car import card as card_module
+        from openpilot.starpilot.aol.transport_pause import TransportPauseFeedback
+
+        capture.add(type(selected.CI), 'update', 'CI.update')
+        capture.add(card_module, 'can_capnp_to_list', 'CAN.deserialize')
+        capture.add(TransportPauseFeedback, 'observe', 'Transport.observe')
+        for name in (
+          'radar_update',
+          '_feedback_capture',
+          'observe_ioniq6_long_authority',
+          'update_vehicle_state_context',
+          'refresh_slc_configuration',
+          'update_cruise_speed',
+        ):
+          capture.add(Car, name, 'Car.' + name)
+    with capture:
+      if os.environ.get('SP_PRIVATE_CARD_PROFILE') == '1':
+        import cProfile
+
+        profile = cProfile.Profile()
+        profile.enable()
+        try:
+          selected.card_thread()
+        finally:
+          profile.disable()
+          profile.dump_stats(str(args.receipt.parent / 'card.pstats'))
+      else:
+        selected.card_thread()
+  finally:
+    selected.vehicle_startup.close()
+    # There is no ECU/keepalive owner to restore in this virtual checkpoint.
+    assert selected.ioniq6_keepalive is None and selected.ioniq6_restore_callbacks is None
+
+
+if __name__ == '__main__':
+  main()

tools/heatdeath/heatdeath.py

@@ -0,0 +1,528 @@
+#!/usr/bin/env python3
+"""Reproducible joined Controls replay benchmark; run --help for commands."""
+
+import argparse
+import base64
+from contextlib import contextmanager
+import fcntl
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import platform
+import shutil
+import signal
+import subprocess
+import sys
+import sysconfig
+import time
+import uuid
+
+if __package__:
+  from .placement import load_plan, parked
+else:
+  from placement import load_plan, parked
+
+HERE = Path(__file__).resolve().parent
+REPO = HERE.parents[1]
+SOURCES = (
+  'openpilot/selfdrive/car/card.py',
+  'openpilot/selfdrive/selfdrived/selfdrived.py',
+  'openpilot/selfdrive/controls/controlsd.py',
+  'openpilot/selfdrive/controls/plannerd.py',
+  'openpilot/selfdrive/controls/radard.py',
+  'openpilot/starpilot/flm/steerlearnerd.py',
+  'openpilot/common/realtime.py',
+  'opendbc_repo/opendbc/car/car.capnp',
+  'openpilot/cereal/log.capnp',
+  'openpilot/cereal/custom.capnp',
+)
+
+
+def sha(path):
+  return hashlib.sha256(Path(path).read_bytes()).hexdigest()
+
+
+def write(path, value):
+  with Path(path).open('x') as stream:
+    json.dump(value, stream, indent=2)
+    stream.write('\n')
+
+
+def git(repo, *args):
+  return subprocess.check_output(
+    ['git', '--no-optional-locks', '-c', 'gc.auto=0', '-c', 'maintenance.auto=false', '-C', str(repo), *args], text=True, timeout=30
+  ).strip()
+
+
+def clean(repo):
+  if git(repo, 'status', '--porcelain'):
+    raise RuntimeError('Use a clean, built checkout; benchmark outputs must be outside it')
+  return git(repo, 'rev-parse', 'HEAD')
+
+
+def external(path, repo):
+  path, repo = path.resolve(), repo.resolve()
+  if path.is_relative_to(repo) or path.exists():
+    raise ValueError('Output must be a new path outside the source checkout')
+  return path
+
+
+def prepare(args):
+  repo, output = args.repo.resolve(), external(args.output, args.repo.resolve())
+  head = clean(repo)
+  for part in ('', 'opendbc_repo', 'msgq_repo', 'rednose_repo', 'tinygrad_repo'):
+    sys.path.insert(0, str(repo / part))
+  from openpilot.tools.lib.logreader import LogReader
+  from opendbc.car.structs import car
+
+  cp, cp_time, prior, source = None, -1, [], None
+  for path in args.rlog:
+    for msg in LogReader(str(path.resolve())):
+      if msg.logMonoTime >= args.start_mono_ns:
+        continue
+      if msg.which() == 'carParams' and int(msg.logMonoTime) > cp_time:
+        cp, cp_time = msg.carParams.as_builder().to_bytes(), int(msg.logMonoTime)
+      if msg.which() == 'carState' and args.start_mono_ns - 1_000_000_000 <= msg.logMonoTime:
+        cs = msg.carState
+        prior.append(
+          {
+            'monoNs': int(msg.logMonoTime),
+            'vCruiseKph': float(cs.vCruise),
+            'vCruiseClusterKph': float(cs.vCruiseCluster),
+            'canValid': bool(cs.canValid),
+            'canTimeout': bool(cs.canTimeout),
+            'cruiseEnabled': bool(cs.cruiseState.enabled),
+            'cruiseAvailable': bool(cs.cruiseState.available),
+            'buttonCount': len(cs.buttonEvents),
+            'rawEventBase64': base64.b64encode(msg.as_builder().to_bytes()).decode(),
+            'source': str(path.resolve()),
+          }
+        )
+  if cp is None:
+    raise ValueError('Include the route startup rlog containing its original CarParams')
+  with car.CarParams.from_bytes(cp) as parsed:
+    if parsed.carFingerprint != 'HYUNDAI_IONIQ_6' or not parsed.openpilotLongitudinalControl:
+      raise ValueError('This exported steady-state fixture supports active-long Ioniq 6 recordings only')
+  prior.sort(key=lambda row: row['monoNs'])
+  last = prior[-2:]
+  if (
+    len(last) != 2
+    or last[0]['source'] != last[1]['source']
+    or last[0]['monoNs'] >= last[1]['monoNs']
+    or any(not row['canValid'] or row['canTimeout'] or row['buttonCount'] or not 0 < row['vCruiseKph'] < 150 for row in last)
+    or any(last[0][key] != last[1][key] for key in ('vCruiseKph', 'vCruiseClusterKph'))
+  ):
+    raise ValueError('Need two clean stable cruise-speed samples in the second before the window')
+  source = Path(last[0]['source'])
+  for row in last:
+    del row['source']
+  output.mkdir(parents=True, mode=0o700)
+  packet = output / 'packet'
+  packet.mkdir()
+  for path in HERE.iterdir():
+    if path.suffix in ('.py', '.cc'):
+      shutil.copyfile(path, packet / path.name)
+  pins = {'sourceHead': head, 'sources': {name: sha(repo / name) for name in SOURCES}, 'helpers': {p.name: sha(p) for p in packet.iterdir()}}
+  write(packet / 'joined-source-pins.json', pins)
+  (output / 'carparams.bin').write_bytes(cp)
+  shutil.copyfile(args.params_json, output / 'params.json')
+  write(
+    output / 'cruise.json',
+    {
+      'kind': 'private-recorded-cruise-speed-checkpoint',
+      'startupExcluded': True,
+      'sourceStartMonoNs': args.start_mono_ns,
+      'sourceRlog': str(source),
+      'sourceRlogSha256': sha(source),
+      'cpSha256': sha(output / 'carparams.bin'),
+      'samples': last,
+      'sourceHead': head,
+      'carSchemaSha256': pins['sources']['opendbc_repo/opendbc/car/car.capnp'],
+      'values': {'v_cruise_kph': last[-1]['vCruiseKph'], 'v_cruise_cluster_kph': last[-1]['vCruiseClusterKph'], 'v_cruise_kph_last': last[0]['vCruiseKph']},
+    },
+  )
+  subprocess.run(
+    [
+      sys.executable,
+      '-B',
+      str(packet / 'driver_aa_v47_fixture.py'),
+      'prepare',
+      '--repo',
+      str(repo),
+      *sum((['--rlog', str(path.resolve())] for path in args.rlog), []),
+      '--start-mono-ns',
+      str(args.start_mono_ns),
+      '--duration',
+      str(args.duration),
+      '--source-boot-minus-mono-ns',
+      str(args.source_boot_minus_mono_ns),
+      '--device-motion-clock',
+      args.device_motion_clock,
+      '--output',
+      str(output / 'inputs'),
+    ],
+    check=True,
+  )
+  if clean(repo) != head:
+    raise RuntimeError('Source changed during preparation')
+  write(
+    output / 'bundle.json', {'sourceHead': head, 'repo': str(repo), 'files': {str(p.relative_to(output)): sha(p) for p in output.rglob('*') if p.is_file()}}
+  )
+  print(f'Prepared {output}; next run ./heatdeath build --bundle {output}')
+
+
+def verify(bundle):
+  bundle = bundle.resolve()
+  info = json.loads((bundle / 'bundle.json').read_text())
+  repo = Path(info['repo']).resolve()
+  if clean(repo) != info['sourceHead']:
+    raise RuntimeError('Bundle is pinned to another source revision; prepare it again for this checkout')
+  for name, expected in info['files'].items():
+    path = (bundle / name).resolve()
+    if not path.is_relative_to(bundle) or sha(path) != expected:
+      raise RuntimeError('Bundle changed: ' + name)
+  pins = json.loads((bundle / 'packet/joined-source-pins.json').read_text())
+  if any(sha(repo / name) != wanted for name, wanted in pins['sources'].items()):
+    raise RuntimeError('Pinned production source changed')
+  return bundle, repo, info
+
+
+def require_device():
+  if sys.platform != 'linux' or platform.machine() not in ('aarch64', 'arm64') or os.geteuid() != 0:
+    raise RuntimeError('Execution/build require a root Linux ARM64 comma bench environment')
+  parked()
+
+
+@contextmanager
+def device_lease():
+  path = Path('/data/starbench/heatdeath.lock')
+  path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+  with path.open('a') as stream:
+    fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    parked()
+    yield
+
+
+def build(args):
+  require_device()
+  bundle, repo, info = verify(args.bundle)
+  out = external(bundle / 'native', repo)
+  out.mkdir(mode=0o700)
+  site = Path(sysconfig.get_path('purelib'))
+  base = ['clang++', '-std=c++17', '-O2', '-g', '-fPIC', '-pthread', '-D__COMMA_HARDWARE__=1']
+  for name in ('', 'openpilot', 'msgq_repo', 'opendbc_repo', 'openpilot/cereal/gen/cpp'):
+    base += ['-I', str(repo / name)]
+  for name in ('capnproto', 'json11', 'zeromq'):
+    base += ['-I', str(site / name / 'install/include')]
+  sources = [bundle / 'packet/native_endpoint.cc', repo / 'openpilot/selfdrive/pandad/panda.cc', repo / 'openpilot/selfdrive/pandad/panda_safety.cc']
+  libs = [repo / name for name in ('openpilot/common/libcommon.a', 'openpilot/cereal/libsocketmaster.a', 'msgq_repo/libmsgq.a', 'openpilot/cereal/libcereal.a')]
+  libs += [
+    site / name / 'install/lib' / lib
+    for name, lib in (('capnproto', 'libcapnp.a'), ('capnproto', 'libkj.a'), ('json11', 'libjson11.a'), ('zeromq', 'libzmq.a'))
+  ]
+  inputs = {str(p): sha(p) for p in [*sources, *libs, repo / 'opendbc_repo/opendbc/safety/tests/libsafety/safety.c']}
+  commands = []
+  for index, source in enumerate(sources):
+    commands.append([*base, '-c', str(source), '-o', str(out / f'endpoint-{index}.o')])
+  commands.append(
+    [
+      'clang++',
+      *(str(out / f'endpoint-{index}.o') for index in range(3)),
+      '-Wl,--start-group',
+      *(str(p) for p in libs),
+      '-Wl,--end-group',
+      '-ldl',
+      '-lrt',
+      '-pthread',
+      '-Wl,--no-undefined',
+      '-o',
+      str(out / 'native_endpoint'),
+    ]
+  )
+  commands.append(
+    [
+      'cc',
+      '-fPIC',
+      '-Wall',
+      '-Wextra',
+      '-Werror',
+      '-nostdlib',
+      '-fno-builtin',
+      '-std=gnu11',
+      '-Wfatal-errors',
+      '-Wno-pointer-to-int-cast',
+      '-g',
+      '-O0',
+      '-fno-omit-frame-pointer',
+      '-fsanitize=undefined',
+      '-fno-sanitize-recover=undefined',
+      '-I',
+      str(repo / 'opendbc_repo'),
+      '-c',
+      str(repo / 'opendbc_repo/opendbc/safety/tests/libsafety/safety.c'),
+      '-o',
+      str(out / 'libsafety.o'),
+    ]
+  )
+  commands.append(['cc', '-shared', str(out / 'libsafety.o'), '-o', str(out / 'libsafety.so'), '-fsanitize=undefined', '-fno-sanitize-recover=undefined'])
+  for index, command in enumerate(commands):
+    parked()
+    with (out / f'build-{index}.log').open('x') as stream:
+      subprocess.run(command, stdout=stream, stderr=subprocess.STDOUT, check=True, timeout=90)
+  verify(bundle)
+  if any(sha(name) != expected for name, expected in inputs.items()):
+    raise RuntimeError('Native build inputs changed')
+  write(
+    out / 'receipt.json',
+    {
+      'sourceHead': info['sourceHead'],
+      'commands': commands,
+      'inputs': inputs,
+      'outputs': {name: sha(out / name) for name in ('native_endpoint', 'libsafety.so')},
+    },
+  )
+  print(f'Built virtual endpoint in {out}; no USB/SPI implementation linked')
+
+
+def run(args):
+  require_device()
+  bundle, repo, info = verify(args.bundle)
+  plan = load_plan(args.placement)
+  available = os.sched_getaffinity(0)
+  if any(not set(cpus) <= available for cpus in plan.values()):
+    raise ValueError('Requested CPU is unavailable in this environment')
+  out = external(args.output, repo)
+  if out.is_relative_to(bundle):
+    raise ValueError('Keep run outputs outside the immutable bundle')
+  receipt = json.loads((bundle / 'native/receipt.json').read_text())
+  if receipt['sourceHead'] != info['sourceHead']:
+    raise ValueError('Native build belongs to another source revision')
+  for name, expected in receipt['outputs'].items():
+    if sha(bundle / 'native' / name) != expected:
+      raise ValueError('Native artifact changed')
+  packet = bundle / 'packet'
+  cp = bundle / 'carparams.bin'
+  command = [
+    sys.executable,
+    '-B',
+    str(packet / 'driver_aa_v47_fixture.py'),
+    'run',
+    '--repo',
+    str(repo),
+    '--inputs',
+    str(bundle / 'inputs'),
+    '--car-params',
+    str(cp),
+    '--car-params-sha256',
+    sha(cp),
+    '--route-car-params',
+    str(cp),
+    '--route-car-params-sha256',
+    sha(cp),
+    '--car-schema-sha256',
+    sha(repo / 'opendbc_repo/opendbc/car/car.capnp'),
+    '--params-json',
+    str(bundle / 'params.json'),
+    '--native-endpoint',
+    str(bundle / 'native/native_endpoint'),
+    '--native-endpoint-sha256',
+    receipt['outputs']['native_endpoint'],
+    '--native-library',
+    str(bundle / 'native/libsafety.so'),
+    '--native-library-sha256',
+    receipt['outputs']['libsafety.so'],
+    '--output',
+    str(out),
+    '--helpers-dir',
+    str(packet),
+    '--root-grant',
+    'execute-private-native-ipc-bench',
+    '--steady-state-checkpoint',
+    '--cruise-checkpoint',
+    str(bundle / 'cruise.json'),
+    '--cruise-checkpoint-sha256',
+    sha(bundle / 'cruise.json'),
+  ]
+  if args.native_core2:
+    command.append('--private-native-core2')
+  forbidden = ('REPLAY', 'SIMULATION', 'CEREAL_FAKE', 'FAKESEND', 'STARTED')
+  if any(key in os.environ for key in forbidden) or any(key.startswith(('SP_PRIVATE_', 'SP_PERF_')) for key in os.environ):
+    raise ValueError('Run from a clean environment, outside ./dev; private candidate hooks are not accepted')
+  env = dict(
+    os.environ, PYTHONDONTWRITEBYTECODE='1', HEATDEATH_PLACEMENT=json.dumps(plan), SP_PRIVATE_LOOP_MACRO='1', SP_PRIVATE_IRQ_LEASE_TOKEN=uuid.uuid4().hex
+  )
+  reserve = None
+  if args.reserve_core is not None:
+    if args.reserve_core not in available or not 0 < args.reserve_ms < args.reserve_period_ms <= 1000:
+      raise ValueError('Reserve needs an available core and 0 < dose < period <= 1000 ms')
+    reserve = {'core': args.reserve_core, 'cpuMs': args.reserve_ms, 'periodMs': args.reserve_period_ms}
+  env['HEATDEATH_RESERVE'] = json.dumps(reserve)
+  context_before = context()
+  old_handlers = {}
+  child = None
+
+  def stop(_signum, _frame):
+    raise KeyboardInterrupt
+
+  try:
+    for number in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
+      old_handlers[number] = signal.signal(number, stop)
+    child = subprocess.Popen(command, env=env, start_new_session=True)
+    deadline = time.monotonic() + 180
+    while child.poll() is None:
+      parked()
+      if time.monotonic() >= deadline:
+        raise TimeoutError('Benchmark exceeded its 180-second wall bound')
+      time.sleep(0.2)
+    if child.returncode:
+      raise RuntimeError(f'Benchmark failed ({child.returncode}); inspect {out}')
+  finally:
+    for number in old_handlers:
+      signal.signal(number, signal.SIG_IGN)
+    try:
+      try:
+        if child is not None and child.poll() is None:
+          child.terminate()
+          try:
+            child.wait(timeout=15)
+          except subprocess.TimeoutExpired:
+            child.kill()
+            child.wait(timeout=5)
+      finally:
+        reap_token(env['SP_PRIVATE_IRQ_LEASE_TOKEN'])
+        journal = out / 'shm-ownership.json'
+        if journal.exists():
+          cleanup = importlib.util.spec_from_file_location('heatdeath_cleanup', packet / 'owned_msgq_cleanup_v2.py')
+          module = importlib.util.module_from_spec(cleanup)
+          cleanup.loader.exec_module(module)
+          previous = os.environ.get('SP_PRIVATE_IRQ_LEASE_TOKEN')
+          os.environ['SP_PRIVATE_IRQ_LEASE_TOKEN'] = env['SP_PRIVATE_IRQ_LEASE_TOKEN']
+          try:
+            module.cleanup(journal, out / 'cleanup.json')
+          finally:
+            if previous is None:
+              os.environ.pop('SP_PRIVATE_IRQ_LEASE_TOKEN', None)
+            else:
+              os.environ['SP_PRIVATE_IRQ_LEASE_TOKEN'] = previous
+        verify(bundle)
+        if out.is_dir():
+          write(out / 'context.json', {'before': context_before, 'after': context()})
+    finally:
+      for number, handler in old_handlers.items():
+        signal.signal(number, handler)
+  print(json.dumps(json.loads((out / 'result.json').read_text()), indent=2))
+
+
+def context():
+  result = {
+    'monotonicNs': time.monotonic_ns(),
+    'machine': platform.machine(),
+    'bootId': Path('/proc/sys/kernel/random/boot_id').read_text().strip(),
+    'files': {},
+  }
+  paths = [Path('/sys/devices/system/cpu/online')]
+  for pattern in ('/sys/devices/system/cpu/cpufreq/policy*/scaling_*', '/sys/class/thermal/thermal_zone*/temp'):
+    import glob
+
+    paths.extend(Path(path) for path in glob.glob(pattern))
+  for path in paths:
+    try:
+      result['files'][str(path)] = path.read_text().strip()
+    except OSError as error:
+      result['files'][str(path)] = type(error).__name__
+  return result
+
+
+def token_processes(token):
+  key = ('SP_PRIVATE_IRQ_LEASE_TOKEN=' + token).encode()
+  owned = []
+  for path in Path('/proc').iterdir():
+    if not path.name.isdecimal() or int(path.name) == os.getpid():
+      continue
+    try:
+      if key in (path / 'environ').read_bytes().split(b'\0'):
+        owned.append(int(path.name))
+    except (FileNotFoundError, ProcessLookupError):
+      pass
+  return owned
+
+
+def reap_token(token):
+  for number, seconds in ((signal.SIGTERM, 3), (signal.SIGKILL, 3)):
+    for pid in token_processes(token):
+      try:
+        descriptor = os.pidfd_open(pid)
+      except ProcessLookupError:
+        continue
+      try:
+        # Bind the signal to this process identity, even if its PID is reused.
+        if pid in token_processes(token):
+          signal.pidfd_send_signal(descriptor, number)
+      except ProcessLookupError:
+        pass
+      finally:
+        os.close(descriptor)
+    deadline = time.monotonic() + seconds
+    while token_processes(token) and time.monotonic() < deadline:
+      time.sleep(0.05)
+  if token_processes(token):
+    raise RuntimeError('Owned benchmark actors remain; namespace retained for inspection')
+
+
+def report(args):
+  if __package__:
+    from .report import summarize
+  else:
+    from report import summarize
+
+  result = summarize(args.output, args.skip_seconds)
+  print(json.dumps(result, indent=2))
+
+
+def main():
+  parser = argparse.ArgumentParser(description=__doc__)
+  commands = parser.add_subparsers(dest='command', required=True)
+  prep = commands.add_parser('prepare', help='Extract a pinned recorded window and helper bundle')
+  prep.add_argument('--repo', type=Path, default=REPO)
+  prep.add_argument('--rlog', type=Path, action='append', required=True)
+  prep.add_argument('--start-mono-ns', type=int, required=True)
+  prep.add_argument('--duration', type=float, default=60)
+  prep.add_argument('--source-boot-minus-mono-ns', type=int, required=True)
+  prep.add_argument('--device-motion-clock', choices=('monotonic', 'boottime'), required=True)
+  prep.add_argument('--params-json', type=Path, required=True, help='Explicit base64-valued private benchmark preference snapshot')
+  prep.add_argument('--output', type=Path, required=True)
+  compile_parser = commands.add_parser('build', help='Build the virtual native endpoint using existing compiled dependencies')
+  compile_parser.add_argument('--bundle', type=Path, required=True)
+  execute = commands.add_parser('run', help='Run an offroad joined replay; no physical CAN I/O')
+  execute.add_argument('--bundle', type=Path, required=True)
+  execute.add_argument('--output', type=Path, required=True)
+  execute.add_argument('--placement', default='{}', help='JSON actor-to-CPU-list map, e.g. controlsd:[6], card:[4]')
+  execute.add_argument('--native-core2', action='store_true', help='Move only the virtual endpoint to core 2 after native startup')
+  execute.add_argument('--reserve-core', type=int, help='Add an explicitly synthetic CFS reserve on this CPU')
+  execute.add_argument('--reserve-ms', type=float, default=8)
+  execute.add_argument('--reserve-period-ms', type=float, default=50)
+  results = commands.add_parser('report', help='Summarize captured timing, native evidence and limitations')
+  results.add_argument('--output', type=Path, required=True)
+  results.add_argument('--skip-seconds', type=float, default=10)
+  if len(sys.argv) == 1:
+    parser.print_help()
+    return
+  args = parser.parse_args()
+  if args.command == 'prepare' and not 1 <= args.duration <= 120:
+    parser.error('Duration must be between 1 and 120 seconds')
+  try:
+    if args.command in ('build', 'run'):
+      require_device()
+      with device_lease():
+        {'build': build, 'run': run}[args.command](args)
+    else:
+      {'prepare': prepare, 'report': report}[args.command](args)
+  except (OSError, ValueError, RuntimeError, subprocess.SubprocessError) as error:
+    parser.exit(1, f'heatdeath: {error}\n')
+
+
+if __name__ == '__main__':
+  sys.dont_write_bytecode = True
+  main()

tools/heatdeath/native_endpoint.cc

@@ -0,0 +1,314 @@
+// Private bench transport. Compile current production pandad into this TU;
+// never link panda_usb.cc, spi.cc, libpanda.a, or libusb.
+#include <dlfcn.h>
+#include <sys/prctl.h>
+#include <chrono>
+#include <filesystem>
+#include <fstream>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <thread>
+#include "common/tests/native_test.h"
+#include "common/hardware/hw.h"
+
+// Virtualize the remaining physical peripheral write. Inherit real hardware
+// identity/PC() so the production main retains its real FIFO/core setup.
+#if __COMMA_HARDWARE__
+class BenchHardware : public HardwareComma {
+public:
+  static void set_ir_power(int) {}
+};
+#else
+class BenchHardware : public HardwarePC {
+public:
+  static void set_ir_power(int) {}
+};
+#endif
+#undef Hardware
+#define Hardware BenchHardware
+#include "selfdrive/pandad/pandad.cc"
+
+namespace {
+constexpr const char *SERIAL = "bench-virtual-owner";
+struct Frame { uint32_t address; uint8_t bus; std::vector<uint8_t> data; };
+struct TransportSource { uint8_t bus; bool returned; bool rejected; };
+// Match Panda::unpack_can_buffer exactly. CP1 physical buses are 0..3;
+// historical TX return/rejection markers are separate transport categories.
+static_assert(CAN_RETURNED_BUS_OFFSET == 0x80U && CAN_REJECTED_BUS_OFFSET == 0xC0U);
+TransportSource transport_source(uint8_t source) {
+  if (source < 4) return {source, false, false};
+  if (source >= CAN_RETURNED_BUS_OFFSET && source < CAN_RETURNED_BUS_OFFSET + 4)
+    return {static_cast<uint8_t>(source - CAN_RETURNED_BUS_OFFSET), true, false};
+  if (source >= CAN_REJECTED_BUS_OFFSET && source < CAN_REJECTED_BUS_OFFSET + 4)
+    return {static_cast<uint8_t>(source - CAN_REJECTED_BUS_OFFSET), false, true};
+  CHECK(false); return {}; // Unknown source / additional Panda bus is forbidden.
+}
+struct Batch { uint64_t offset; std::vector<Frame> frames; };
+std::vector<Batch> trace;
+std::string library_path, release_path, receipt_path;
+uint64_t duration_ns = 0, start_ns = 0;
+size_t cursor = 0;
+std::mutex native_lock;
+uint64_t rx_count = 0, rx_rejected = 0, tx_count = 0, tx_accepted = 0, active_tx_accepted = 0;
+uint64_t max_feed_late_ns = 0;
+uint64_t trace_rx = 0, trace_returned = 0, trace_rejected = 0;
+uint64_t playback_returned = 0, playback_rejected = 0;
+struct AcceptedTxSample {
+  uint64_t index, mono_ns;
+  uint32_t address;
+  uint8_t bus, permission;
+  bool controls;
+  std::string payload_hex;
+};
+std::vector<AcceptedTxSample> granted_tx_samples, actuator_hint_tx_samples;
+AcceptedTxSample tx_sample(const Frame &frame, uint8_t permission, bool controls) {
+  static constexpr char HEX[] = "0123456789abcdef";
+  std::string payload; payload.reserve(frame.data.size() * 2);
+  for (const auto byte : frame.data) { payload.push_back(HEX[byte >> 4]); payload.push_back(HEX[byte & 15]); }
+  return {tx_count, aol_monotonic_ns(), frame.address, frame.bus, permission, controls, std::move(payload)};
+}
+void write_tx_samples(std::ostream &out, const std::vector<AcceptedTxSample> &samples) {
+  out << '[';
+  for (size_t i = 0; i < samples.size(); ++i) {
+    const auto &sample = samples[i]; if (i) out << ',';
+    out << "{\"nativeAccepted\":true,\"txIndex\":" << sample.index << ",\"monoNs\":" << sample.mono_ns
+        << ",\"address\":" << sample.address << ",\"bus\":" << (unsigned)sample.bus
+        << ",\"permissionMask\":" << (unsigned)sample.permission
+        << ",\"controlsAllowed\":" << (sample.controls ? "true" : "false")
+        << ",\"payloadHex\":\"" << sample.payload_hex << "\"}";
+  }
+  out << ']';
+}
+
+template <typename T> T read(std::ifstream &in) {
+  T value{}; in.read(reinterpret_cast<char *>(&value), sizeof(value)); CHECK(in.good()); return value;
+}
+void load_trace(const char *path) {
+  std::ifstream in(path, std::ios::binary); CHECK(in.good());
+  CHECK(read<uint64_t>(in) == 0x3143415254425053ULL); // SPBTRAC1, little endian
+  duration_ns = read<uint64_t>(in);
+  CHECK(duration_ns >= 1000000000ULL && duration_ns <= 120000000000ULL);
+  const auto count = read<uint32_t>(in); CHECK(count > 0 && count <= 24000);
+  uint64_t previous = 0;
+  for (uint32_t i = 0; i < count; ++i) {
+    Batch batch; batch.offset = read<uint64_t>(in);
+    CHECK(batch.offset >= previous && batch.offset < duration_ns); previous = batch.offset;
+    const auto frames = read<uint32_t>(in); CHECK(frames <= 1024);
+    for (uint32_t n = 0; n < frames; ++n) {
+      Frame frame; frame.address = read<uint32_t>(in); frame.bus = read<uint8_t>(in);
+      const auto length = read<uint8_t>(in); CHECK(length <= 64);
+      const auto source = transport_source(frame.bus);
+      trace_rx += !source.returned && !source.rejected;
+      trace_returned += source.returned; trace_rejected += source.rejected;
+      frame.data.resize(length); in.read(reinterpret_cast<char *>(frame.data.data()), length); CHECK(in.good());
+      batch.frames.push_back(std::move(frame));
+    }
+    size_t bytes = 0; for (const auto &frame : batch.frames) bytes += sizeof(can_header) + frame.data.size();
+    CHECK(bytes <= RECV_SIZE);
+    trace.push_back(std::move(batch));
+  }
+  CHECK(in.peek() == std::ifstream::traits_type::eof());
+}
+CANPacket_t packet(const Frame &frame) {
+  const auto source = transport_source(frame.bus);
+  CANPacket_t result{}; result.addr = frame.address; result.bus = source.bus;
+  result.returned = source.returned; result.rejected = source.rejected;
+  result.fd = frame.data.size() > 8; result.extended = frame.address >= 0x800;
+  bool found = false;
+  for (uint8_t i = 0; i < std::size(dlc_to_len); ++i) if (dlc_to_len[i] == frame.data.size()) {
+    result.data_len_code = i; found = true; break;
+  }
+  CHECK(found); memcpy(result.data, frame.data.data(), frame.data.size()); return result;
+}
+struct Native {
+  void *library;
+  int (*set_hooks)(uint16_t, uint16_t); void (*init)(); void (*timer)(uint32_t); void (*tick)();
+  void (*set_ae)(int); void (*heartbeat)(bool); void (*request)(uint8_t);
+  bool (*rx)(CANPacket_t *); bool (*tx)(CANPacket_t *); bool (*controls)(); bool (*relay)();
+  int (*mode)(); int (*param)(); uint8_t (*permission)(); uint8_t (*mask)();
+  int *ae; bool *rx_invalid; bool *optional; void **policy;
+  uint64_t last_tick = 0;
+  template <typename T> T symbol(const char *name) {
+    auto ptr = dlsym(library, name); CHECK(ptr != nullptr); return reinterpret_cast<T>(ptr);
+  }
+  Native() {
+    library = dlopen(library_path.c_str(), RTLD_NOW | RTLD_LOCAL); CHECK(library != nullptr);
+    set_hooks = symbol<decltype(set_hooks)>("set_safety_hooks"); init = symbol<decltype(init)>("init_tests");
+    timer = symbol<decltype(timer)>("set_timer"); tick = symbol<decltype(tick)>("safety_tick");
+    set_ae = symbol<decltype(set_ae)>("set_alternative_experience");
+    heartbeat = symbol<decltype(heartbeat)>("set_aol_test_heartbeat"); request = symbol<decltype(request)>("aol_set_host_request");
+    rx = symbol<decltype(rx)>("safety_rx_hook"); tx = symbol<decltype(tx)>("safety_tx_hook");
+    controls = symbol<decltype(controls)>("get_controls_allowed"); relay = symbol<decltype(relay)>("get_relay_malfunction");
+    mode = symbol<decltype(mode)>("get_current_safety_mode"); param = symbol<decltype(param)>("get_current_safety_param");
+    permission = symbol<decltype(permission)>("aol_get_permission_mask"); mask = symbol<decltype(mask)>("aol_get_request_mask");
+    ae = symbol<int *>("alternative_experience"); rx_invalid = symbol<bool *>("safety_rx_checks_invalid");
+    optional = symbol<bool *>("aol_optional_release_capable"); policy = symbol<void **>("aol_policy");
+    set_ae(0); CHECK(set_hooks((uint16_t)cereal::CarParams::SafetyModel::NO_OUTPUT, 0) == 0); init();
+  }
+  void clock() {
+    const auto now = aol_monotonic_ns();
+    timer(static_cast<uint32_t>(now / 1000ULL)); // Actual clock; native timer wrap is normal.
+    if (last_tick && now - last_tick >= 1000000000ULL) tick();
+    if (!last_tick || now - last_tick >= 1000000000ULL) last_tick = now;
+  }
+  aol_safety_health_t status() {
+    return {AOL_SAFETY_PROTOCOL_MAGIC, AOL_SAFETY_PROTOCOL_VERSION, mask(), permission(),
+            static_cast<uint8_t>(mode()), static_cast<uint16_t>(param()),
+            static_cast<uint8_t>(*policy ? (1U | (*optional ? 2U : 0U)) : 0U)};
+  }
+  ~Native() { set_ae(0); set_hooks((uint16_t)cereal::CarParams::SafetyModel::NO_OUTPUT, 0); init(); dlclose(library); }
+};
+std::unique_ptr<Native> native;
+
+int write_control(uint8_t command, uint16_t p1, uint16_t p2) {
+  std::lock_guard lock(native_lock); native->clock();
+  if (command == 0xdc) {
+    const int ae = *native->ae;
+    CHECK(native->set_hooks(p1, p2) == 0); native->init(); native->set_ae(ae); native->clock();
+  } else if (command == 0xdf) native->set_ae(p1);
+  else if (command == 0xf3) native->heartbeat(p1 != 0);
+  else if (command == 0xf5) native->request(p1);
+  // Other setup/peripheral requests terminate here, without physical I/O.
+  return 0;
+}
+int read_control(uint8_t command, unsigned char *data, uint16_t length) {
+  std::lock_guard lock(native_lock); native->clock();
+  if (command == 0xc1 && length == 1) { data[0] = (uint8_t)cereal::PandaState::PandaType::RED_PANDA; return 1; }
+  if ((command == 0xd3 || command == 0xd4) && length == 64) {
+    const auto fw = util::read_file("../../../panda/board/obj/panda_h7.bin.signed"); CHECK(fw.size() >= 128);
+    memcpy(data, fw.data() + fw.size() - 128 + (command == 0xd4 ? 64 : 0), 64); return 64;
+  }
+  if (command == 0xd2 && length == sizeof(health_t)) {
+    health_t state{}; state.flags_pkt = HEALTH_FLAG_IGNITION_LINE;
+    if (native->controls()) state.flags_pkt |= HEALTH_FLAG_CONTROLS_ALLOWED;
+    if (*native->rx_invalid) state.flags_pkt |= HEALTH_FLAG_SAFETY_RX_CHECKS_INVALID;
+    state.safety_mode_pkt = native->mode(); state.safety_param_pkt = native->param(); state.alternative_experience_pkt = *native->ae;
+    if (native->relay()) state.faults_pkt |= 1U;
+    memcpy(data, &state, sizeof(state)); return sizeof(state);
+  }
+  if (command == 0xd5 && length == sizeof(aol_safety_health_t)) {
+    const auto state = native->status(); memcpy(data, &state, sizeof(state)); return sizeof(state);
+  }
+  if ((command == 0xc2 && length == sizeof(can_health_t)) || (command == 0xb2 && length == 2)) {
+    memset(data, 0, length); return length;
+  }
+  return 0;
+}
+int write_bulk(unsigned char endpoint, unsigned char *data, int length) {
+  CHECK(endpoint == 3); std::lock_guard lock(native_lock); native->clock();
+  for (int pos = 0; pos < length;) {
+    CHECK(pos + (int)sizeof(can_header) <= length); can_header header{}; memcpy(&header, data + pos, sizeof(header));
+    const auto size = dlc_to_len[header.data_len_code]; CHECK(pos + (int)sizeof(header) + size <= length);
+    uint8_t checksum = 0; for (size_t i = 0; i < sizeof(header) + size; ++i) checksum ^= data[pos + i]; CHECK(checksum == 0);
+    Frame frame{header.addr, header.bus, {data + pos + sizeof(header), data + pos + sizeof(header) + size}};
+    auto msg = packet(frame); const bool accepted = native->tx(&msg); ++tx_count; tx_accepted += accepted;
+    const uint8_t permission = native->permission(); const bool controls = native->controls();
+    const bool granted = (permission & 3U) != 0 || controls;
+    active_tx_accepted += accepted && granted;
+    if (accepted && granted && granted_tx_samples.size() < 16)
+      granted_tx_samples.push_back(tx_sample(frame, permission, controls));
+    // Bounded observer hint only: Ioniq LKAS_ALT/LFA ActToiSta is LE bit52|2.
+    // Root must independently decode the full recorded payload and CP identity.
+    if (accepted && (frame.address == 0x110 || frame.address == 0x12A) && frame.data.size() >= 7 &&
+        ((frame.data[6] >> 4) & 3U) != 0 && actuator_hint_tx_samples.size() < 16)
+      actuator_hint_tx_samples.push_back(tx_sample(frame, permission, controls));
+    pos += sizeof(header) + size;
+  }
+  return length;
+}
+int read_bulk(unsigned char endpoint, unsigned char *data, int length) {
+  CHECK(endpoint == 0x81);
+  if (!start_ns) {
+    const auto deadline = aol_monotonic_ns() + 15000000000ULL;
+    while (!do_exit && !std::filesystem::exists(release_path)) {
+      CHECK(aol_monotonic_ns() < deadline); std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    std::ifstream release(release_path); start_ns = read<uint64_t>(release); CHECK(start_ns > 0);
+  }
+  const auto now = aol_monotonic_ns();
+  if (now >= start_ns + duration_ns) { do_exit = true; return 0; }
+  if (now < start_ns || cursor >= trace.size() || trace[cursor].offset > now - start_ns) return 0;
+  std::lock_guard lock(native_lock); native->clock();
+  int pos = 0;
+  // Keep due CAN in source order, including overdue batches. Never reset the
+  // source anchor or invent a healthy replacement for missed input.
+  while (cursor < trace.size() && trace[cursor].offset <= now - start_ns) {
+    size_t bytes = 0; for (const auto &frame : trace[cursor].frames) bytes += sizeof(can_header) + frame.data.size();
+    CHECK(bytes <= static_cast<size_t>(length)); if (pos + bytes > static_cast<size_t>(length)) break;
+    max_feed_late_ns = std::max(max_feed_late_ns, now - start_ns - trace[cursor].offset);
+    for (const auto &frame : trace[cursor].frames) {
+      const auto source = transport_source(frame.bus);
+      auto msg = packet(frame);
+      if (!source.returned && !source.rejected) {
+        ++rx_count; rx_rejected += !native->rx(&msg);
+      } else {
+        // A recorded TX echo/rejection is transport history, never a new
+        // vehicle RX packet or evidence for native health/permission.
+        playback_returned += source.returned; playback_rejected += source.rejected;
+      }
+      can_header header{}; header.addr = frame.address; header.bus = source.bus;
+      header.returned = source.returned; header.rejected = source.rejected;
+      header.data_len_code = msg.data_len_code; header.extended = msg.extended;
+      memcpy(data + pos, &header, sizeof(header)); memcpy(data + pos + sizeof(header), frame.data.data(), frame.data.size());
+      uint8_t checksum = 0; for (size_t i = 0; i < sizeof(header) + frame.data.size(); ++i) checksum ^= data[pos + i];
+      header.checksum = checksum; memcpy(data + pos, &header, sizeof(header)); pos += sizeof(header) + frame.data.size();
+    }
+    ++cursor;
+  }
+  return pos;
+}
+void opened(PandaCommsHandle *handle, const std::string &serial) { CHECK(serial == SERIAL); handle->hw_serial = serial; }
+} // namespace
+
+PandaUsbHandle::PandaUsbHandle(std::string serial) { opened(this, serial); }
+PandaUsbHandle::~PandaUsbHandle() { cleanup(); }
+void PandaUsbHandle::cleanup() {}
+std::vector<std::string> PandaUsbHandle::list() { return {SERIAL}; }
+int PandaUsbHandle::control_write(uint8_t r, uint16_t p1, uint16_t p2, unsigned int) { return write_control(r, p1, p2); }
+int PandaUsbHandle::control_read(uint8_t r, uint16_t, uint16_t, unsigned char *d, uint16_t n, unsigned int) { return read_control(r, d, n); }
+int PandaUsbHandle::bulk_write(unsigned char e, unsigned char *d, int n, unsigned int) { return write_bulk(e, d, n); }
+int PandaUsbHandle::bulk_read(unsigned char e, unsigned char *d, int n, unsigned int) { return read_bulk(e, d, n); }
+PandaSpiHandle::PandaSpiHandle(std::string) { throw std::runtime_error("SPI unavailable in private bench"); }
+PandaSpiHandle::~PandaSpiHandle() {}
+void PandaSpiHandle::cleanup() {}
+std::vector<std::string> PandaSpiHandle::list() { return {}; }
+int PandaSpiHandle::control_write(uint8_t, uint16_t, uint16_t, unsigned int) { CHECK(false); return -1; }
+int PandaSpiHandle::control_read(uint8_t, uint16_t, uint16_t, unsigned char *, uint16_t, unsigned int) { CHECK(false); return -1; }
+int PandaSpiHandle::bulk_write(unsigned char, unsigned char *, int, unsigned int) { CHECK(false); return -1; }
+int PandaSpiHandle::bulk_read(unsigned char, unsigned char *, int, unsigned int) { CHECK(false); return -1; }
+
+#define main production_pandad_entry
+#include "selfdrive/pandad/main.cc"
+#undef main
+
+int main(int argc, char **argv) {
+  CHECK(argc == 5);
+  CHECK(getenv("SP_PRIVATE_NATIVE_BENCH") && std::string(getenv("SP_PRIVATE_NATIVE_BENCH")) == "1");
+  CHECK(getenv("OPENPILOT_PREFIX") && std::string(getenv("OPENPILOT_PREFIX")).rfind("replay-", 0) == 0);
+  CHECK(getenv("PARAMS_ROOT") && std::string(getenv("PARAMS_ROOT")).rfind("/data/params", 0) != 0);
+  for (const char *name : {"REPLAY", "SIMULATION", "STARTED", "FAKESEND", "BOARDD_LOOPBACK", "BOARDD_SKIP_FW_CHECK"}) CHECK(!getenv(name));
+  for (const char *name : {"PANDAD_MAXOUT", "CEREAL_FAKE", "STARPILOT_PANDA_FIRMWARE_MAP", "STARPILOT_PANDA_FIRMWARE_NAME"}) CHECK(!getenv(name));
+  CHECK(!getenv("AOL_REPLAY_RUNTIME") || std::string(getenv("AOL_REPLAY_RUNTIME")) == "0");
+  const auto parent = getppid(); CHECK(parent > 1);
+  CHECK(prctl(PR_SET_PDEATHSIG, SIGKILL) == 0 && getppid() == parent);
+  library_path = argv[1]; release_path = argv[3]; receipt_path = argv[4]; load_trace(argv[2]);
+  native = std::make_unique<Native>(); setenv("NO_FAN_CONTROL", "1", 1);
+  std::string command = "private-virtual-pandad", serial = SERIAL;
+  char *args[] = {command.data(), serial.data()}; const int result = production_pandad_entry(2, args);
+  std::ofstream out(receipt_path); CHECK(out.good());
+  out << "{\"physicalCanIo\":false,\"actualProductionPandad\":true,\"actualNativeSafety\":true,"
+      << "\"nativeTimer\":\"host CLOCK_MONOTONIC microseconds\",\"traceBatches\":" << trace.size()
+      << ",\"consumedBatches\":" << cursor << ",\"rx\":" << rx_count << ",\"rxRejected\":" << rx_rejected
+      << ",\"traceIncomingFrames\":" << trace_rx << ",\"traceHistoricalReturnedFrames\":" << trace_returned
+      << ",\"traceHistoricalRejectedFrames\":" << trace_rejected
+      << ",\"publishedHistoricalReturnedFrames\":" << playback_returned
+      << ",\"publishedHistoricalRejectedFrames\":" << playback_rejected
+      << ",\"tx\":" << tx_count << ",\"txAccepted\":" << tx_accepted << ",\"txAcceptedWhileNativeGranted\":" << active_tx_accepted
+      << ",\"maxFeedLateNs\":" << max_feed_late_ns << ",\"permissionMask\":" << (unsigned)native->permission()
+      << ",\"requestMask\":" << (unsigned)native->mask() << ",\"nativeRxInvalid\":" << (*native->rx_invalid ? "true" : "false")
+      << ",\"acceptedGrantedTxSamples\":";
+  write_tx_samples(out, granted_tx_samples);
+  out << ",\"acceptedIoniqActuatorHintSamples\":"; write_tx_samples(out, actuator_hint_tx_samples); out << "}\n";
+  native.reset(); return result;
+}

tools/heatdeath/normal_daemon_entry_v1.py

@@ -0,0 +1,155 @@
+"""Private early import / final-CP gate around unchanged normal daemon mains."""
+
+import argparse
+import hashlib
+import importlib
+import json
+import os
+from pathlib import Path
+import sys
+import time
+
+
+def digest(path):
+  return hashlib.sha256(Path(path).read_bytes()).hexdigest()
+
+
+def main():
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('--module', required=True)
+  parser.add_argument('--module-sha256', required=True)
+  parser.add_argument('--gate', type=Path, required=True)
+  parser.add_argument('--output', type=Path)
+  parser.add_argument('--limit', type=int, default=16000)
+  args = parser.parse_args()
+  assert args.module in ('openpilot.selfdrive.selfdrived.selfdrived', 'openpilot.selfdrive.controls.controlsd')
+  assert os.geteuid() == 0 and sys.platform == 'linux' and sys.dont_write_bytecode
+  assert os.environ.get('SP_PRIVATE_NATIVE_BENCH') == '1'
+  assert os.environ['OPENPILOT_PREFIX'].startswith('replay-bench-')
+  assert all(name not in os.environ for name in ('REPLAY', 'SIMULATION', 'CEREAL_FAKE', 'FAKESEND', 'STARTED'))
+  gate = args.gate.resolve()
+  assert gate.parent / 'params' == Path(os.environ['PARAMS_ROOT']).resolve()
+  assert not gate.is_relative_to(Path.cwd().resolve())
+  # Cold imports overlap Card initialization; normal main/scheduler remain
+  # untouched and execute only after the parent validates final CP bytes.
+  module = importlib.import_module(args.module)
+  assert digest(module.__file__) == args.module_sha256
+  from actor import install
+
+  install(module, 'controlsd' if args.module.endswith('controlsd') else 'selfdrived')
+  deadline = time.monotonic_ns() + 12_000_000_000
+  while True:
+    assert time.monotonic_ns() < deadline, 'No parent CP token within12s after imports'
+    try:
+      token = json.loads(gate.read_text())
+    except FileNotFoundError:
+      time.sleep(0.005)
+      continue
+    assert token['kind'] == 'private-verified-car-params-main-gate-v1'
+    assert token['prefix'] == os.environ['OPENPILOT_PREFIX']
+    assert token['paramsRoot'] == os.environ['PARAMS_ROOT']
+    assert token['carParamsSha256'] == os.environ['SP_DAEMON_MAIN_GATE_CP_SHA256']
+    assert 0 < token['verifiedMonoNs'] <= time.monotonic_ns()
+    from openpilot.common.params import Params
+
+    assert hashlib.sha256(Params().get('CarParams')).hexdigest() == token['carParamsSha256']
+    break
+  if args.output is None:
+    module.main()
+    return
+  from phase_timing import PhaseCapture
+  from openpilot.common.realtime import Ratekeeper
+
+  capture = PhaseCapture(args.output, args.limit, 'private-normal-daemon-ratekeeper-only')
+  capture.add(
+    Ratekeeper,
+    'monitor_time',
+    'Ratekeeper.monitor_time',
+    sample=lambda values, kwargs, result, raised: (round(values[0].avg_dt.get_average() * 1e9), int(values[0].lagging)),
+    sample_columns=('averagePeriodNs', 'lagging'),
+  )
+  if os.environ.get('SP_PRIVATE_LOOP_MACRO') == '1':
+    if args.module.endswith('selfdrived'):
+      owner = module.SelfdriveD
+      capture.add(owner, 'step', 'SelfdriveD.step', cycle=True)
+      for name in ('data_sample', 'update_events', 'update_alerts', '_decide_aol_axes', 'publish_axis_proof', 'publish_selfdriveState'):
+        capture.add(owner, name, 'SelfdriveD.' + name)
+    else:
+      owner = module.Controls
+      capture.add(owner, 'update', 'Controls.update', cycle=True)
+      capture.add(owner, 'state_control', 'Controls.state_control')
+      capture.add(owner, 'publish', 'Controls.publish')
+  profiler = None
+  if os.environ.get('SP_PRIVATE_SPARSE_PROFILE') == '1':
+    import cProfile
+    from functools import wraps
+
+    profiler = cProfile.Profile()
+    owner, name = (module.SelfdriveD, 'update_events') if args.module.endswith('selfdrived') else (module.Controls, 'state_control')
+    original, samples = getattr(owner, name), [0, 0]
+
+    @wraps(original)
+    def sparse(*values, **kwargs):
+      samples[0] += 1
+      if samples[0] % 197:
+        return original(*values, **kwargs)
+      samples[1] += 1
+      profiler.enable()
+      try:
+        return original(*values, **kwargs)
+      finally:
+        profiler.disable()
+
+    setattr(owner, name, sparse)
+  if os.environ.get('SP_PRIVATE_CONTROLS_FIFO52') == '1' and args.module.endswith('controlsd'):
+    from functools import wraps
+    from openpilot.common.realtime import Priority
+
+    original_run = module.Controls.run
+
+    @wraps(original_run)
+    def private_fifo_run(owner):
+      assert Path('/data/params/d/IsOffroad').read_bytes().strip() == b'1'
+      assert Path('/data/params/d/IsEngaged').read_bytes().strip() == b'0'
+      before = {'policy': os.sched_getscheduler(0), 'priority': os.sched_getparam(0).sched_priority, 'affinity': sorted(os.sched_getaffinity(0))}
+      assert before == {'policy': os.SCHED_OTHER, 'priority': 0, 'affinity': [6]}
+      target = int(Priority.CTRL_HIGH) - 1
+      assert target == 52
+      os.sched_setscheduler(0, os.SCHED_FIFO, os.sched_param(target))
+      after = {'policy': os.sched_getscheduler(0), 'priority': os.sched_getparam(0).sched_priority, 'affinity': sorted(os.sched_getaffinity(0))}
+      assert after == {'policy': os.SCHED_FIFO, 'priority': 52, 'affinity': [6]}
+      args.output.with_suffix('.private-policy.json').write_text(
+        json.dumps(
+          {
+            'pid': os.getpid(),
+            'before': before,
+            'after': after,
+            'scope': 'Private owned benchmark main thread only; production source and normal daemons untouched',
+            'normal_constructor_and_scheduler_completed': True,
+          }
+        )
+      )
+      return original_run(owner)
+
+    module.Controls.run = private_fifo_run
+  try:
+    with capture:
+      module.main()
+  finally:
+    if profiler is not None:
+      profiler.dump_stats(str(args.output.with_suffix('.pstats')))
+      args.output.with_suffix('.profile.json').write_text(
+        json.dumps(
+          {
+            'function': name,
+            'calls': samples[0],
+            'sampled': samples[1],
+            'period': 197,
+            'scope': 'Sparse diagnostic only; cProfile overhead and periodic sampling do not prove headroom or whole-cycle CPU',
+          }
+        )
+      )
+
+
+if __name__ == '__main__':
+  main()

tools/heatdeath/owned_msgq_cleanup_v2.py

@@ -0,0 +1,118 @@
+"""Private exact-created namespace cleanup, only after caller reaps all actors."""
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import stat
+import time
+
+
+def atomic(path, value):
+  path = Path(path)
+  temporary = path.with_name(path.name + '.tmp-' + str(os.getpid()))
+  with temporary.open('x') as stream:
+    stream.write(json.dumps(value, indent=2) + '\n')
+  os.replace(temporary, path)
+
+
+def journal_created(path, prefix, destination):
+  path = Path(path)
+  assert re.fullmatch(r'replay-bench-(?:camera-)?[0-9a-f]{12}', prefix)
+  assert path == Path('/dev/shm') / ('msgq_' + prefix)
+  info = path.lstat()
+  assert stat.S_ISDIR(info.st_mode) and info.st_uid == 0
+  fields = Path('/proc/self/stat').read_text().rsplit(')', 1)[1].split()
+  value = {
+    'path': str(path),
+    'prefix': prefix,
+    'device': info.st_dev,
+    'inode': info.st_ino,
+    'creatorPid': os.getpid(),
+    'creatorStartTicks': int(fields[19]),
+    'createdMonoNs': time.monotonic_ns(),
+    'bootId': Path('/proc/sys/kernel/random/boot_id').read_text().strip(),
+    'leaseTokenSha256': hashlib.sha256(os.environ['SP_PRIVATE_IRQ_LEASE_TOKEN'].encode()).hexdigest(),
+  }
+  assert not Path(destination).exists()
+  atomic(destination, value)
+
+
+def identity(pid):
+  try:
+    return int(Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split()[19])
+  except (FileNotFoundError, ProcessLookupError):
+    return None
+
+
+def no_users(path, prefix, owned_inodes):
+  users, inspected = [], 0
+  for proc in Path('/proc').iterdir():
+    if not proc.name.isdecimal():
+      continue
+    pid = int(proc.name)
+    start = identity(pid)
+    if start is None:
+      continue
+    try:
+      environ = (proc / 'environ').read_bytes().split(b'\0')
+      if ('OPENPILOT_PREFIX=' + prefix).encode() in environ:
+        users.append({'pid': pid, 'startTicks': start, 'kind': 'environment'})
+      for line in (proc / 'maps').read_text().splitlines():
+        fields = line.split(maxsplit=5)
+        if len(fields) == 6 and (fields[5] == str(path) or fields[5].startswith(str(path) + '/')):
+          users.append({'pid': pid, 'startTicks': start, 'kind': 'mapping'})
+      for fd in (proc / 'fd').iterdir():
+        try:
+          target = os.readlink(fd)
+          info = fd.stat()
+        except FileNotFoundError:
+          continue  # descriptor closed during scan
+        if target == str(path) or target.startswith(str(path) + '/') or (info.st_dev, info.st_ino) in owned_inodes:
+          users.append({'pid': pid, 'startTicks': start, 'kind': 'descriptor'})
+      assert identity(pid) == start or identity(pid) is None, 'PID changed while scanning'
+      inspected += 1
+    except (FileNotFoundError, ProcessLookupError):
+      assert identity(pid) is None, 'Live process incompletely scanned'
+  return {'monoNs': time.monotonic_ns(), 'inspectedProcesses': inspected, 'users': users}
+
+
+def cleanup(journal, receipt):
+  result = {'removed': False, 'failure': None, 'scans': []}
+  try:
+    value = json.loads(Path(journal).read_text())
+    result['ownership'] = value
+    assert value['bootId'] == Path('/proc/sys/kernel/random/boot_id').read_text().strip()
+    assert value['leaseTokenSha256'] == hashlib.sha256(os.environ['SP_PRIVATE_IRQ_LEASE_TOKEN'].encode()).hexdigest()
+    prefix, path = value['prefix'], Path(value['path'])
+    assert re.fullmatch(r'replay-bench-(?:camera-)?[0-9a-f]{12}', prefix)
+    assert path == Path('/dev/shm') / ('msgq_' + prefix)
+    info = path.lstat()
+    assert stat.S_ISDIR(info.st_mode) and info.st_uid == 0
+    assert (info.st_dev, info.st_ino) == (value['device'], value['inode'])
+    children = list(path.iterdir())
+    states = {child.name: child.lstat() for child in children}
+    assert all(stat.S_ISREG(item.st_mode) and item.st_uid == 0 and item.st_dev == info.st_dev for item in states.values())
+    inodes = {(item.st_dev, item.st_ino) for item in states.values()} | {(info.st_dev, info.st_ino)}
+    result['allocatedBytesBefore'] = sum(item.st_blocks * 512 for item in states.values())
+    for _ in range(2):
+      scan = no_users(path, prefix, inodes)
+      result['scans'].append(scan)
+      assert not scan['users'], 'Namespace has active users'
+      time.sleep(0.05)
+    current = path.lstat()
+    assert (current.st_dev, current.st_ino) == (info.st_dev, info.st_ino)
+    assert {child.name: (child.lstat().st_dev, child.lstat().st_ino) for child in path.iterdir()} == {
+      name: (item.st_dev, item.st_ino) for name, item in states.items()
+    }
+    assert shutil.rmtree.avoids_symlink_attacks
+    shutil.rmtree(path)
+    assert not path.exists()
+    result['removed'] = True
+  except BaseException as error:
+    result['failure'] = type(error).__name__ + ': ' + str(error)
+    raise
+  finally:
+    atomic(receipt, result)

tools/heatdeath/peer_feeder_v7_protected.py

@@ -0,0 +1,214 @@
+"""Owned independent peer publisher; no observer, disk I/O or authority checks in timed loop."""
+
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import signal
+import struct
+import sys
+import time
+
+from peer_timing_probe_v2 import Probe, point
+
+
+def main():
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('--repo', required=True)
+  parser.add_argument('--inputs', type=Path, required=True)
+  parser.add_argument('--release', type=Path, required=True)
+  parser.add_argument('--clocks', type=Path, required=True)
+  parser.add_argument('--output', type=Path, required=True)
+  args = parser.parse_args()
+  assert os.geteuid() == 0 and sys.platform == 'linux' and sys.dont_write_bytecode
+  assert os.environ.get('SP_PRIVATE_NATIVE_BENCH') == '1'
+  assert os.environ['OPENPILOT_PREFIX'].startswith('replay-bench-')
+  assert all(name not in os.environ for name in ('REPLAY', 'SIMULATION', 'CEREAL_FAKE', 'FAKESEND', 'STARTED'))
+  output = args.output.resolve()
+  assert output.parent / 'params' == Path(os.environ['PARAMS_ROOT']).resolve()
+  assert output.parent == args.release.resolve().parent == args.clocks.resolve().parent
+  assert not output.exists()
+  from driver_aa_v47_fixture import configure_imports, read_peers, reanchor, OWNED
+
+  configure_imports(args.repo)
+  from openpilot.cereal import messaging
+
+  meta = json.loads((args.inputs / 'inputs.json').read_text())
+  for name, expected in meta['files'].items():
+    assert hashlib.sha256((args.inputs / name).read_bytes()).hexdigest() == expected
+  suppressed = {}
+  peers = read_peers(args.inputs / 'peers.trace', messaging, suppressed)
+  services = sorted({event.which() for _, event in peers})
+  assert not set(services) & (OWNED | {'managerState'})
+  os.environ['MSGQ_PREALLOC'] = '1'  # Current msgq prefaults this private stimulus producer's queues before readiness.
+  publisher = messaging.PubMaster(services)
+  os.sched_setscheduler(0, os.SCHED_OTHER, os.sched_param(0))
+  # Only the fixture moves: real UI uses0, virtual pandad2, physical pandad3.
+  fixture_cpus = {1} if os.environ.get('SP_PRIVATE_PEER_CORE1') == '1' else {3} if os.environ.get('SP_PRIVATE_PEER_CORE3') == '1' else {0, 1, 2, 3}
+  os.sched_setaffinity(0, fixture_cpus)
+  from placement import apply
+
+  fixture_cpus = set(apply('peer'))
+  # Materialize real source payloads before readiness, without sending any
+  # warm-up stimulus. Exercise the exact field/serializer path at zero clock
+  # delta, prove byte equality, then retain each single-use mutable builder.
+  prewarm_start = time.monotonic_ns()
+  prepared, prepared_bytes = [], 0
+  assert len(peers) <= 100000, 'Bounded prepared peer event count exceeded'
+  for offset, event in peers:
+    service, builder = event.which(), event.as_builder()
+    original = builder.to_bytes()
+    builder.clear_write_flag()
+    reanchor(builder, offset, meta, meta['sourceStartMonoNs'], meta['sourceStartMonoNs'] + meta['sourceBootMinusMonoNs'])
+    assert builder.to_bytes() == original, 'Prewarm changed actual source bytes'
+    builder.clear_write_flag()
+    prepared_bytes += len(original)
+    assert prepared_bytes <= 128 * 1024 * 1024, 'Bounded prepared peer payload bytes exceeded'
+    prepared.append((offset, service, builder))
+  del peers
+  peers = prepared
+  prewarm = {
+    'events': len(peers),
+    'payloadBytes': prepared_bytes,
+    'durationNs': time.monotonic_ns() - prewarm_start,
+    'zeroDeltaByteEquality': True,
+    'publishedStimulus': 0,
+    'transportWarmup': 'PubMaster queue construction only; no warm-up send or IPC memory modification',
+  }
+  os.sched_setscheduler(0, os.SCHED_FIFO, os.sched_param(1))
+  assert os.sched_getscheduler(0) == os.SCHED_FIFO and os.sched_getparam(0).sched_priority == 1
+  fixture_policy = {
+    'scope': 'Private test stimulus main thread only; after CFS prewarm, before readiness/release',
+    'policy': os.sched_getscheduler(0),
+    'priority': os.sched_getparam(0).sched_priority,
+    'affinity': sorted(os.sched_getaffinity(0)),
+    'productionPolicyChange': False,
+  }
+  probe = Probe(services)
+  counts = dict.fromkeys(services, 0)
+  index = maximum_late = maximum_dispatch = maximum_send = budget_yields = 0
+  failure = None
+  started = finished = None
+  pre_release = True
+
+  def stop(signum, _frame):
+    raise SystemExit('peer feeder signal ' + str(signum))
+
+  for signum in (signal.SIGTERM, signal.SIGHUP, signal.SIGINT):
+    signal.signal(signum, stop)
+  try:
+    ready = output.with_name('peer-ready.json')
+    assert not ready.exists()
+    ready_temporary = ready.with_suffix('.tmp')
+    assert not ready_temporary.exists()
+    ready_temporary.write_text(
+      json.dumps(
+        {
+          'pid': os.getpid(),
+          'services': services,
+          'events': len(peers),
+          'prewarm': prewarm,
+          'fixturePolicy': fixture_policy,
+          'suppressedRecordedOwners': suppressed,
+          'readyMonoNs': time.monotonic_ns(),
+        }
+      )
+      + '\n'
+    )
+    ready_temporary.replace(ready)
+    wait_deadline = time.monotonic_ns() + 30_000_000_000
+    while not args.release.exists():
+      assert time.monotonic_ns() < wait_deadline, 'No native release within30s of peer readiness'
+      probe.sleep(0.005, 0)
+    clocks = json.loads(args.clocks.read_text())
+    mono, boot = clocks['monoStartNs'], clocks['bootStartNs']
+    assert type(mono) is type(boot) is int and mono > 0 and boot > 0
+    assert args.release.read_bytes() == struct.pack('<Q', mono)
+    assert clocks['prefix'] == os.environ['OPENPILOT_PREFIX']
+    deadline = mono + meta['durationNs']
+    started = time.monotonic_ns()
+    pre_release = False
+    while time.monotonic_ns() < deadline:
+      now = time.monotonic_ns()
+      next_offset = peers[index][0] if index < len(peers) else meta['durationNs']
+      probe.loop(index, next_offset, now - mono - next_offset)
+      if now < mono:
+        probe.sleep(min(0.001, (mono - now) / 1e9), 1)
+        continue
+      dispatch_start, dispatched = time.monotonic_ns(), 0
+      dispatch_point = point()
+      while index < len(peers) and peers[index][0] <= now - mono:
+        if dispatched >= 64 or time.monotonic_ns() - dispatch_start >= 2_000_000:
+          budget_yields += 1
+          break  # Fairness yield; actual source lateness remains a hard failure.
+        offset, service, builder = peers[index]
+        late = time.monotonic_ns() - mono - offset
+        maximum_late = max(maximum_late, late)
+        if late > 20_000_000:
+          probe.dispatch(index, dispatched, dispatch_point, point())
+          probe.send(service, index, offset, late, -1, 0, -1, [point()])
+        assert late <= 20_000_000, 'Peer source feed >20ms late; harness invalid'
+        send_start = time.monotonic_ns()
+        points, stage, completed = [point()], 0, 0
+        try:
+          publisher.send(service, reanchor(builder, offset, meta, mono, boot))
+          completed = time.monotonic_ns()
+          stage = 3
+        finally:
+          points.append(point())
+          probe.send(service, index, offset, completed - mono - offset if completed else late, stage, completed, -1, points)
+        counts[service] += 1
+        index += 1
+        dispatched += 1
+        maximum_send = max(maximum_send, completed - send_start)
+        maximum_late = max(maximum_late, completed - mono - offset)
+        assert completed - mono - offset <= 20_000_000, 'Peer source publication completed >20ms late; harness invalid'
+      probe.dispatch(index, dispatched, dispatch_point, point())
+      maximum_dispatch = max(maximum_dispatch, time.monotonic_ns() - dispatch_start)
+      probe.sleep(0.001, 2)
+    assert index == len(peers), 'Unpublished peer source events at source deadline'
+  except BaseException as error:
+    failure = type(error).__name__ + ': ' + str(error)
+  finally:
+    finished = time.monotonic_ns()
+    probe_path = probe.export(output)
+    output.write_text(
+      json.dumps(
+        {
+          'kind': 'private-independent-peer-feeder-v7-protected-source',
+          'timingProbePath': probe_path,
+          'timingProbeErrors': probe.errors,
+          'msgqPreallocEnv': os.environ.get('MSGQ_PREALLOC'),
+          'failure': failure,
+          'events': len(peers),
+          'published': index,
+          'counts': counts,
+          'maximumLatenessNs': maximum_late,
+          'maximumDispatchNs': maximum_dispatch,
+          'startedMonoNs': started,
+          'finishedMonoNs': finished,
+          'maximumSendNs': maximum_send,
+          'dispatchBudgetYields': budget_yields,
+          'dispatchBudgetNs': 2000000,
+          'dispatchMessagesPerPassMaximum': 64,
+          'sourceLatenessLimitNs': 20000000,
+          'latenessCheckedBeforeAndAfterPublication': True,
+          'prewarm': prewarm,
+          'fixturePolicy': fixture_policy,
+          'failedBeforeRelease': pre_release,
+          'suppressedRecordedOwners': suppressed,
+          'policy': os.sched_getscheduler(0),
+          'priority': os.sched_getparam(0).sched_priority,
+          'affinity': sorted(os.sched_getaffinity(0)),
+          'timedDiskWrites': False,
+        },
+        indent=2,
+      )
+      + '\n'
+    )
+  assert failure is None, failure
+
+
+if __name__ == '__main__':
+  main()

tools/heatdeath/peer_timing_probe_v2.py

@@ -0,0 +1,163 @@
+"""Bounded private observations only; no input, policy, IPC or clock overrides."""
+
+from array import array
+import gc
+import hashlib
+import json
+import resource
+import sys
+import time
+
+WIDTH = 32
+POINT_COLUMNS = ['wallMonoNs', 'threadCpuNs', 'minflt', 'majflt', 'nvcsw', 'nivcsw']
+
+
+def point():
+  wall, cpu = time.monotonic_ns(), time.thread_time_ns()
+  usage = resource.getrusage(resource.RUSAGE_THREAD)
+  return wall, cpu, usage.ru_minflt, usage.ru_majflt, usage.ru_nvcsw, usage.ru_nivcsw
+
+
+class Ring:
+  def __init__(self, limit):
+    self.limit, self.seen = limit, 0
+    self.data = array('q', [0]) * (limit * WIDTH)
+
+  def put(self, values):
+    assert len(values) < WIDTH
+    position = (self.seen % self.limit) * WIDTH
+    self.data[position] = self.seen
+    for i in range(WIDTH - 1):
+      self.data[position + 1 + i] = int(values[i]) if i < len(values) else 0
+    self.seen += 1
+
+  def export(self, path):
+    with path.open('xb') as stream:
+      self.data.tofile(stream)
+    return {
+      'path': str(path),
+      'sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
+      'callsSeen': self.seen,
+      'rowsRetained': min(self.seen, self.limit),
+      'rowsDropped': max(0, self.seen - self.limit),
+      'limit': self.limit,
+    }
+
+
+class Probe:
+  def __init__(self, services):
+    assert len(services) <= 64
+    self.services = services
+    self.ids = {name: i for i, name in enumerate(services)}
+    self.attempts = dict.fromkeys(services, 0)
+    self.first, self.outliers, self.sleeps = Ring(32), Ring(16), Ring(32)
+    self.sleep_calls, self.errors = 0, []
+    self.sleep_kinds = dict.fromkeys(range(3), 0)
+    self.dispatch_calls = 0
+    self.last_loop = point()
+    self.gc = Ring(64)
+    self.gc_callback = self.observe_gc
+    gc.callbacks.append(self.gc_callback)
+
+  def observe_gc(self, phase, info):
+    try:
+      self.gc.put(
+        (0 if phase == "start" else 1, info["generation"], time.monotonic_ns(), time.thread_time_ns(), info.get("collected", 0), info.get("uncollectable", 0))
+      )
+    except Exception as exc:
+      self.note_error(exc)
+
+  def note_error(self, exc):
+    if len(self.errors) < 16:
+      self.errors.append(type(exc).__name__ + ': ' + str(exc))
+
+  def send(self, service, event_index, offset, late, stage, completed, payload_bytes, points):
+    try:
+      attempt = self.attempts[service]
+      self.attempts[service] = attempt + 1
+      # -1: pre-guard failure;0: exception in original mixed PubMaster call;
+      # 3: completed send (including later unchanged post-guard failure).
+      values = (self.ids[service], event_index, offset, late, stage, completed, payload_bytes, *sum((tuple(p) for p in points), ()))
+      if self.first.seen < 32:
+        self.first.put(values)
+      elapsed = points[-1][0] - points[0][0] if len(points) > 1 else 0
+      if stage != 3 or late > 20_000_000 or elapsed > 1_000_000:
+        self.outliers.put(values)
+    except Exception as exc:
+      self.note_error(exc)
+
+  def sleep(self, seconds, kind):
+    before = point()
+    try:
+      return time.sleep(seconds)
+    finally:
+      after = point()
+      try:
+        self.sleep_calls += 1
+        self.sleep_kinds[kind] += 1
+        requested = round(seconds * 1e9)
+        elapsed = after[0] - before[0]
+        if self.sleep_kinds[kind] <= 8 or elapsed - requested > 2_000_000:
+          self.sleeps.put((kind, requested, elapsed - requested, *before, *after))
+      except Exception as exc:
+        self.note_error(exc)
+
+  def loop(self, event_index, offset, late):
+    # Capture off-CPU gaps not limited to the explicit sleep syscall.
+    current = point()
+    previous, self.last_loop = self.last_loop, current
+    if current[0] - previous[0] > 5_000_000 or late > 20_000_000:
+      try:
+        self.sleeps.put((3, event_index, offset, late, *previous, *current))
+      except Exception as exc:
+        self.note_error(exc)
+
+  def dispatch(self, event_index, dispatched, before, after):
+    try:
+      self.dispatch_calls += 1
+      if self.dispatch_calls <= 4 or after[0] - before[0] > 2_000_000:
+        self.sleeps.put((4, event_index, dispatched, *before, *after))
+    except Exception as exc:
+      self.note_error(exc)
+
+  def export(self, output):
+    gc.callbacks.remove(self.gc_callback)
+    result = {
+      'kind': 'private-peer-timing-probe-v2',
+      'services': self.services,
+      'pointColumns': POINT_COLUMNS,
+      'widthInt64': WIDTH,
+      'byteorder': sys.byteorder,
+      'scalarBytes': array('q').itemsize,
+      'observerErrors': self.errors,
+      'sendAttempts': self.attempts,
+      'sleepCalls': self.sleep_calls,
+      'sleepKinds': self.sleep_kinds,
+      'dispatchCalls': self.dispatch_calls,
+      'sendColumnsAfterSequence': [
+        'serviceIndex',
+        'eventIndex',
+        'sourceOffsetNs',
+        'latenessNs',
+        'stage',
+        'completedMonoNs',
+        'payloadBytes',
+        'points:beforeMixedPubMasterSend,afterMixedPubMasterSend',
+      ],
+      'sleepColumnsAfterSequence': ['kind0release1prestart2yield', 'requestedNs', 'oversleepNs', 'beforePoint', 'afterPoint'],
+      'dispatchKind4ColumnsAfterSequence': ['kind4', 'eventIndex', 'dispatched', 'beforePoint', 'afterPoint'],
+      'loopGapKind3ColumnsAfterSequence': ['kind3', 'eventIndex', 'offsetNs', 'lateNs', 'previousLoopPoint', 'currentLoopPoint'],
+      'gcColumnsAfterSequence': ['phase0start1stop', 'generation', 'wallMonoNs', 'threadCpuNs', 'collected', 'uncollectable'],
+      'gcCallbackScope': 'Installed before readiness/release; removed before export; no GC policy change',
+      'rings': {
+        name: ring.export(output.with_name(output.stem + '-probe-' + name + '.bin'))
+        for name, ring in (('first', self.first), ('outliers', self.outliers), ('sleep', self.sleeps), ('gc', self.gc))
+      },
+      'scope': 'Diagnostic overhead included in original strict20ms delivery budget; no warm-up messages or scheduling/prealloc change',
+    }
+    path = output.with_name(output.stem + '-probe.json')
+    temporary = path.with_suffix('.json.tmp')
+    assert not temporary.exists() and not path.exists()
+    temporary.write_text(json.dumps(result, indent=2) + '\n')
+    temporary.replace(path)
+    return str(path)

tools/heatdeath/phase_timing.py

@@ -0,0 +1,194 @@
+"""Private bounded observers; original calls, clocks and scheduling stay intact."""
+
+from array import array
+from functools import wraps
+import hashlib
+import itertools
+import json
+import os
+from pathlib import Path
+import signal
+import sys
+import threading
+import time
+
+
+SAMPLE_WIDTH = 12
+BASE_COLUMNS = ['sequence', 'nativeTid', 'iteration', 'startMonoNs', 'endMonoNs', 'startThreadCpuNs', 'endThreadCpuNs', 'raised']
+
+
+def private_output(value):
+  output = Path(value).resolve()
+  assert sys.dont_write_bytecode and sys.platform == 'linux'
+  assert os.environ.get('SP_PRIVATE_NATIVE_BENCH') == '1'
+  assert os.environ.get('OPENPILOT_PREFIX', '').startswith('replay-bench-')
+  assert all(name not in os.environ for name in ('REPLAY', 'SIMULATION', 'CEREAL_FAKE', 'FAKESEND', 'STARTED'))
+  assert Path(os.environ['PARAMS_ROOT']).resolve() == output.parent / 'params'
+  assert not output.exists() and not output.is_relative_to(Path.cwd().resolve())
+  assert output.parent.is_dir()
+  return output
+
+
+class Rows:
+  WIDTH = len(BASE_COLUMNS) + SAMPLE_WIDTH
+
+  def __init__(self, limit):
+    self.limit, self.seen = limit, 0
+    self.data = array('q', [0]) * (limit * self.WIDTH)
+
+  def put(self, tid, iteration, start, end, cpu_start, cpu_end, raised, samples):
+    sequence = self.seen
+    position = (sequence % self.limit) * self.WIDTH
+    data = self.data
+    data[position] = sequence
+    data[position + 1] = tid
+    data[position + 2] = iteration
+    data[position + 3] = start
+    data[position + 4] = end
+    data[position + 5] = cpu_start
+    data[position + 6] = cpu_end
+    data[position + 7] = int(raised)
+    for index in range(SAMPLE_WIDTH):
+      data[position + 8 + index] = int(samples[index]) if index < len(samples) else 0
+    self.seen = sequence + 1
+
+  def export(self, path):
+    # Native numeric ring is written directly; no millions-of-ints JSON pass.
+    with path.open('xb') as stream:
+      self.data.tofile(stream)
+    return {
+      'callsSeen': self.seen,
+      'rowsDropped': max(0, self.seen - self.limit),
+      'rowsRetained': min(self.seen, self.limit),
+      'path': str(path),
+      'sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
+      'bytes': path.stat().st_size,
+      'rowOrder': 'For each retained sequence use ring slot sequence modulo rowsPerPhaseMaximum',
+    }
+
+
+class PhaseCapture:
+  def __init__(self, output, limit, kind):
+    assert 1 <= limit <= 16000
+    self.output, self.limit, self.kind = private_output(output), limit, kind
+    self.specs, self.bindings, self.rows, self.sources = [], [], {}, {}
+    self.local, self.iterations = threading.local(), itertools.count()
+    self.errors, self.sample_columns, self.old_handlers = {}, {}, {}
+    self.stop_signal = None
+
+  def add(self, owner, name, label, *, step=False, cycle=False, sample=None, sample_columns=()):
+    assert label not in self.rows and len(sample_columns) <= SAMPLE_WIDTH
+    self.rows[label] = Rows(self.limit)
+    self.sample_columns[label] = list(sample_columns)
+    self.specs.append((owner, name, label, step, cycle, sample))
+
+  def note_error(self, exc):
+    key = type(exc).__module__ + '.' + type(exc).__qualname__
+    self.errors[key] = self.errors.get(key, 0) + 1
+
+  def stop(self, signum, _frame):
+    self.stop_signal = signum
+    raise SystemExit(128 + signum)
+
+  def install(self, owner, name, label, step, cycle, sample):
+    original = getattr(owner, name)
+    source = sys.modules.get(getattr(original, '__module__', ''))
+    if source is not None and getattr(source, '__file__', None):
+      path = Path(source.__file__).resolve()
+      self.sources[str(path)] = hashlib.sha256(path.read_bytes()).hexdigest()
+    recorder, rows = self, self.rows[label]
+
+    @wraps(original)
+    def timed(*args, **kwargs):
+      prior_iteration = getattr(recorder.local, 'iteration', -1)
+      if step or cycle:
+        recorder.local.iteration = next(recorder.iterations)
+      iteration = getattr(recorder.local, 'iteration', -1)
+      tid = threading.get_native_id()
+      start, cpu_start = time.monotonic_ns(), time.thread_time_ns()
+      result, raised = None, False
+      try:
+        if recorder.stop_signal is not None:
+          raise SystemExit(128 + recorder.stop_signal)
+        result = original(*args, **kwargs)
+        return result
+      except BaseException:
+        raised = True
+        raise
+      finally:
+        end, cpu_end = time.monotonic_ns(), time.thread_time_ns()
+        try:
+          values = sample(args, kwargs, result, raised) if sample is not None else ()
+          assert isinstance(values, tuple) and len(values) <= SAMPLE_WIDTH
+          rows.put(tid, iteration, start, end, cpu_start, cpu_end, raised, values)
+        except Exception as exc:
+          recorder.note_error(exc)
+        finally:
+          if step and not cycle:
+            recorder.local.iteration = prior_iteration
+
+    self.bindings.append((owner, name, original, timed))
+    setattr(owner, name, timed)
+
+  def __enter__(self):
+    self.started_ns = time.monotonic_ns()
+    try:
+      for spec in self.specs:
+        self.install(*spec)
+      for signum in (signal.SIGTERM, signal.SIGHUP, signal.SIGINT):
+        self.old_handlers[signum] = signal.getsignal(signum)
+        signal.signal(signum, self.stop)
+    except BaseException:
+      self.restore()
+      raise
+    return self
+
+  def restore(self):
+    for owner, name, original, wrapped in reversed(self.bindings):
+      if getattr(owner, name) is wrapped:
+        setattr(owner, name, original)
+    for signum, original in self.old_handlers.items():
+      signal.signal(signum, original)
+
+  def __exit__(self, exc_type, exc, traceback):
+    # Ignore further parent signals only during the bounded final export.
+    for signum in self.old_handlers:
+      signal.signal(signum, signal.SIG_IGN)
+    try:
+      for owner, name, original, wrapped in reversed(self.bindings):
+        if getattr(owner, name) is wrapped:
+          setattr(owner, name, original)
+      record = {
+        'kind': self.kind,
+        'pid': os.getpid(),
+        'mainNativeTid': threading.get_native_id(),
+        'startedMonoNs': self.started_ns,
+        'finishedMonoNs': time.monotonic_ns(),
+        'stopSignal': self.stop_signal,
+        'rowsPerPhaseMaximum': self.limit,
+        'baseColumns': BASE_COLUMNS,
+        'sampleColumns': self.sample_columns,
+        'sources': self.sources,
+        'observerErrors': self.errors,
+        'timing': {
+          label: rows.export(self.output.with_name(self.output.stem + '.' + str(index) + '.bin')) for index, (label, rows) in enumerate(self.rows.items())
+        },
+        'binaryFormat': {
+          'scalar': 'signed 64-bit native endian',
+          'byteorder': sys.byteorder,
+          'scalarBytes': array('q').itemsize,
+          'columnsPerRow': Rows.WIDTH,
+          'unusedSlots': 'zero; only rowsRetained sequences are observations',
+        },
+        'scope': 'Real calls, wall and per-thread CPU; no new clocks, grants, inputs, GC policy or scheduler changes',
+        'observerEffect': 'Python wrappers, clock reads, primitive sampling and preallocated ring writes add overhead; no per-call I/O or JSON',
+        'inclusivePhases': True,
+        'samplingOrder': 'exit clocks precede observational scalar sampling',
+      }
+      temporary = self.output.with_suffix(self.output.suffix + '.tmp')
+      assert not temporary.exists() and not self.output.exists()
+      temporary.write_text(json.dumps(record, separators=(',', ':')) + '\n')
+      temporary.replace(self.output)
+    finally:
+      self.restore()
+    return False

tools/heatdeath/placement.py

@@ -0,0 +1,34 @@
+"""Affinity changes confined to the benchmark's owned processes."""
+
+import json
+import os
+from pathlib import Path
+
+NAMES = {'card', 'controlsd', 'selfdrived', 'radard', 'plannerd', 'peer', 'observer'}
+
+
+def load_plan(value):
+  plan = json.loads(value)
+  if not isinstance(plan, dict) or not set(plan) <= NAMES:
+    raise ValueError('Placement must map known actor names to CPU lists')
+  for cpus in plan.values():
+    if not isinstance(cpus, list) or not cpus or any(type(cpu) is not int or not 0 <= cpu < 8 for cpu in cpus):
+      raise ValueError('Each placement needs CPU integers from 0 through 7')
+    if len(set(cpus)) != len(cpus):
+      raise ValueError('Duplicate CPUs in placement')
+  return plan
+
+
+def apply(name):
+  plan = load_plan(os.environ.get('HEATDEATH_PLACEMENT', '{}'))
+  if name in plan:
+    os.sched_setaffinity(0, set(plan[name]))
+    if os.sched_getaffinity(0) != set(plan[name]):
+      raise RuntimeError('Requested affinity was not applied')
+  return sorted(os.sched_getaffinity(0))
+
+
+def parked():
+  for name, expected in (('IsOffroad', b'1'), ('IsEngaged', b'0')):
+    if (Path('/data/params/d') / name).read_bytes().strip() != expected:
+      raise RuntimeError('Heatdeath requires an offroad, disengaged bench device')

tools/heatdeath/report.py

@@ -0,0 +1,77 @@
+"""Summarize actual captured phase timing without granting a health verdict."""
+
+import hashlib
+import json
+from pathlib import Path
+import struct
+
+
+def quantile(values, fraction):
+  return sorted(values)[round((len(values) - 1) * fraction)] if values else None
+
+
+def phase(rows, start, end):
+  selected = sorted((row for row in rows if start <= row[3] < end), key=lambda row: row[0])
+  wall = [(row[4] - row[3]) / 1e6 for row in selected]
+  cpu = [(row[6] - row[5]) / 1e6 for row in selected]
+  gaps = [(b[3] - a[3]) / 1e6 for a, b in zip(selected, selected[1:], strict=False)]
+  return {
+    'calls': len(selected),
+    'observedCallsPerSecond': len(selected) / ((end - start) / 1e9),
+    'wallMsP50': quantile(wall, 0.5),
+    'wallMsP95': quantile(wall, 0.95),
+    'wallMsP99': quantile(wall, 0.99),
+    'wallMsMax': max(wall, default=None),
+    'threadCpuMsP95': quantile(cpu, 0.95),
+    'startGapMsMax': max(gaps, default=None),
+    'raised': sum(row[7] for row in selected),
+  }
+
+
+def summarize(output, skip_seconds=10):
+  output = Path(output)
+  result = json.loads((output / 'result.json').read_text())
+  (start,) = struct.unpack('<Q', (output / 'release.bin').read_bytes())
+  # The feeder exports the same source duration used by the driver.
+  duration = result['sourceDurationNs']
+  end = start + duration
+  start += round(skip_seconds * 1e9)
+  if not 0 <= skip_seconds or start >= end:
+    raise ValueError('Skip must leave a nonempty source window')
+  report = {
+    'sourceHead': result['sourcePins']['sourceHead'],
+    'failure': result['failure'],
+    'everAuthorityReady': result['everAuthorityReady'],
+    'activeObserverSamples': result['activeObserverSamples'],
+    'qualifiedActiveLatencyBench': False,
+    'placement': result['placement'],
+    'skipSeconds': skip_seconds,
+    'syntheticReserve': result['syntheticReserve'],
+    'actors': {},
+    'limits': result['limits'] + ['No actual inference/audio/Android Auto workload is launched. Timing alone does not qualify steering.'],
+  }
+  for name in ('card', 'controlsd', 'selfdrived'):
+    metadata = json.loads((output / f'{name}-timing.json').read_text())
+    if metadata.get('observerErrors'):
+      raise ValueError('Timing observer failed; inspect actor metadata')
+    report['actors'][name] = {}
+    for label, entry in metadata['timing'].items():
+      raw = (output / Path(entry['path']).name).read_bytes()
+      if hashlib.sha256(raw).hexdigest() != entry['sha256']:
+        raise ValueError('Timing capture hash mismatch')
+      rows = list(struct.iter_unpack('<' + 'q' * 20, raw))
+      if entry['rowsDropped'] or entry['rowsRetained'] > len(rows):
+        raise ValueError('Timing ring lost rows; shorten source window')
+      values = phase(rows[: entry['rowsRetained']], start, end)
+      if 'averagePeriodNs' in metadata['sampleColumns'].get(label, []):
+        values['rollingLagFlags'] = sum(row[9] for row in rows[: entry['rowsRetained']] if start <= row[3] < end)
+      report['actors'][name][label] = values
+  reserve = output / 'reserve.json'
+  if result['syntheticReserve'] is not None:
+    proof = json.loads(reserve.read_text())
+    report['reserveCompleted'] = proof['completed']
+    report['reserveMissedPulses'] = proof['missedPulses']
+  native = output / 'native-receipt.json'
+  if native.exists():
+    report['nativeReceipt'] = json.loads(native.read_text())
+  return report

tools/heatdeath/reserve.py

@@ -0,0 +1,58 @@
+"""Explicit synthetic CFS CPU pulses; never presented as actual model workload."""
+
+import argparse
+import json
+import os
+from pathlib import Path
+import struct
+import time
+
+
+def main():
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('--core', type=int, required=True)
+  parser.add_argument('--cpu-ms', type=float, required=True)
+  parser.add_argument('--period-ms', type=float, default=50)
+  parser.add_argument('--duration-ns', type=int, required=True)
+  parser.add_argument('--release', type=Path, required=True)
+  parser.add_argument('--output', type=Path, required=True)
+  args = parser.parse_args()
+  assert 0 < args.cpu_ms < args.period_ms <= 1000
+  os.sched_setscheduler(0, os.SCHED_OTHER, os.sched_param(0))
+  os.sched_setaffinity(0, {args.core})
+  os.setpriority(os.PRIO_PROCESS, 0, 0)
+  deadline = time.monotonic() + 20
+  while not args.release.exists():
+    if time.monotonic() > deadline:
+      raise TimeoutError('No source release')
+    time.sleep(0.005)
+  (start,) = struct.unpack('<Q', args.release.read_bytes())
+  period, dose = round(args.period_ms * 1e6), round(args.cpu_ms * 1e6)
+  rows = []
+  for due in range(start, start + args.duration_ns - period + 1, period):
+    delay = (due - time.monotonic_ns()) / 1e9
+    if delay > 0:
+      time.sleep(delay)
+    begun, cpu = time.monotonic_ns(), time.thread_time_ns()
+    while time.thread_time_ns() - cpu < dose and time.monotonic_ns() < due + period:
+      pass
+    rows.append({'dueNs': due, 'startNs': begun, 'endNs': time.monotonic_ns(), 'cpuNs': time.thread_time_ns() - cpu})
+  with args.output.open('x') as stream:
+    json.dump(
+      {
+        'kind': 'synthetic-cfs-cpu-reserve',
+        'policy': os.sched_getscheduler(0),
+        'nice': os.getpriority(os.PRIO_PROCESS, 0),
+        'core': args.core,
+        'doseNs': dose,
+        'periodNs': period,
+        'rows': rows,
+        'completed': True,
+        'missedPulses': sum(row['cpuNs'] < dose for row in rows),
+      },
+      stream,
+    )
+
+
+if __name__ == '__main__':
+  main()