Drive Lifecycle

Track drive identity and evidence lifecycle so consumers reset state at the same drive boundaries.

openpilot/starpilot/drive_state/control.py

@@ -0,0 +1,47 @@
+"""Local and authenticated presentation of the shared drive-state owner."""
+
+import threading
+import time
+
+from .resolver import Mode
+
+
+LABELS = {Mode.AUTO: 'Auto', Mode.OFFROAD: 'Offroad', Mode.ONROAD: 'Onroad'}
+CYCLE = (Mode.OFFROAD, Mode.ONROAD, Mode.AUTO)
+
+
+class DriveStateControl:
+  def __init__(self, owner, physical, *, effective=None, clock=time.monotonic):
+    self.owner, self.physical, self.effective, self.clock = owner, physical, effective or physical.effective, clock
+    self.cached = None
+    self.lock = threading.Lock()
+
+  def snapshot(self):
+    with self.lock:
+      now = self.clock()
+      if self.cached is None or not 0 <= now - self.cached[0] < 0.5:
+        state = self.owner.snapshot()
+        status = getattr(self.physical, 'status', None)
+        admission = status() if callable(status) else {'allowed': self.physical.allowed(), 'reason': None}
+        self.cached = now, state, bool(admission['allowed']), admission['reason']
+      state = self.cached[1]
+      actual = self.effective()
+      return {'mode': state.mode.value, 'revision': state.revision, 'available': state.available,
+              'effective': 'onroad' if actual is True else 'offroad' if actual is False else None,
+              'overrideAllowed': self.cached[2], 'reason': self.cached[3]}
+
+  def change(self, mode, revision, authorized):
+    # Deliberate local requests may stop driving services while moving; Galaxy rejects Offroad in its action handler.
+    result = self.owner.request(mode, expected_revision=revision, authorized=authorized,
+                                override_allowed=lambda: True)
+    with self.lock:
+      self.cached = None
+    return result
+
+  @staticmethod
+  def next_mode(status):
+    mode = Mode(status['mode'])
+    return CYCLE[(CYCLE.index(mode) + 1) % len(CYCLE)]
+
+  def cycle(self, status, authorized):
+    return self.change(self.next_mode(status).value, status['revision'], authorized)

openpilot/starpilot/drive_state/evidence.py

@@ -0,0 +1,135 @@
+"""Fresh physical vehicle evidence independent of the requested pipeline mode."""
+
+import threading
+import time
+
+from openpilot.starpilot.parked_evidence import DEVICE_TTL_NS, PANDA_TTL_NS, RESUME_SKEW_NS
+
+
+PYTHON_TTL_NS = 300_000_000
+
+
+class PhysicalSource:
+  def __init__(self, messages=None, *, owns_messages=False, mono=time.monotonic_ns,
+               boot=lambda: time.clock_gettime_ns(getattr(time, 'CLOCK_BOOTTIME', time.CLOCK_MONOTONIC))):
+    self.messages, self.owns_messages = messages, owns_messages
+    self.mono, self.boot = mono, boot
+    self.floor = getattr(messages, "after_mono_ns", mono())
+    self.offset = None
+    self.lock = threading.Lock()
+    self.closed = False
+
+  def allowed(self) -> bool:
+    return self.status()['allowed']
+
+  def status(self):
+    def result(allowed, reason=None):
+      return {'allowed': bool(allowed), 'reason': reason}
+    with self.lock:
+      if self.closed:
+        return result(False, 'Vehicle status source is closed')
+      try:
+        if self.messages is None:
+          from openpilot.cereal import messaging
+          self.messages = messaging.SubMaster(['deviceState', 'pandaStates', 'carState', 'selfdriveState'])
+          self.owns_messages = True
+        snapshot = getattr(self.messages, "snapshot", None)
+        sm = snapshot() if callable(snapshot) else self.messages
+        if sm is None:
+          return result(False, 'Current vehicle status is unavailable')
+        self.floor = max(self.floor, getattr(sm, "after_mono_ns", self.floor))
+        if self.owns_messages:
+          sm.update(0)
+        before, boot, now = self.mono(), self.boot(), self.mono()
+        if not 0 <= now - before <= RESUME_SKEW_NS:
+          self.floor, self.offset = now, None
+          return result(False, 'Device clock changed; fresh vehicle status is required')
+        offset = boot - (before + now) // 2
+        if self.offset is not None and abs(offset - self.offset) > RESUME_SKEW_NS:
+          self.floor = now
+        self.offset = offset
+
+        def fresh(service, python=False):
+          stamp = int(sm.logMonoTime[service])
+          receipt = int(sm.recv_time[service] * 1e9)
+          if not all((sm.seen[service], sm.alive[service], sm.valid[service])) or receipt <= self.floor:
+            return False
+          if python and stamp <= self.floor:
+            return False
+          age = now - stamp if python else boot - stamp
+          ttl = PYTHON_TTL_NS if python else PANDA_TTL_NS
+          return stamp > 0 and 0 <= age <= ttl and 0 <= now - receipt <= ttl
+
+        if not fresh('pandaStates'):
+          return result(False, 'Current Panda status is missing, invalid or older than 300 ms')
+        pandas = sm['pandaStates']
+        if len(pandas) == 0 or any(str(p.pandaType) == 'unknown' for p in pandas):
+          return result(False, 'No recognized Panda is reporting vehicle status')
+        ignition = any(p.ignitionLine or p.ignitionCan for p in pandas)
+        if not ignition:
+          safe = all(str(p.safetyModel) == 'noOutput' and not p.controlsAllowed for p in pandas)
+          return result(safe, None if safe else 'Panda must confirm noOutput safety and controls disabled')
+        if not fresh('carState', True):
+          return result(False, 'Current speed and gear are unavailable; turn off the vehicle ignition to confirm parked state')
+        if not fresh('selfdriveState', True):
+          return result(False, 'Driving services cannot confirm disengagement; turn off the vehicle ignition to confirm parked state')
+        car, selfdrive = sm['carState'], sm['selfdriveState']
+        if not car.canValid or car.canTimeout:
+          return result(False, 'Vehicle CAN data is invalid or timed out')
+        if not car.standstill or str(car.gearShifter) != 'park':
+          return result(False, 'Park the vehicle before forcing Offroad or updating')
+        if selfdrive.enabled or selfdrive.active:
+          return result(False, 'Disengage driving controls before forcing Offroad or updating')
+        return result(True)
+      except (OSError, RuntimeError, AttributeError, KeyError, TypeError, ValueError, OverflowError):
+        return result(False, 'Vehicle status could not be read')
+
+  def update_status(self, manager_offroad):
+    if manager_offroad is not True:
+      return {'allowed': False, 'reason': 'Switch the device Offroad before updating'}
+    physical = self.status()
+    if not physical['allowed']:
+      return physical
+    actual = self.effective()
+    if actual is None:
+      return {'allowed': False, 'reason': 'Current device offroad status is missing or older than 1.5 seconds'}
+    if actual:
+      return {'allowed': False, 'reason': 'Manager is still running driving services; wait for the Offroad transition'}
+    return {'allowed': True, 'reason': None}
+
+  def update_ready(self, manager_offroad):
+    return self.update_status(manager_offroad)['allowed']
+
+  def effective(self):
+    with self.lock:
+      if self.closed or self.messages is None or self.offset is None:
+        return None
+      try:
+        before, boot, now = self.mono(), self.boot(), self.mono()
+        offset = boot - (before + now) // 2
+        if not 0 <= now - before <= RESUME_SKEW_NS or abs(offset - self.offset) > RESUME_SKEW_NS:
+          self.floor, self.offset = now, None
+          return None
+        snapshot = getattr(self.messages, "snapshot", None)
+        sm = snapshot() if callable(snapshot) else self.messages
+        if sm is None:
+          return None
+        self.floor = max(self.floor, getattr(sm, "after_mono_ns", self.floor))
+        stamp, receipt = int(sm.logMonoTime['deviceState']), int(sm.recv_time['deviceState'] * 1e9)
+        if (not all((sm.seen['deviceState'], sm.alive['deviceState'], sm.valid['deviceState'])) or
+            stamp <= self.floor or receipt <= self.floor or
+            not 0 <= now - stamp <= DEVICE_TTL_NS or not 0 <= now - receipt <= DEVICE_TTL_NS):
+          return None
+        return bool(sm['deviceState'].started)
+      except (OSError, RuntimeError, AttributeError, KeyError, TypeError, ValueError, OverflowError):
+        return None
+
+  def close(self):
+    with self.lock:
+      self.closed = True
+      if self.owns_messages and self.messages is not None:
+        for socket in getattr(self.messages, 'sock', {}).values():
+          close = getattr(socket, 'close', None)
+          if close is not None:
+            close()
+      self.messages = None

openpilot/starpilot/drive_state/owner.py

@@ -0,0 +1,157 @@
+"""One session-bound request owner shared by local and authenticated controls."""
+
+from collections.abc import Callable
+from contextlib import contextmanager
+from dataclasses import dataclass
+import fcntl
+import os
+from pathlib import Path
+import re
+import stat
+import threading
+import uuid
+
+from openpilot.common.params import UnknownKeyName
+
+from .resolver import Mode
+
+
+SESSION_KEY = 'DriveStateSession'
+REQUEST_KEY = 'DriveStateRequest'
+_HEX = re.compile(r'[0-9a-f]{32}\Z')
+
+
+class Rejected(ValueError):
+  pass
+
+
+@dataclass(frozen=True)
+class State:
+  mode: Mode = Mode.AUTO
+  revision: str | None = None
+  available: bool = False
+
+
+def valid_session(value) -> bool:
+  return (type(value) is dict and set(value) == {'version', 'id', 'pid', 'birth', 'boot'} and
+          type(value['version']) is int and value['version'] == 1 and
+          isinstance(value['id'], str) and _HEX.fullmatch(value['id']) is not None and
+          isinstance(value['boot'], str) and re.fullmatch(r'[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}', value['boot']) is not None and
+          all(type(value[key]) is int and value[key] > 0 for key in ('pid', 'birth')))
+
+
+def request_state(value, session) -> State:
+  if (type(value) is not dict or set(value) != {'version', 'session', 'revision', 'mode'} or
+      type(value['version']) is not int or value['version'] != 1 or value['session'] != session['id'] or
+      not isinstance(value['revision'], str) or _HEX.fullmatch(value['revision']) is None):
+    return State()
+  try:
+    return State(Mode(value['mode']), value['revision'], True)
+  except (TypeError, ValueError):
+    return State()
+
+
+def process_identity(pid: int):
+  try:
+    raw = Path(f"/proc/{pid}/stat").read_bytes()
+    fields = raw[raw.rfind(b')') + 2:].split()
+    return int(fields[19]), fields[0]
+  except (OSError, IndexError, ValueError):
+    return None, None
+
+
+def manager_alive(session) -> bool:
+  try:
+    if Path('/proc/sys/kernel/random/boot_id').read_text().strip() != session['boot']:
+      return False
+  except OSError:
+    return False
+  birth, state = process_identity(session['pid'])
+  return birth == session['birth'] and state not in (None, b'Z', b'X')
+
+
+class DriveStateOwner:
+  def __init__(self, params, root: Path, *, alive: Callable[[dict], bool] = manager_alive):
+    self.params, self.root, self.alive = params, Path(root), alive
+    self._lock = threading.RLock()
+
+  @contextmanager
+  def _exclusive(self):
+    with self._lock:
+      self.root.mkdir(mode=0o700, parents=True, exist_ok=True)
+      info = self.root.lstat()
+      if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o022:
+        raise Rejected('Drive state storage is unavailable')
+      fd = os.open(self.root / '.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
+      try:
+        info = os.fstat(fd)
+        if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077:
+          raise Rejected('Drive state storage is unavailable')
+        try:
+          fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+          raise Rejected("Drive state changed; refresh and try again") from None
+        yield
+      finally:
+        os.close(fd)
+
+  def _read(self):
+    try:
+      session = self.params.get(SESSION_KEY)
+      if not valid_session(session) or not self.alive(session):
+        return None, State()
+      return session, request_state(self.params.get(REQUEST_KEY), session)
+    except (OSError, KeyError, TypeError, ValueError, UnknownKeyName):
+      return None, State()
+
+  def snapshot(self) -> State:
+    return self._read()[1]
+
+  def initialize(self, *, pid: int, birth: int, boot: str) -> State:
+    session = {'version': 1, 'id': uuid.uuid4().hex, 'pid': pid, 'birth': birth, 'boot': boot}
+    if not valid_session(session) or not self.alive(session):
+      raise Rejected('Manager session is unavailable')
+    with self._exclusive():
+      request = self._record(session, Mode.AUTO)
+      # Mismatched generations resolve Auto during either write or a crash.
+      self.params.put(REQUEST_KEY, request, block=True)
+      self.params.put(SESSION_KEY, session, block=True)
+      return request_state(request, session)
+
+  def initialize_manager(self) -> State:
+    pid = os.getpid()
+    boot = Path('/proc/sys/kernel/random/boot_id').read_text().strip()
+    return self.initialize(pid=pid, birth=process_identity(pid)[0], boot=boot)
+
+  @staticmethod
+  def _record(session, mode):
+    return {'version': 1, 'session': session['id'], 'revision': uuid.uuid4().hex, 'mode': mode.value}
+
+  def request(self, mode: str, *, expected_revision: str, authorized: Callable[[], bool],
+              override_allowed: Callable[[], bool]) -> State:
+    try:
+      desired = Mode(mode)
+    except (TypeError, ValueError):
+      raise Rejected('Choose Auto, Offroad or Onroad') from None
+    with self._exclusive():
+      session, current = self._read()
+      if not current.available or current.revision != expected_revision:
+        raise Rejected('Drive state changed; refresh and try again')
+      if not authorized():
+        raise Rejected('Drive state changes are unavailable')
+      if desired == Mode.OFFROAD and not override_allowed():
+        raise Rejected('Park the vehicle and disengage before forcing Offroad')
+      if desired == current.mode:
+        return current
+      # These callbacks recheck the authenticated caller and physical source,
+      # rather than relying on the state that produced the button.
+      if not authorized() or (desired == Mode.OFFROAD and not override_allowed()) or not self.alive(session):
+        raise Rejected('Drive state changes are unavailable')
+      observed_session, observed = self._read()
+      if observed_session != session or observed != current:
+        raise Rejected('Drive state changed; refresh and try again')
+      record = self._record(session, desired)
+      if not authorized() or (desired == Mode.OFFROAD and not override_allowed()):
+        raise Rejected('Drive state changes are unavailable')
+      self.params.put(REQUEST_KEY, record, block=True)
+      return request_state(record, session)

openpilot/starpilot/drive_state/resolver.py

@@ -0,0 +1,25 @@
+"""Pipeline state selection; physical ignition remains separate."""
+
+from enum import StrEnum
+from collections.abc import Mapping
+
+
+class Mode(StrEnum):
+  AUTO = 'auto'
+  OFFROAD = 'offroad'
+  ONROAD = 'onroad'
+
+
+def effective_onroad(mode: Mode, conditions: Mapping[str, bool]) -> bool:
+  if mode == Mode.OFFROAD:
+    return False
+  if mode == Mode.ONROAD:
+    if 'ignition' not in conditions:
+      return False
+    return all(value for key, value in conditions.items() if key != 'ignition')
+  return all(conditions.values())
+
+
+def should_start(mode: Mode, onroad: Mapping[str, bool], startup: Mapping[str, bool], *, already_started: bool) -> bool:
+  ready = effective_onroad(mode, onroad)
+  return ready if already_started else ready and all(startup.values())

openpilot/starpilot/drive_state/tests/test_evidence.py

@@ -0,0 +1,142 @@
+from types import SimpleNamespace as NS
+import pytest
+from typing import Any
+from openpilot.starpilot.drive_state.evidence import PhysicalSource
+
+
+class Messages:
+  def __init__(self):
+    self.values: dict[str, Any] = {
+      'pandaStates': [NS(pandaType='uno', controlsAllowed=False, ignitionLine=False, ignitionCan=False, safetyModel='noOutput')],
+      'carState': NS(canValid=True, canTimeout=False, standstill=True, gearShifter='park'),
+      'selfdriveState': NS(enabled=False, active=False),
+    }
+    keys = self.values
+    self.seen = dict.fromkeys(keys, True)
+    self.alive = dict.fromkeys(keys, True)
+    self.valid = dict.fromkeys(keys, True)
+    self.logMonoTime = dict.fromkeys(keys, 2_000_000_000)
+    self.logMonoTime['pandaStates'] += 10_000_000_000
+    self.recv_time = dict.fromkeys(keys, 2.0)
+
+  def __getitem__(self, key):
+    return self.values[key]
+
+
+@pytest.fixture
+def fixture():
+  now = [1_000_000_000, 10_000_000_000]
+  messages = Messages()
+  source = PhysicalSource(messages, mono=lambda: now[0], boot=lambda: sum(now))
+  now[0] = 2_100_000_000
+  return NS(now=now, sm=messages, source=source)
+
+
+def test_genuine_ignition_off_needs_all_panda_no_output_not_pipeline_state(fixture):
+  f = fixture
+  assert f.source.allowed()
+  f.sm.values['deviceState'] = NS(started=True)
+  assert f.source.allowed()
+  f.sm.values['pandaStates'][0].safetyModel = 'toyota'
+  assert not f.source.allowed()
+
+
+@pytest.mark.parametrize('field,value', [('controlsAllowed', True), ('pandaType', 'unknown')])
+def test_unknown_or_actuation_authority_never_admits(fixture, field, value):
+  setattr(fixture.sm.values['pandaStates'][0], field, value)
+  assert not fixture.source.allowed()
+
+
+def test_live_can_park_disabled_can_admit_with_ignition_on_and_started(fixture):
+  f = fixture
+  f.sm.values['pandaStates'][0].ignitionLine = True
+  f.sm.values['pandaStates'][0].safetyModel = 'toyota'
+  f.sm.values['deviceState'] = NS(started=True)
+  assert f.source.allowed()
+  for message, field, value in [
+    ('carState', 'canValid', False),
+    ('carState', 'canTimeout', True),
+    ('carState', 'standstill', False),
+    ('carState', 'gearShifter', 'drive'),
+    ('selfdriveState', 'enabled', True),
+    ('selfdriveState', 'active', True),
+  ]:
+    before = getattr(f.sm.values[message], field)
+    setattr(f.sm.values[message], field, value)
+    assert not f.source.allowed()
+    setattr(f.sm.values[message], field, before)
+
+
+def test_stale_forced_offroad_cannot_supply_park_authority(fixture):
+  f = fixture
+  f.sm.values['pandaStates'][0].ignitionLine = True
+  f.sm.logMonoTime['carState'] = 1_000_000_000
+  f.sm.values['deviceState'] = NS(started=False)
+  assert not f.source.allowed()
+
+
+@pytest.mark.parametrize('service', ['pandaStates', 'carState', 'selfdriveState'])
+def test_missing_invalid_stale_future_sources_fail_closed(fixture, service):
+  f = fixture
+  f.sm.values['pandaStates'][0].ignitionLine = True
+  for table, value in [(f.sm.valid, False), (f.sm.alive, False), (f.sm.seen, False), (f.sm.logMonoTime, 0), (f.sm.recv_time, 100.0)]:
+    before = table[service]
+    table[service] = value
+    assert not f.source.allowed()
+    table[service] = before
+
+
+def test_resume_and_collector_start_need_post_boundary_evidence(fixture):
+  f = fixture
+  f.sm.values['pandaStates'][0].ignitionLine = True
+  assert f.source.allowed()
+  f.now[1] += 20_000_000_000
+  assert not f.source.allowed()
+  f.sm.logMonoTime['pandaStates'] = sum(f.now)
+  f.sm.recv_time['pandaStates'] = f.now[0] / 1e9
+  assert not f.source.allowed()
+  f.now[0] += 10_000_000
+  for key in f.sm.values:
+    if key == 'deviceState':
+      continue
+    f.sm.logMonoTime[key] = sum(f.now) if key == 'pandaStates' else f.now[0]
+    f.sm.recv_time[key] = f.now[0] / 1e9
+  assert f.source.allowed()
+  f.source.close()
+  assert not f.source.allowed()
+
+
+def test_real_capnp_physical_fields_and_fresh_effective_state(fixture):
+  from openpilot.cereal import log
+
+  f = fixture
+  panda_event = log.Event.new_message()
+  pandas = panda_event.init('pandaStates', 1)
+  pandas[0].pandaType = 'uno'
+  pandas[0].safetyModel = 'noOutput'
+  f.sm.values['pandaStates'] = panda_event.as_reader().pandaStates
+  assert f.source.allowed()
+  device = log.Event.new_message().init('deviceState')
+  device.started = True
+  f.sm.values['deviceState'] = device.as_reader()
+  for table, value in ((f.sm.seen, True), (f.sm.alive, True), (f.sm.valid, True), (f.sm.logMonoTime, 2_000_000_000), (f.sm.recv_time, 2.0)):
+    table['deviceState'] = value
+  assert f.source.effective() is True
+  f.sm.logMonoTime['deviceState'] = 1
+  assert f.source.effective() is None
+  f.now[1] += 10_000_000_000
+  assert f.source.effective() is None
+
+
+def test_park_disengaged_does_not_confuse_native_permission_with_active_controls(fixture):
+  f = fixture
+  f.sm.values['pandaStates'][0].ignitionLine = True
+  f.sm.values['pandaStates'][0].safetyModel = 'hyundaiCanfd'
+  f.sm.values['pandaStates'][0].controlsAllowed = True
+  f.sm.values['deviceState'] = NS(started=True)
+  assert f.source.allowed()
+  f.sm.values['selfdriveState'].active = True
+  assert not f.source.allowed()
+  f.sm.values['selfdriveState'].active = False
+  f.sm.values['carState'].gearShifter = 'drive'
+  assert not f.source.allowed()

openpilot/starpilot/drive_state/tests/test_owner.py

@@ -0,0 +1,288 @@
+"""Temporary typed-store fakes exercise the shared request boundary."""
+
+import copy
+from types import SimpleNamespace as NS
+
+import pytest
+
+from openpilot.starpilot.drive_state.owner import DriveStateOwner, Mode, Rejected, REQUEST_KEY, SESSION_KEY
+
+BOOT = '00000000-0000-0000-0000-000000000001'
+
+
+class Params:
+  def __init__(self):
+    self.values = {}
+    self.writes = []
+
+  def get(self, key):
+    return copy.deepcopy(self.values.get(key))
+
+  def put(self, key, value, block=False):
+    assert block
+    self.values[key] = copy.deepcopy(value)
+    self.writes.append((key, copy.deepcopy(value)))
+
+
+@pytest.fixture
+def fixture(tmp_path):
+  params = Params()
+  live = [True]
+  owner = DriveStateOwner(params, tmp_path / 'owner', alive=lambda _: live[0])
+  owner.initialize(pid=1, birth=2, boot=BOOT)
+  return NS(params=params, live=live, owner=owner)
+
+
+def request(f, mode, **kwargs):
+  return f.owner.request(mode, expected_revision=f.owner.snapshot().revision,
+                         authorized=kwargs.get('authorized', lambda: True),
+                         override_allowed=kwargs.get('override_allowed', lambda: True))
+
+
+def test_shared_owner_idempotence_atomic_mode_and_manager_reset(fixture):
+  f = fixture
+  state = request(f, 'onroad')
+  writes = len(f.params.writes)
+  assert request(f, 'onroad') == state and len(f.params.writes) == writes
+  assert request(f, 'offroad').mode == Mode.OFFROAD
+  assert set(f.params.values) == {REQUEST_KEY, SESSION_KEY}
+  old_revision = f.owner.snapshot().revision
+  f.owner.initialize(pid=1, birth=3, boot=BOOT)
+  assert f.owner.snapshot().mode == Mode.AUTO
+  with pytest.raises(Rejected):
+    f.owner.request('onroad', expected_revision=old_revision, authorized=lambda: True, override_allowed=lambda: True)
+
+
+def test_only_deliberate_override_requires_physical_admission(fixture):
+  f = fixture
+  with pytest.raises(Rejected):
+    request(f, 'offroad', override_allowed=lambda: False)
+  request(f, 'onroad')
+  assert request(f, 'auto', override_allowed=lambda: False).mode == Mode.AUTO
+  assert not any(key in f.params.values for key in ('ForceOffroad', 'ForceOnroad'))
+
+
+@pytest.mark.parametrize('change', ['permission', 'physical', 'manager'])
+def test_revocation_before_commit_never_writes_request(fixture, change):
+  f = fixture
+  writes = len(f.params.writes)
+  calls = [0]
+
+  def checked():
+    calls[0] += 1
+    if calls[0] == 2 and change == 'manager':
+      f.live[0] = False
+    return calls[0] < 2
+
+  with pytest.raises(Rejected):
+    request(f, 'offroad', authorized=checked if change == 'permission' else lambda: True,
+            override_allowed=checked if change in ('physical', 'manager') else lambda: True)
+  assert len(f.params.writes) == writes
+
+
+@pytest.mark.parametrize('damage', ['missing', 'invalid-mode', 'extra-key', 'wrong-generation', 'boolean-version', 'dead'])
+def test_bad_or_old_requests_never_grant_override(fixture, damage):
+  f = fixture
+  request(f, 'onroad')
+  if damage == 'missing':
+    f.params.values.pop(REQUEST_KEY)
+  elif damage == 'invalid-mode':
+    f.params.values[REQUEST_KEY]['mode'] = 'both'
+  elif damage == 'extra-key':
+    f.params.values[REQUEST_KEY]['ForceOnroad'] = True
+  elif damage == 'wrong-generation':
+    f.params.values[REQUEST_KEY]['session'] = '0' * 32
+  elif damage == 'boolean-version':
+    f.params.values[REQUEST_KEY]['version'] = True
+  else:
+    f.live[0] = False
+  assert f.owner.snapshot().mode == Mode.AUTO
+  assert not f.owner.snapshot().available
+
+
+def test_two_key_initialization_crash_defaults_auto(fixture):
+  f = fixture
+  request(f, 'onroad')
+  old_session = f.params.get(SESSION_KEY)
+  original_put = f.params.put
+
+  def interrupted(key, value, block=False):
+    if key == SESSION_KEY:
+      raise OSError('interrupted manager initialization')
+    original_put(key, value, block=block)
+
+  f.params.put = interrupted
+  with pytest.raises(OSError):
+    f.owner.initialize(pid=1, birth=3, boot=BOOT)
+  assert f.params.get(SESSION_KEY) == old_session
+  assert f.owner.snapshot().mode == Mode.AUTO and not f.owner.snapshot().available
+
+
+def test_session_storage_symlink_is_rejected(fixture, tmp_path):
+  f = fixture
+  target = tmp_path / 'elsewhere'
+  target.mkdir()
+  alias = tmp_path / 'alias'
+  alias.symlink_to(target)
+  owner = DriveStateOwner(f.params, alias, alive=lambda _: True)
+  with pytest.raises(Rejected):
+    owner.request('onroad', expected_revision=f.owner.snapshot().revision,
+                  authorized=lambda: True, override_allowed=lambda: True)
+
+
+class FileParams:
+  def __init__(self, root):
+    self.root = root
+
+  def get(self, key):
+    import json
+    try:
+      return json.loads((self.root / key).read_text())
+    except FileNotFoundError:
+      return None
+
+  def put(self, key, value, block=False):
+    import json
+    import os
+    import uuid
+    assert block
+    temporary = self.root / ('.put-' + uuid.uuid4().hex)
+    temporary.write_text(json.dumps(value))
+    os.replace(temporary, self.root / key)
+
+
+
+PROCESS_PROGRAM = """
+import json,sys,time,types
+from pathlib import Path
+sys.path.insert(0,sys.argv[2])
+sys.path.insert(0,sys.argv[1])
+import openpilot.starpilot
+pkg=types.ModuleType('openpilot.starpilot.drive_state')
+pkg.__path__=[str(Path(sys.argv[1])/'openpilot/starpilot/drive_state')]
+sys.modules[pkg.__name__]=pkg
+from openpilot.starpilot.drive_state.owner import DriveStateOwner, Rejected
+# FILE_PARAMS
+params_root,lock_root,revision,mode,ready,release,result,started,go=sys.argv[3:]
+owner=DriveStateOwner(FileParams(Path(params_root)),Path(lock_root),alive=lambda _: True)
+if started != '-':
+  Path(started).touch()
+  startup_deadline=time.monotonic()+15
+  while not Path(go).exists():
+    if time.monotonic()>startup_deadline:raise RuntimeError('request barrier timeout')
+    time.sleep(.01)
+def authorized():
+  if ready != '-':
+    Path(ready).touch()
+    deadline=time.monotonic()+30
+    while not Path(release).exists():
+      if time.monotonic()>deadline:raise RuntimeError('fixture timeout')
+      time.sleep(.01)
+  return True
+try:
+  state=owner.request(mode,expected_revision=revision,authorized=authorized,override_allowed=lambda: True)
+  output=['accepted',state.mode.value]
+except Rejected:
+  output=['rejected',mode]
+Path(result).write_text(json.dumps(output))
+"""
+
+
+def test_crossprocess_revision_commit_and_nonblocking_hardware_snapshot(tmp_path):
+  import json
+  import subprocess
+  import sys
+  import inspect
+  import time
+  from pathlib import Path
+  params_root = tmp_path / 'params'
+  params_root.mkdir()
+  lock_root = tmp_path / 'owner'
+  owner = DriveStateOwner(FileParams(params_root), lock_root, alive=lambda _: True)
+  state = owner.initialize(pid=1, birth=2, boot=BOOT)
+  ready, release = tmp_path / 'ready', tmp_path / 'release'
+  source = Path(__file__).parents[4]
+  from openpilot.common import params as native_params
+  native_source = Path(native_params.__file__).parents[2]
+  result1, result2 = tmp_path / 'result1', tmp_path / 'result2'
+  program = PROCESS_PROGRAM.replace('# FILE_PARAMS', inspect.getsource(FileParams))
+  common = [sys.executable, '-c', program, str(source), str(native_source), str(params_root), str(lock_root), state.revision]
+  assert state.revision is not None
+  common = [str(value) for value in common]
+  first = subprocess.Popen(common + ['onroad', str(ready), str(release), str(result1), '-', '-'])
+  second = None
+  second_ready, second_go = tmp_path / 'second-ready', tmp_path / 'second-go'
+  try:
+    deadline = time.monotonic() + 15
+    while not ready.exists() and time.monotonic() < deadline:
+      time.sleep(.01)
+    assert ready.exists()
+    # Hardware reads do not join a UI/auth request holding the writer lock.
+    assert owner.snapshot() == state
+    assert not release.exists()
+    second = subprocess.Popen(common + ['offroad', '-', str(release), str(result2), str(second_ready), str(second_go)])
+    # Interpreter/native imports have their own startup budget, separate from
+    # the writer latency proof. The holder is never released by a timer.
+    deadline = time.monotonic() + 15
+    while not second_ready.exists() and time.monotonic() < deadline:
+      time.sleep(.01)
+    assert second_ready.exists() and not release.exists()
+    second_go.touch()
+    assert second.wait(timeout=2) == 0
+    assert not release.exists()  # A competing UI request fails without waiting on the holder.
+    release.touch()
+    assert first.wait(timeout=5) == 0
+    assert sorted((json.loads(result1.read_text()), json.loads(result2.read_text()))) == [
+      ['accepted', 'onroad'], ['rejected', 'offroad']]
+    assert owner.snapshot().mode == Mode.ONROAD
+  finally:
+    release.touch()
+    for process in (first, second):
+      if process is not None and process.poll() is None:
+        process.kill()
+        process.wait(timeout=3)
+
+
+def test_process_identity_and_boot_liveness_use_single_stat_read(monkeypatch):
+  from pathlib import Path
+  from openpilot.starpilot.drive_state.owner import process_identity, manager_alive
+  reads = []
+  # Linux field 22 follows comm (which may contain spaces and closing parentheses).
+  fields = [b'S'] + [b'0'] * 18 + [b'12345']
+  raw = b'42 (manager ) worker) ' + b' '.join(fields)
+  def read_bytes(path):
+    reads.append(str(path))
+    return raw
+  monkeypatch.setattr(Path, 'read_bytes', read_bytes)
+  monkeypatch.setattr(Path, 'read_text', lambda path: BOOT)
+  session = {'pid': 42, 'birth': 12345, 'boot': BOOT}
+  assert manager_alive(session)
+  assert reads == ['/proc/42/stat']
+  assert not manager_alive(dict(session, birth=12346))
+  assert not manager_alive(dict(session, boot='different-boot'))
+  raw = raw.replace(b') S ', b') Z ')
+  assert not manager_alive(session)
+  raw = b'malformed'
+  assert process_identity(42) == (None, None)
+
+
+def test_empty_real_native_registry_resolves_auto_and_unsupported_key_is_explicit(tmp_path):
+  from openpilot.common.params import Params as NativeParams, UnknownKeyName
+  from openpilot.starpilot.drive_state.owner import State
+  params = NativeParams(str(tmp_path / 'params'))
+  owner = DriveStateOwner(params, tmp_path / 'owner')
+  assert owner.snapshot() == State()
+  class Unsupported:
+    def get(self, key):
+      raise UnknownKeyName(key)
+  assert DriveStateOwner(Unsupported(), tmp_path / 'unsupported').snapshot() == State()
+
+
+def test_force_onroad_requires_no_park_proof_but_offroad_does(fixture):
+  f = fixture
+  assert request(f, 'onroad', override_allowed=lambda: False).mode == Mode.ONROAD
+  with pytest.raises(Rejected):
+    request(f, 'offroad', override_allowed=lambda: False)
+  with pytest.raises(Rejected):
+    request(f, 'auto', authorized=lambda: False, override_allowed=lambda: False)
+  assert f.owner.snapshot().mode == Mode.ONROAD