Jetlink API 3 adapter

StarPilotAdapter subclasses jetlink's Stock: StarPilot's param names, the
AOL engagement rule, the model root and a CatalogModelState history reset
that keeps its JIT buffers. jetlinkd runs jetlink's owner as it is, modeld
calls the hooks, and Galaxy and analytics read jetlink's runtime record in
place of JetlinkRuntime.

- no big-model slot or catalog keys: jetlink provisions its default model
- jetlink restores the owner's tuning itself; the port draw stays cut with
  a host plugged in
- the Jetson is asked to power off on DoShutdown only, capped at 30 s

openpilot/selfdrive/modeld/modeld.py

@@ -45,8 +45,8 @@ from openpilot.starpilot.models.status import ModelVariant
 from openpilot.starpilot.models.startup import report_load_timeout, wait_for_chestnut_power
 from openpilot.starpilot.navigation.intent import TurnIntent, matching_turn_signal
 from openpilot.starpilot.models.catalog import BUNDLED_CURRENT, BY_ID, DEFAULT_SMALL, DEFAULT_SMALL_SHA256
-from openpilot.starpilot.models.runner import CatalogModelState, action_from_outputs, load_verified_model
-from openpilot.starpilot.models.jetlink_adapter import JetlinkModel, attach as attach_jetlink
+from openpilot.starpilot.models.runner import CatalogModelState, load_verified_model
+from openpilot.starpilot.models import jetlink_adapter
 import uuid
 from openpilot.selfdrive.controls.lib.drive_helpers import get_accel_from_plan, should_stop, smooth_value, get_curvature_from_plan
 from openpilot.selfdrive.modeld.parse_model_outputs import Parser
@@ -159,6 +159,7 @@ def input_view(buffer: Buffer, shape: tuple[int, ...], dtype: DType, offset: int
 
 class ModelState:
   prev_desire: np.ndarray  # for tracking the rising edge of the pulse
+  get_action_from_model = staticmethod(get_action_from_model)
 
   def __init__(self, cam_w: int, cam_h: int, chestnut: bool):
     if not chestnut:
@@ -388,8 +389,13 @@ def main(demo=False):
                        "chestnut-load-failed" if initial_chestnut_fallback else
                        "selected-load-failed" if (selected_small_failed and not model.chestnut) or requested_model_id != model.model_id else None,
                        model_id=model.model_id)
-  model = attach_jetlink(model, vipc_client_main.width, vipc_client_main.height,
-                         chestnut=CHESTNUT, recovery=recovery_small_only)
+  HOOKS = jetlink_adapter.hooks()
+  jetlink_model = None
+  # before going realtime: the GPU thread it starts would inherit SCHED_FIFO 54 on core 7
+  if HOOKS.prepare(CHESTNUT or recovery_small_only):
+    # the joining model drives as small_model until the big model is ready
+    model = HOOKS.attach(model, small_model, vipc_client_main.width, vipc_client_main.height)
+    jetlink_model = model if model is not small_model else None
   params.put_bool("ChestnutLoading", False)
   cloudlog.warning(f"models loaded in {time.monotonic() - st:.1f}s, modeld starting")
   release_compile_pool()
@@ -517,14 +523,13 @@ def main(demo=False):
       'action_t': np.array([lat_action_t, long_action_t], dtype=np.float32),
     }
 
-    if isinstance(model, (CatalogModelState, JetlinkModel)):
+    if isinstance(model, CatalogModelState) or model is jetlink_model:
       inputs['prev_action'] = np.array([prev_action.desiredCurvature * max(1.0, v_ego) ** 2,
                                         prev_action.desiredAcceleration], dtype=np.float32)
       inputs['lateral_control_params'] = np.array([v_ego, lat_delay], dtype=np.float32)
 
-    jetlink_handovers = model.handovers if isinstance(model, JetlinkModel) else None
-    if isinstance(model, JetlinkModel):
-      model.observe_frame(lat_delay, frame_drop_ratio)
+    model.lat_delay = lat_delay
+    model.frame_drop_ratio = frame_drop_ratio
     mt1 = time.perf_counter()
     try:
       send_chestnut = (chestnut_state is not None and
@@ -535,7 +540,7 @@ def main(demo=False):
         raise
       # fallback to small model
       fallback_reason = "chestnut-load-failed"
-      if isinstance(model, JetlinkModel):
+      if model is jetlink_model:
         model.close()
         fallback_reason = None
       else:
@@ -550,27 +555,21 @@ def main(demo=False):
       model_output = None
     mt2 = time.perf_counter()
     model_execution_time = mt2 - mt1
-    if isinstance(model, JetlinkModel) and model.handovers != jetlink_handovers:
+    if model is jetlink_model and HOOKS.handed_over(model):
       frame_dropped_filter.x = 0.
       frame_drop_ratio = 0.
       if model.chestnut:
         receipt_owner.loaded(None, ModelVariant.CHESTNUT)
       else:
         receipt_owner.loaded(small_prepared, ModelVariant.SMALL, model_id=small_model.model_id)
-      cloudlog.event("jetlinkHandover", state=model.big_model_state, handovers=model.handovers)
+      cloudlog.event("jetlinkHandover", state=HOOKS.state(model), handovers=model.handovers)
 
     if model_output is not None:
       modelv2_send = messaging.new_message('modelV2')
       drivingdata_send = messaging.new_message('drivingModelData')
       posenet_send = messaging.new_message('cameraOdometry')
 
-      if isinstance(model, JetlinkModel):
-        action = model.action(model_output, prev_action, lat_action_t, long_action_t, v_ego)
-      elif isinstance(model, CatalogModelState):
-        action = action_from_outputs(model_output, model.behavior_version, prev_action, lat_action_t, long_action_t, v_ego,
-                                     LAT_SMOOTH_SECONDS, LONG_SMOOTH_SECONDS)
-      else:
-        action = get_action_from_model(model_output, prev_action, lat_action_t, long_action_t, v_ego)
+      action = model.get_action_from_model(model_output, prev_action, lat_action_t, long_action_t, v_ego)
       prev_action = action
       fill_model_msg(modelv2_send, model_output, action,
                      publish_state, meta_main.frame_id, meta_extra.frame_id, frame_id,

openpilot/starpilot/analytics/runtime.py

@@ -17,13 +17,12 @@ COUNTERS = ('drives', 'meters', 'seconds', 'current_months_meters', 'total_track
 
 
 def executed_model(sm, params, now):
-  from openpilot.starpilot.models.jetlink_adapter import runtime_status
+  from openpilot.starpilot.models.jetlink_adapter import hooks
   from openpilot.starpilot.models.runtime import snapshot
   from openpilot.starpilot.models.status import ModelHealth
   from openpilot.starpilot.models.catalog import model_entries
-  remote = runtime_status()
-  if remote.get('active') and bool(sm['modelV2'].big):
-    digest = remote['artifact_sha256']
+  if bool(sm['modelV2'].big) and (remote := hooks().runtime()) is not None and remote.active and remote.sha256:
+    digest = remote.sha256
     return {'driving_model': 'jetlink:' + digest[:16], 'metrics': {'model_sha256': digest, 'model_variant': 'jetlink'}}
   status = snapshot(sm, params, now)
   entries = model_entries()

openpilot/starpilot/analytics/tests/test_runtime.py

@@ -4,6 +4,11 @@ from unittest.mock import Mock
 from openpilot.starpilot.analytics.runtime import AnalyticsWorker
 
 
+def jetlink(record):
+  """Jetlink's hooks with runtime() answering `record`."""
+  return SimpleNamespace(runtime=lambda: record)
+
+
 class Source:
   def __init__(self):
     self.data = {'deviceState': SimpleNamespace(started=False), 'carState': SimpleNamespace(vEgo=10),
@@ -168,15 +173,15 @@ def test_executed_model_reports_verified_fallback_not_saved_request():
   source = Source()
   status = SimpleNamespace(health=ModelHealth.ACTIVE, loaded_id='rdf43', artifact_sha256='a' * 64,
                            variant=ModelVariant.SMALL, requested_id='some-selected-large-model')
-  with patch('openpilot.starpilot.models.jetlink_adapter.runtime_status', return_value={'active': False}), \
+  with patch('openpilot.starpilot.models.jetlink_adapter.hooks', return_value=jetlink(None)), \
        patch('openpilot.starpilot.models.runtime.snapshot', return_value=status):
     result = executed_model(source, Mock(), 1_000_000_000)
   assert result['driving_model'] == BY_ID['rdf43'].name
   assert result['metrics']['model_variant'] == 'small'
   assert 'some-selected-large-model' not in str(result)
   source.data['modelV2'].big = True
-  with patch('openpilot.starpilot.models.jetlink_adapter.runtime_status',
-             return_value={'active': True, 'artifact_sha256': 'b' * 64}):
+  with patch('openpilot.starpilot.models.jetlink_adapter.hooks',
+             return_value=jetlink(SimpleNamespace(active=True, sha256='b' * 64))):
     remote = executed_model(source, Mock(), 1_000_000_000)
   assert remote['metrics']['model_sha256'] == 'b' * 64
   assert remote['metrics']['model_variant'] == 'jetlink'
@@ -228,7 +233,7 @@ def test_executed_model_real_message_catalog_identity_and_actual_small_fallback(
   source.data['modelV2'] = messaging.new_message('modelV2').modelV2
   status = ModelStatus('selected-large', 'rdf43', ModelVariant.SMALL, ModelHealth.ACTIVE, True,
                        'large model unavailable', 'a' * 64)
-  with patch('openpilot.starpilot.models.jetlink_adapter.runtime_status', return_value={'active': False}), \
+  with patch('openpilot.starpilot.models.jetlink_adapter.hooks', return_value=jetlink(None)), \
        patch('openpilot.starpilot.models.runtime.snapshot', return_value=status) as projection:
     params = Mock()
     result = executed_model(source, params, 1_000_000_000)
@@ -247,8 +252,8 @@ def test_executed_model_remote_requires_actual_big_output_not_remote_selection()
   source = Source()
   source.data['modelV2'] = messaging.new_message('modelV2').modelV2
   small = ModelStatus('selected-large', 'rdf43', ModelVariant.SMALL, ModelHealth.ACTIVE, False, None, 'a' * 64)
-  remote = {'active': True, 'artifact_sha256': 'b' * 64}
-  with patch('openpilot.starpilot.models.jetlink_adapter.runtime_status', return_value=remote), \
+  remote = SimpleNamespace(active=True, sha256='b' * 64)
+  with patch('openpilot.starpilot.models.jetlink_adapter.hooks', return_value=jetlink(remote)), \
        patch('openpilot.starpilot.models.runtime.snapshot', return_value=small) as projection:
     source.data['modelV2'].big = True
     big = executed_model(source, Mock(), 1_000_000_000)
@@ -271,6 +276,6 @@ def test_executed_model_unverified_receipt_never_reports_requested_identity():
   for health in (ModelHealth.UNAVAILABLE, ModelHealth.IDENTITY_UNAVAILABLE, ModelHealth.LOADING,
                  ModelHealth.STALE, ModelHealth.FAILED):
     status = ModelStatus('selected-large', None, None, health, True, 'unverified', None)
-    with patch('openpilot.starpilot.models.jetlink_adapter.runtime_status', return_value={'active': False}), \
+    with patch('openpilot.starpilot.models.jetlink_adapter.hooks', return_value=jetlink(None)), \
          patch('openpilot.starpilot.models.runtime.snapshot', return_value=status):
       assert executed_model(source, Mock(), 1_000_000_000) == {}

openpilot/starpilot/models/jetlink_adapter.py

@@ -1,333 +1,165 @@
-"""StarPilot's boundary to the pinned Jetlink public API."""
+"""StarPilot's adapter to jetlink's openpilot layer, and the guard every Jetlink hook line calls.
+
+The top level imports the standard library only: manager builds its process list from this module and the
+gadget owner (jetlinkd) runs it.
+"""
 from __future__ import annotations
 
-from functools import cached_property, lru_cache
-import logging
+import functools
 import os
 import time
-import threading
-import uuid
-from pathlib import Path
+
+API = 3  # the jetlink.openpilot API this adapter is written to
+UNSUPPORTED = "Jetlink requires a comma 3X or comma 4"
+# AOL steers on its own axes, and a paused AOL is still armed: only its intent says so
+AOL_SERVICES = ("aolIntentWire", "aolAxisState")
+AOL_AXES = ("desiredLateral", "desiredLongitudinal", "lateralActive", "longitudinalActive")
+# how long after AOL's reader is built an intent not yet heard counts as armed: its first polls run before the
+# subscription connects, though card publishes the intent at 100 Hz
+AOL_GRACE_S = 1.
+# CatalogModelState's per-frame camera input, not history
+FRAME_INPUTS = ("mapped_gpu_image",)
 
 
-ADAPTER_MODULE = "openpilot.starpilot.models.jetlink_adapter"
-BASEDIR = Path(__file__).resolve().parents[3]
-API_VERSION = 1
-
-
-def supported_device(device_type=None):
+def supported_device(device_type=None) -> bool:
   if device_type is None:
     from openpilot.common.hardware import HARDWARE
     device_type = HARDWARE.get_device_type()
   return device_type in ("tizi", "mici")
 
 
-def params_directory() -> Path:
-  # Match Path::params() and Params' prefix without loading native libraries.
-  home = Path.home() / (".comma" + os.environ.get("OPENPILOT_PREFIX", ""))
-  root = os.environ.get("PARAMS_ROOT", "/data/params" if Path("/AGNOS").is_file() else str(home / "params"))
-  return Path(root) / os.environ.get("OPENPILOT_PREFIX", "d")
+def aol_in_control(intent, alive, axes, waiting=False) -> bool:
+  """AOL armed (paused or not) or on its axes: the half of the swap's in-control rule that is not openpilot's,
+  shared by engaged() and selfdrived. `intent` is aolIntentWire's message, None until heard; `alive` that it still
+  is; `axes` the AOL_AXES flags, empty when unknown; `waiting` that its reader is younger than AOL_GRACE_S, when an
+  intent not heard yet may be armed."""
+  if intent is None:
+    if waiting:
+      return True
+  elif not alive:
+    return True
+  else:
+    from openpilot.starpilot.aol.wire import decode_intent
+    decoded = decode_intent(intent)
+    if decoded is None or decoded.allowedLatch or decoded.lateralArmed:
+      return True
+  return any(axes)
 
 
-def keys():
-  from jetlink.openpilot import Keys
-  return Keys(link="JetlinkMode", offroad="IsOffroad", progress="JetlinkProgress", spec="JetlinkSpec",
-              pointers="JetlinkPointers", big_model="JetlinkBigModel", catalog="JetlinkCatalog",
-              charge_phone="JetlinkChargePhone")
+@functools.cache
+def _adapter_class():
+  from jetlink.openpilot.interface import Keys
+  from jetlink.openpilot.stock import Stock
+
+  class StarPilotAdapter(Stock):
+    # No big-model slot or catalog: jetlink provisions its default model
+    KEYS = Keys(link="JetlinkMode", offroad="IsOffroad", progress="JetlinkProgress", spec="JetlinkSpec",
+                pointers="JetlinkPointers", charge_phone="JetlinkChargePhone")
+    IN_CONTROL_SERVICES = ("carControl", "selfdriveState")
+    _aol = None
+    _aol_since = 0.
+
+    def model_root(self):
+      from openpilot.starpilot.models.manager import ROOT
+      return ROOT
+
+    def engaged(self, sm) -> bool:
+      # openpilot, or aol_in_control(); Stock.in_control() answers True if this raises. AOL's reader is built and
+      # polled on every call, openpilot engaged or not, so it has heard AOL by the time an offer or a fall asks,
+      # rather than answering from its grace
+      if self._aol is None:
+        from openpilot.cereal import messaging
+        self._aol, self._aol_since = messaging.SubMaster(list(AOL_SERVICES)), time.monotonic()
+      aol = self._aol
+      aol.update(0)
+      if super().engaged(sm):
+        return True
+      heard = aol.seen["aolIntentWire"]
+      axes = aol["aolAxisState"] if aol.seen["aolAxisState"] and aol.alive["aolAxisState"] else None
+      return aol_in_control(aol["aolIntentWire"] if heard else None, aol.alive["aolIntentWire"],
+                            () if axes is None else [getattr(axes, axis) for axis in AOL_AXES],
+                            waiting=time.monotonic() - self._aol_since < AOL_GRACE_S)
+
+    def small_reset(self, small):
+      # CatalogModelState's history zeroed in place: GPU queues, numpy inputs (prev_feat, packed inputs) and
+      # the desire edge. Never _reset_state(), which makes tensors the run JIT was not captured with
+      from openpilot.starpilot.models.runner import CatalogModelState
+      if not isinstance(small, CatalogModelState):
+        return None
+      from jetlink.openpilot.warp import reset_queues
+      queues = [q for name, q in small.input_queues.items() if q.device != "NPY" and name not in FRAME_INPUTS]
+      return reset_queues(queues, small.npy.values(), small.prev_desire)
+
+  return StarPilotAdapter
 
 
-def process_start_ticks(pid):
-  try:
-    return int(Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[19])
-  except (OSError, ValueError, IndexError):
-    return None
-
-
-def runtime_snapshot(record, now_ns, boot_id, start_ticks):
-  unavailable = {"state": "unavailable", "active": False}
-  if not isinstance(record, dict) or type(record.get("version")) is not int or record["version"] != 1:
-    return unavailable
-  mono, output = record.get("mono_ns"), record.get("remote_output_mono_ns")
-  session = record.get("session")
-  if (not boot_id or record.get("boot_id") != boot_id or
-      not isinstance(session, str) or len(session) != 32 or any(c not in "0123456789abcdef" for c in session) or
-      type(record.get("pid")) is not int or record["pid"] <= 0 or type(start_ticks) is not int or start_ticks <= 0 or
-      type(record.get("process_start_ticks")) is not int or
-      record.get("process_start_ticks") != start_ticks or type(mono) is not int or
-      not 0 < mono <= now_ns or now_ns - mono > 1_500_000_000 or type(output) is not int or
-      not 0 <= output <= mono or record.get("state") not in
-      ("none", "joining", "retrying", "ready", "running", "unavailable")):
-    return unavailable
-  digest = record.get("artifact_sha256")
-  identity = (isinstance(digest, str) and len(digest) == 64 and all(c in "0123456789abcdef" for c in digest))
-  active = record["state"] == "running" and identity and output > 0 and mono - output <= 250_000_000
-  snapshot = {"state": record["state"], "active": active}
-  if active:
-    snapshot["artifact_sha256"] = digest
-    checkpoint = record.get("model_checkpoint")
-    if isinstance(checkpoint, str) and len(checkpoint) <= 256:
-      snapshot["model_checkpoint"] = checkpoint
-  return snapshot
-
-
-def runtime_status():
-  op = adapter()
-  record = op.get("JetlinkRuntime")
-  try:
-    boot_id = Path("/proc/sys/kernel/random/boot_id").read_text().strip()
-  except OSError:
-    boot_id = None
-  pid = record.get("pid") if isinstance(record, dict) else None
-  start = process_start_ticks(pid) if type(pid) is int and pid > 0 else None
-  return runtime_snapshot(record, time.monotonic_ns(), boot_id, start)
-
-
-class StarPilotAdapter:
-  basedir = BASEDIR
-  catalog_selector = 0
-
-  @property
-  def keys(self):
-    return keys()
-
-  @cached_property
-  def log(self):
-    from openpilot.common.swaglog import cloudlog
-    return cloudlog
-
-  def params_dir(self) -> Path:
-    return params_directory()
-
-  @property
-  def params(self):
-    from openpilot.common.params import Params
-    directory = self.params_dir()
-    if getattr(self, "_params_directory", None) != directory:
-      self._params = Params()
-      self._params_directory = directory
-    return self._params
-
-  def get(self, key):
-    try:
-      return self.params.get(key)
-    except Exception:
-      return None
-
-  def put(self, key, value, *, block=False):
-    self.params.put(key, value, block=block)
-
-  def remove(self, key):
-    self.params.remove(key)
-
-  def chestnut_present(self):
-    from openpilot.common.hardware.usb import is_chestnut_usb_id, read_int, usb_devices
-    return any(is_chestnut_usb_id(read_int(path / "idVendor", 16), read_int(path / "idProduct", 16), True)
-               for path in usb_devices())
-
-  def camera(self):
-    from openpilot.common.hardware import HARDWARE
-    camera = (1344, 760) if HARDWARE.get_device_type() == "mici" else (1928, 1208)
-    return (*camera, 512, 256)
-
-  def warp_path(self, cam_w, cam_h, model_w, model_h):
-    return BASEDIR / "openpilot/selfdrive/modeld/models" / f"jetlink_warp_{cam_w}x{cam_h}_{model_w}x{model_h}.pkl"
-
-  def model_root(self):
-    from openpilot.starpilot.models.manager import ROOT
-    return ROOT
-
-  def model_face(self):
-    from jetlink.openpilot import ModelFace
-    from openpilot.selfdrive.modeld.constants import ModelConstants
-    from openpilot.selfdrive.modeld.modeld import LAT_SMOOTH_SECONDS, LONG_SMOOTH_SECONDS, get_action_from_model
-    from openpilot.system.camerad.cameras.nv12_info import get_nv12_info
-    return ModelFace(remote_parser, lambda w, h: get_nv12_info(w, h)[3], ModelConstants.DESIRE_LEN, ModelConstants,
-                     LAT_SMOOTH_SECONDS, LONG_SMOOTH_SECONDS, get_action_from_model)
-
-  def engagement(self):
-    from openpilot.cereal.messaging import SubMaster
-    sm = SubMaster(["selfdriveState", "carControl"])
-
-    def engaged(timeout_ms):
-      sm.update(timeout_ms)
-      return (not sm.all_checks() or sm["selfdriveState"].enabled or sm["selfdriveState"].active or
-              sm["carControl"].latActive or sm["carControl"].longActive)
-    return engaged
-
-  def event(self, name, **fields):
-    self.log.event(name, **fields)
-
-  def make_warp(self, cam_w, cam_h, model_w, model_h):
-    from openpilot.starpilot.models.jetlink_warp import make_warp
-    return make_warp(cam_w, cam_h, model_w, model_h)
-
-
-def remote_parser():
-  import numpy as np
-  from openpilot.selfdrive.modeld.parse_model_outputs import Parser
-
-  class RemoteParser(Parser):
-    def parse_outputs(self, outs):
-      if not all(np.isfinite(value).all() for value in outs.values()):
-        raise ValueError("non-finite Jetlink model output")
-      parsed = super().parse_outputs(outs)
-      if not all(np.isfinite(value).all() for value in parsed.values()):
-        raise ValueError("non-finite parsed Jetlink model output")
-      return parsed
-  return RemoteParser()
-
-
-@lru_cache(maxsize=1)
 def adapter():
-  return StarPilotAdapter()
+  return _adapter_class()()
 
 
 def owner_config():
-  from jetlink.openpilot import OwnerConfig
-  from openpilot.common.hardware.usb import CHESTNUT_USB_IDS, CHESTNUT_ROM_USB_IDS
-  return OwnerConfig(params_directory(), keys(), frozenset(CHESTNUT_USB_IDS + CHESTNUT_ROM_USB_IDS),
-                     ADAPTER_MODULE, BASEDIR, {}, Path("/data/log/jetlink-owner.log"))
+  return _adapter_class().owner_config()
 
 
-@lru_cache(maxsize=1)
-def get_jetlink():
+def main():
+  _adapter_class().main()
+
+
+def link_on() -> bool:
+  """The Jetlink setting, read as a file, so it works with no jetlink at all."""
+  prefix = os.environ.get("OPENPILOT_PREFIX", "")
+  root = os.environ.get("PARAMS_ROOT") or ("/data/params" if os.path.isfile("/AGNOS") else
+                                           os.path.expanduser(f"~/.comma{prefix}/params"))
   try:
-    from jetlink.openpilot import API, bind
-    if API != API_VERSION:
-      raise RuntimeError(f"unsupported Jetlink API {API}")
-    return bind(adapter())
-  except Exception:
-    logging.getLogger(__name__).exception("Jetlink unavailable; retaining local model")
+    with open(os.path.join(root, os.environ.get("OPENPILOT_PREFIX", "d"), "JetlinkMode")) as f:
+      return f.read().strip() in ("1", "2")
+  except OSError:
+    return False
+
+
+def _cloudlog():
+  from openpilot.common.swaglog import cloudlog
+  return cloudlog
+
+
+@functools.cache
+def hooks():
+  """What every hook line calls, one per process. Never raises."""
+  try:
+    from jetlink.openpilot.hooks import Hooks, Off
+  except Exception as e:
+    return _NoJetlink(f"jetlink is missing: {e}")
+  try:
+    if not supported_device():
+      return Off(UNSUPPORTED, link_on=link_on)
+    return Hooks(API, adapter, link_on=link_on, log=_cloudlog)
+  except Exception as e:
+    return Off(str(e), link_on=link_on)
+
+
+class _NoJetlink:
+  """The hooks with no usable jetlink checkout, each answering link off. Only these are reached: modeld stops at
+  prepare()."""
+  def __init__(self, why):
+    self._why = why
+
+  def should_run(self, started, params, CP):
+    return False
+
+  def status(self):
     return None
 
+  def reason(self):
+    return self._why if link_on() else None
 
-class SmallModelFace:
-  """Expose catalog history to Jetlink without changing the catalog runner."""
-  def __init__(self, model):
-    self.model = model
-    self.lat_delay = 0.0
-    self.PLANPLUS_CONTROL = 1.0
-    self.frame_drop_ratio = 0.0
+  def prepare(self, chestnut):
+    return False
 
-  @property
-  def prev_desire(self):
-    return self.model.prev_desire
+  def shutdown_step(self, should_shutdown, reason=""):
+    return should_shutdown
 
-  @property
-  def numpy_inputs(self):
-    return self.model.npy
+  def runtime(self):
+    return None
 
-  def __getattr__(self, name):
-    return getattr(self.model, name)
-
-
-class JetlinkModel:
-  """Keep the current catalog's action contract while the remote model shadows."""
-  def __init__(self, joined, small):
-    self.joined = joined
-    self.small = small
-    self._session = uuid.uuid4().hex
-    self._process_start_ticks = process_start_ticks(os.getpid())
-    try:
-      self._boot_id = Path("/proc/sys/kernel/random/boot_id").read_text().strip()
-    except OSError:
-      self._boot_id = None
-    self._published_ns = 0
-    self._published_state = None
-    self._output = None
-    self._output_ns = 0
-    self._runtime_pending = None
-    self._runtime_closed = False
-    self._runtime_condition = threading.Condition()
-    self._runtime_thread = threading.Thread(target=self._write_runtime, name="jetlink-runtime", daemon=True)
-    self._runtime_thread.start()
-
-  def __getattr__(self, name):
-    return getattr(self.joined, name)
-
-  def run(self, bufs, transforms, inputs, after_enqueue=None):
-    outputs = self.joined.run(bufs, transforms, inputs, after_enqueue)
-    client = self.joined.client
-    if self.chestnut and outputs is not None and client is not None and client.last_output is not self._output:
-      self._output = client.last_output
-      self._output_ns = time.monotonic_ns()
-    self.publish_runtime()
-    return outputs
-
-  def action(self, outputs, previous, lat_action_t, long_action_t, v_ego):
-    from openpilot.selfdrive.modeld.modeld import LAT_SMOOTH_SECONDS, LONG_SMOOTH_SECONDS, get_action_from_model
-    from openpilot.starpilot.models.runner import action_from_outputs
-    if self.chestnut:
-      return get_action_from_model(outputs, previous, lat_action_t, long_action_t, v_ego)
-    return action_from_outputs(outputs, self.small.behavior_version, previous, lat_action_t, long_action_t, v_ego,
-                               LAT_SMOOTH_SECONDS, LONG_SMOOTH_SECONDS)
-
-  def observe_frame(self, lat_delay, drop_ratio):
-    self.joined.lat_delay = lat_delay
-    self.joined.frame_drop_ratio = drop_ratio
-
-  def publish_runtime(self, *, closed=False):
-    now = time.monotonic_ns()
-    state = "unavailable" if closed else self.big_model_state
-    if state == self._published_state and now - self._published_ns < 1_000_000_000:
-      return
-    self._published_ns, self._published_state = now, state
-    record = {"version": 1, "boot_id": self._boot_id, "session": self._session, "pid": os.getpid(),
-              "process_start_ticks": self._process_start_ticks,
-              "mono_ns": now, "state": state, "handovers": self.handovers,
-              "remote_output_mono_ns": self._output_ns if self.chestnut and not closed else 0}
-    if self.chestnut and not closed:
-      spec = self.joined.spec
-      record["artifact_sha256"] = spec.sha256
-      record["artifact_nbytes"] = spec.nbytes
-      if isinstance(spec.checkpoint, str):
-        record["model_checkpoint"] = spec.checkpoint[:256]
-    with self._runtime_condition:
-      self._runtime_pending = record
-      self._runtime_condition.notify()
-
-  def _write_runtime(self):
-    while True:
-      with self._runtime_condition:
-        self._runtime_condition.wait_for(lambda: self._runtime_pending is not None or self._runtime_closed)
-        if self._runtime_pending is None:
-          return
-        record, self._runtime_pending = self._runtime_pending, None
-        closed = self._runtime_closed
-      try:
-        adapter().put("JetlinkRuntime", record, block=True)
-      except Exception:
-        adapter().log.exception("Jetlink runtime status unavailable")
-      if closed:
-        return
-
-  def close(self):
-    if self._runtime_closed:
-      return
-    self.joined.close()
-    self.publish_runtime(closed=True)
-    with self._runtime_condition:
-      self._runtime_closed = True
-      self._runtime_condition.notify()
-
-
-def attach(small, cam_w, cam_h, *, chestnut=False, recovery=False):
-  if chestnut or recovery or not supported_device():
-    return small
-  link = get_jetlink()
-  if link is None:
-    return small
-  try:
-    if not link.prepare():
-      return small
-    face = SmallModelFace(small)
-    joined = link.attach(face, cam_w, cam_h)
-    if joined is None or joined is face:
-      return small
-    model = JetlinkModel(joined, small)
-    import atexit
-    atexit.register(model.close)
-    return model
-  except Exception:
-    adapter().log.exception("Jetlink startup failed; retaining local model")
-    return small
+  def swap_events(self, tick_s):
+    return None

openpilot/starpilot/models/jetlink_settings.py

@@ -4,6 +4,14 @@ from __future__ import annotations
 MODES = ("off", "usb", "ios")
 
 
+def runtime_view(status):
+  # what modeld runs over the link, by jetlink's runtime record
+  if status is None or not status.active:
+    return {"state": getattr(status, "running", None) or "unavailable", "active": False}
+  return {"state": status.running, "active": True, "artifact_sha256": status.running_sha256,
+          "model_checkpoint": status.running_checkpoint}
+
+
 class JetlinkSettings:
   def __init__(self, *, params=None, link=None, device_type=None, chestnut=None, runtime=None):
     self._params = params
@@ -28,7 +36,7 @@ class JetlinkSettings:
               "progress": "", "model": None, "transport": "USB", "chargePhone": False}
 
   def _snapshot(self):
-    from openpilot.starpilot.models.jetlink_adapter import adapter, get_jetlink, runtime_status, supported_device
+    from openpilot.starpilot.models.jetlink_adapter import adapter, hooks, supported_device
     if self._device_type is None:
       from openpilot.common.hardware import HARDWARE
       device = HARDWARE.get_device_type()
@@ -39,7 +47,7 @@ class JetlinkSettings:
     params = self.params()
     raw = params.get("JetlinkMode")
     mode = MODES[raw] if type(raw) is int and 0 <= raw < len(MODES) else "off"
-    link = self._link() if self._link is not None else get_jetlink()
+    link = self._link() if self._link is not None else hooks()
     status = link.status() if link is not None else None
     reason = "" if supported else "Jetlink requires a comma 3X or comma 4."
     if mode != "off" and status is None:
@@ -47,7 +55,7 @@ class JetlinkSettings:
     if status is not None and status.reason:
       reason = str(status.reason)
     progress = status.progress if status is not None and isinstance(status.progress, dict) else {}
-    runtime = self._runtime() if self._runtime is not None else runtime_status()
+    runtime = self._runtime() if self._runtime is not None else runtime_view(status)
     active = mode != "off" and supported and not chestnut and runtime.get("active") is True
     runtime_state = runtime.get("state", "unavailable")
     stage = str(progress.get("stage", ""))

openpilot/starpilot/models/runner.py

@@ -239,6 +239,13 @@ class CatalogModelState:
     parsed.update(results["on_policy" if "on_policy" in results else "policy"])
     return parsed
 
+  def get_action_from_model(self, outputs: dict[str, np.ndarray], previous: log.ModelDataV2.Action,
+                            lat_action_t: float, long_action_t: float, v_ego: float) -> log.ModelDataV2.Action:
+    """modeld's action call: this catalog generation's units, with modeld's smoothing."""
+    from openpilot.selfdrive.modeld.modeld import LAT_SMOOTH_SECONDS, LONG_SMOOTH_SECONDS
+    return action_from_outputs(outputs, self.behavior_version, previous, lat_action_t, long_action_t, v_ego,
+                               LAT_SMOOTH_SECONDS, LONG_SMOOTH_SECONDS)
+
   def warmup(self) -> None:
     frames = {name: np.zeros(self.frame_buf_size, dtype=np.uint8) for name in self.vision_input_names}
     self._blob_cache.update({(name, frame.ctypes.data): Tensor(frame, device=self.warp_device).realize()

openpilot/starpilot/models/tests/test_jetlink_adapter.py

@@ -1,141 +1,62 @@
-import os
 from types import SimpleNamespace
-from unittest.mock import Mock, patch
+from unittest.mock import patch
+
+import numpy as np
 
 from jetlink.openpilot import Openpilot, conformance
+from jetlink.openpilot.hooks import Hooks, Off
 from openpilot.starpilot.models import jetlink_adapter as adapter
 
 
-def test_public_interface_and_params_directory(monkeypatch, tmp_path):
+def test_interface_keys():
+  # the params directory is Stock's, held against openpilot's Params in test_jetlink_lifecycle
   op = adapter.adapter()
-  op.log = Mock()
   assert conformance(op, Openpilot) == []
-  monkeypatch.setenv("PARAMS_ROOT", str(tmp_path))
-  monkeypatch.setenv("OPENPILOT_PREFIX", "jetlink-test")
-  assert op.params_dir() == tmp_path / "jetlink-test"
-  assert adapter.owner_config().params_dir == op.params_dir()
+  assert adapter.owner_config().adapter == adapter.__name__
   assert op.keys.link == "JetlinkMode"
-  assert op.keys.big_model != "ModelSelected"
+  # nothing writes a big-model slot or catalog: jetlink's default model, so a fresh Jetson is provisioned
+  assert op.keys.big_model is None and op.keys.catalog is None
 
 
-def test_small_history_and_current_delay_forwarding():
-  small = SimpleNamespace(prev_desire=object(), npy={"desire": object()}, input_queues={"history": object()},
-                          chestnut=False, behavior_version="v16")
-  face = adapter.SmallModelFace(small)
-  assert face.prev_desire is small.prev_desire
-  assert face.numpy_inputs is small.npy
-  assert face.input_queues is small.input_queues
-  joined = SimpleNamespace(chestnut=False, lat_delay=0., frame_drop_ratio=0.)
-  model = adapter.JetlinkModel(joined, small)
-  model.observe_frame(.23, .004)
-  assert joined.lat_delay == .23
-  assert joined.frame_drop_ratio == .004
+def test_guard_answers_link_off_without_a_usable_jetlink():
+  adapter.hooks.cache_clear()
+  try:
+    with patch.object(adapter, "supported_device", return_value=False):
+      assert isinstance(adapter.hooks(), Off)
+    adapter.hooks.cache_clear()
+    with patch.object(adapter, "supported_device", return_value=True), patch.object(adapter, "API", 2):
+      assert isinstance(adapter.hooks(), Off)
+    adapter.hooks.cache_clear()
+    with patch.object(adapter, "supported_device", return_value=True):
+      assert isinstance(adapter.hooks(), Hooks)
+    adapter.hooks.cache_clear()
+    with patch.dict("sys.modules", {"jetlink.openpilot.hooks": None}):
+      hooks = adapter.hooks()
+    assert not hooks.prepare(False) and not hooks.should_run(False, None, None) and hooks.status() is None
+    assert hooks.shutdown_step(True) and not hooks.shutdown_step(False)
+  finally:
+    adapter.hooks.cache_clear()
 
 
-def test_chestnut_and_recovery_never_prepare_link():
-  small = object()
-  with patch.object(adapter, "get_jetlink") as get:
-    assert adapter.attach(small, 1344, 760, chestnut=True) is small
-    assert adapter.attach(small, 1344, 760, recovery=True) is small
-    get.assert_not_called()
-
-
-def test_link_failure_preserves_small():
-  small = object()
-  link = Mock()
-  link.prepare.side_effect = RuntimeError("link unavailable")
-  op = Mock()
-  with patch.object(adapter, "get_jetlink", return_value=link), patch.object(adapter, "adapter", return_value=op), \
-       patch.object(adapter, "supported_device", return_value=True):
-    assert adapter.attach(small, 1344, 760) is small
-    op.log.exception.assert_called_once()
-
-
-def test_runtime_requires_fresh_same_boot_and_process():
-  record = {"version": 1, "boot_id": "boot", "session": "a" * 32, "pid": os.getpid(), "process_start_ticks": 12,
-            "artifact_sha256": "b" * 64, "mono_ns": 2_000_000_000, "remote_output_mono_ns": 1_999_000_000, "state": "running"}
-  assert adapter.runtime_snapshot(record, 3_000_000_000, "boot", 12)["active"]
-  for changes in ({"boot_id": "other"}, {"process_start_ticks": 13}, {"session": "bad"},
-                  {"mono_ns": 3_000_000_001}, {"remote_output_mono_ns": 0}, {"state": "ready"}):
-    assert not adapter.runtime_snapshot(record | changes, 3_000_000_000, "boot", 12)["active"]
-  assert not adapter.runtime_snapshot(record, 3_500_000_001, "boot", 12)["active"]
-
-
-def test_handover_action_contract_tracks_actual_driver():
+def test_one_action_call_for_every_small_model():
   from openpilot.selfdrive.modeld import modeld
   from openpilot.starpilot.models import runner
 
-  small = SimpleNamespace(behavior_version="v9")
-  joined = SimpleNamespace(chestnut=False)
-  model = adapter.JetlinkModel(joined, small)
+  small = runner.CatalogModelState.__new__(runner.CatalogModelState)
+  small.behavior_version = "v9"
   outputs, previous = object(), object()
-  with patch.object(runner, "action_from_outputs", return_value="catalog") as catalog, \
-       patch.object(modeld, "get_action_from_model", return_value="remote") as remote:
-    assert model.action(outputs, previous, .4, .7, 12.) == "catalog"
-    assert catalog.call_args.args[:6] == (outputs, "v9", previous, .4, .7, 12.)
-    joined.chestnut = True
-    assert model.action(outputs, previous, .4, .7, 12.) == "remote"
-    remote.assert_called_once_with(outputs, previous, .4, .7, 12.)
-    joined.chestnut = False
-    assert model.action(outputs, previous, .4, .7, 12.) == "catalog"
-    assert catalog.call_count == 2
+  with patch.object(runner, "action_from_outputs", return_value="catalog") as catalog:
+    assert small.get_action_from_model(outputs, previous, .4, .7, 12.) == "catalog"
+  catalog.assert_called_once_with(outputs, "v9", previous, .4, .7, 12., modeld.LAT_SMOOTH_SECONDS, modeld.LONG_SMOOTH_SECONDS)
+  assert modeld.ModelState.get_action_from_model is modeld.get_action_from_model
 
 
-def test_remote_parser_rejects_nonfinite_and_missing_outputs():
-  import numpy as np
-  import pytest
-
-  parser = adapter.remote_parser()
-  with pytest.raises(ValueError, match="non-finite"):
-    parser.parse_outputs({"action": np.array([[np.nan, 0.]])})
-  with pytest.raises(ValueError, match="Missing output"):
-    parser.parse_outputs({})
+def test_small_reset_only_for_the_catalog_runner():
+  assert adapter.adapter().small_reset(SimpleNamespace(input_queues={}, npy={}, prev_desire=np.zeros(8))) is None
 
 
-def test_vendor_attach_failure_returns_unwrapped_small():
-  small = object()
-  link = Mock()
-  link.prepare.return_value = True
-  link.attach.side_effect = lambda face, w, h: face
-  with patch.object(adapter, "get_jetlink", return_value=link), patch.object(adapter, "supported_device", return_value=True):
-    assert adapter.attach(small, 1344, 760) is small
-
-
-def test_runtime_persistence_runs_off_frame_thread():
-  import threading
-
-  entered, release = threading.Event(), threading.Event()
-  writes = []
-  def write(key, record, *, block=False):
-    writes.append((threading.get_ident(), key, record, block))
-    entered.set()
-    release.wait(1.)
-
-  joined = SimpleNamespace(chestnut=False, big_model_state="joining", handovers=0, close=Mock())
-  op = Mock()
-  op.put.side_effect = write
-  with patch.object(adapter, "adapter", return_value=op):
-    model = adapter.JetlinkModel(joined, object())
-    try:
-      model.publish_runtime()
-      assert entered.wait(1.)
-      assert writes[0][0] != threading.get_ident()
-      assert writes[0][1] == "JetlinkRuntime"
-      assert writes[0][3] is True
-      model.close()
-    finally:
-      release.set()
-      model._runtime_thread.join(1.)
-    assert not model._runtime_thread.is_alive()
-    assert writes[-1][2]["state"] == "unavailable"
-
-
-def test_unsupported_device_never_prepares_link():
-  small = object()
+def test_unsupported_device():
   assert adapter.supported_device("tizi")
   assert adapter.supported_device("mici")
   for device in ("tici", "pc", "unknown", "neo"):
     assert not adapter.supported_device(device)
-    with patch.object(adapter, "supported_device", return_value=False), patch.object(adapter, "get_jetlink") as get:
-      assert adapter.attach(small, 1928, 1208) is small
-      get.assert_not_called()

openpilot/starpilot/models/tests/test_modeld_startup_fallback.py

@@ -3,12 +3,12 @@
 import ast
 from pathlib import Path
 from types import SimpleNamespace
-from unittest.mock import Mock, patch
+from unittest.mock import Mock
 
 import pytest
 
+from jetlink.openpilot.testing import FakeJoinedModel
 from openpilot.starpilot.models.status import ModelVariant
-from openpilot.starpilot.models.jetlink_adapter import JetlinkModel, attach as attach_jetlink
 
 
 MODEL_SOURCE = Path(__file__).resolve().parents[3] / "selfdrive/modeld/modeld.py"
@@ -30,6 +30,17 @@ def test_realtime_starts_after_model_load_and_receipt():
   assert max(line("load_small") + line("loader.join")) < startup_receipt.lineno < line("config_realtime_process")[0] < line("PubMaster")[0]
 
 
+def test_jetlink_gets_the_frames_delay_before_every_run():
+  # a model that joins starts on the current delay; check_seam holds the frame_drop_ratio write
+  loop = [node for node in main_body() if isinstance(node, ast.While)][-1]  # the frame loop
+  statements = [node for node in ast.walk(loop) if isinstance(node, (ast.Assign, ast.Try))]
+  def line(source):
+    return next(node.lineno for node in statements if isinstance(node, ast.Assign) and ast.unparse(node) == source)
+  run = next(node.lineno for node in statements if isinstance(node, ast.Try) and
+             any(isinstance(child, ast.Call) and ast.unparse(child.func) == "model.run" for child in ast.walk(node)))
+  assert line("lat_delay = sm['lateralDelay'].lateralDelay + LAT_SMOOTH_SECONDS") < line("model.lat_delay = lat_delay") < run
+
+
 @pytest.mark.parametrize("runner_chestnut,stored_active", [(True, False), (True, True), (False, True), (False, False)])
 def test_inference_failure_uses_runner_identity(runner_chestnut, stored_active):
   # Run the production try/except statement with deliberately stale status Params.
@@ -46,7 +57,7 @@ def test_inference_failure_uses_runner_identity(runner_chestnut, stored_active):
   env: dict = {"model": failed_model, "small_model": small_model, "params": params, "receipt_owner": receipt,
              "chestnut_state": chestnut_state, "run_count": 1, "ModelConstants": SimpleNamespace(MODEL_RUN_FREQ=20),
              "SERVICE_LIST": {"chestnutGpuState": SimpleNamespace(frequency=1)}, "bufs": {}, "transforms": {}, "inputs": {},
-             "small_prepared": object(), "ModelVariant": ModelVariant, "JetlinkModel": JetlinkModel, "drop_chestnut": Mock(), "cloudlog": Mock()}
+             "small_prepared": object(), "ModelVariant": ModelVariant, "jetlink_model": None, "drop_chestnut": Mock(), "cloudlog": Mock()}
   code = compile(ast.fix_missing_locations(ast.Module(body=[fallback], type_ignores=[])), str(MODEL_SOURCE), "exec")
   if runner_chestnut:
     exec(code, env)
@@ -71,14 +82,14 @@ def test_remote_inference_failure_does_not_change_chestnut_status():
   fallback = next(node for statement in main_body() for node in ast.walk(statement)
                   if isinstance(node, ast.Try) and any(isinstance(child, ast.Call) and ast.unparse(child.func) == "model.run"
                                                        for child in ast.walk(node)))
-  failed_model = object.__new__(JetlinkModel)
-  failed_model.joined = SimpleNamespace(chestnut=True, run=Mock(side_effect=RuntimeError("remote inference failed")))
+  failed_model = FakeJoinedModel(chestnut=True)
+  failed_model.run = Mock(side_effect=RuntimeError("remote inference failed"))
   failed_model.close = Mock()
   small_model = SimpleNamespace(chestnut=False, model_id="fallback-small")
   params, receipt = Mock(), Mock()
   env = {"model": failed_model, "small_model": small_model, "params": params, "receipt_owner": receipt,
          "chestnut_state": None, "run_count": 1, "bufs": {}, "transforms": {}, "inputs": {},
-         "small_prepared": object(), "ModelVariant": ModelVariant, "JetlinkModel": JetlinkModel, "drop_chestnut": Mock(), "cloudlog": Mock()}
+         "small_prepared": object(), "ModelVariant": ModelVariant, "jetlink_model": failed_model, "drop_chestnut": Mock(), "cloudlog": Mock()}
   execute([fallback], env)
   failed_model.close.assert_called_once_with()
   env["drop_chestnut"].assert_not_called()
@@ -172,6 +183,10 @@ def test_big_worker_waits_before_load_and_timeout_preserves_small_fallback(custo
     return state(width, height, chestnut), "prepared-" + model_id
 
   params, receipt = Mock(), Mock()
+  from jetlink.openpilot.hooks import Hooks
+  from openpilot.starpilot.models.jetlink_adapter import API
+  bind = Mock(side_effect=AssertionError("a chestnut drive must not start jetlink"))
+  hooks = Hooks(API, bind)
   selection = SimpleNamespace(big_path="selected-big" if custom else None, small_path="selected-small" if custom else None,
                               small_version=1, small_id="small", small_sha256="small-hash",
                               big_version=1, big_id="big", big_sha256="hash")
@@ -183,14 +198,15 @@ def test_big_worker_waits_before_load_and_timeout_preserves_small_fallback(custo
          "receipt_owner": receipt, "threading": SimpleNamespace(Thread=Thread), "BIG_MODEL_TIMEOUT": 30,
          "params": params, "requested_model_id": "big", "ModelVariant": ModelVariant,
          "CP": SimpleNamespace(brand="mock"), "demo": False,
-         "wait_for_chestnut_power": lambda CP, timeout: events.append("power"), "attach_jetlink": attach_jetlink,
-         "drop_chestnut": Mock(), "release_compile_pool": Mock(), "gc": Mock()}
+         "wait_for_chestnut_power": lambda CP, timeout: events.append("power"),
+         "jetlink_adapter": SimpleNamespace(hooks=lambda: hooks), "drop_chestnut": Mock(),
+         "release_compile_pool": Mock(), "gc": Mock()}
   wrapper = ast.parse("def startup():\n  pass").body[0]
   wrapper.body = body[start:end] + [ast.Return(value=ast.Call(func=ast.Name(id="locals", ctx=ast.Load()), args=[], keywords=[]))]
   execute([wrapper], env)
-  with patch("openpilot.starpilot.models.jetlink_adapter.get_jetlink") as get_link:
-    env.update(env["startup"]())
-    get_link.assert_not_called()
+  env.update(env["startup"]())
+  bind.assert_not_called()
+  assert env["jetlink_model"] is None
   assert events[0] == ("small" if custom else "wait")
   assert events.count("small") == 1
   assert ("big" in events) is not timeout

openpilot/starpilot/system/jetlink/daemon.py

@@ -1,42 +0,0 @@
-"""Resident FunctionFS owner; release shared resources before waiting on its worker."""
-import signal
-
-
-def owner_class():
-  from jetlink.comma import owner as comma_owner, root
-
-  class ManagedOwner(comma_owner.Owner):
-    def stop_worker(self):
-      if self.stop or self.settings.mode() == 'off':
-        # Mode-off can stop the process before the upstream next poll restores
-        # tuning. Chestnut takeover also needs its host port back immediately.
-        self.port.off()
-        self.close_link()
-        commands = [('vm', 'restore'), ('udc', 'restore')]
-        if self.tuned == 'ios':
-          commands.append(('draw', 'on'))
-        for command in commands:
-          if not root.run(*command, timeout=1.0):
-            comma_owner.gadget.log.error('Jetlink cleanup failed: %s', command)
-        self.tuned = None
-      super().stop_worker()
-
-  return ManagedOwner
-
-
-def main() -> None:
-  from jetlink.comma import gadget, owner as comma_owner
-  from jetlink.openpilot.owner import worker
-  from jetlink.openpilot.settings import FileParams, Settings
-  from openpilot.starpilot.models.jetlink_adapter import owner_config
-  config = owner_config()
-  gadget.set_logger(comma_owner.logger(config.log_file))
-  owner = owner_class()(worker(config), cwd=str(config.cwd), env=dict(config.env),
-                        settings=Settings(FileParams(config.params_dir), config.keys), chestnut_ids=config.chestnut_ids)
-  signal.signal(signal.SIGTERM, owner.request_stop)
-  signal.signal(signal.SIGINT, owner.request_stop)
-  owner.run()
-
-
-if __name__ == '__main__':
-  main()

openpilot/starpilot/system/jetlink/lifecycle.py

@@ -1,25 +1,26 @@
 """Manager's opt-in Jetlink owner and bounded remote power-off handoff."""
 from __future__ import annotations
 
+import time
+
+# jetlink's SHUTDOWN_TIMEOUT (25 s) plus 5 s, written out so it holds without jetlink: a jetlink regression
+# must never keep a comma from powering off and drain the car's battery
+SHUTDOWN_CAP_S = 30.
+
 
 def enabled(params) -> bool:
-  # Keep off/default startup independent of the transport and model runtime.
-  raw = params.get('JetlinkMode')
-  if type(raw) is not int or raw not in (1, 2):
-    return False
-  from openpilot.starpilot.models.jetlink_adapter import get_jetlink, supported_device
-  if not supported_device():
-    return False
-  link = get_jetlink()
-  return link is not None and link.enabled()
+  # with the link off, should_run() binds nothing: jetlink's light hooks module alone loads
+  from openpilot.starpilot.models.jetlink_adapter import hooks
+  return hooks().should_run(False, params, None)
 
 
-def shutdown(params) -> None:
-  reason = next((key for key in ('DoUninstall', 'DoReboot', 'DoShutdown') if params.get_bool(key)), None)
-  if reason is None or not enabled(params):
+def shutdown(params, *, clock=time.monotonic, sleep=time.sleep) -> None:
+  # A power-off only: the Jetson is on an always-on supply and wakes only on a USB edge, so one powered off at
+  # a comma reboot would not come back. The resident owner must still be alive to run it.
+  from openpilot.starpilot.models.jetlink_adapter import hooks, link_on
+  if not params.get_bool('DoShutdown') or not link_on():
     return
-  from openpilot.starpilot.models.jetlink_adapter import get_jetlink
-  # The resident owner must still be alive to lend the shutdown worker its link.
-  link = get_jetlink()
-  if link is not None:
-    link.shutdown(reason, timeout=25.0)
+  link = hooks()
+  deadline = clock() + SHUTDOWN_CAP_S
+  while not link.shutdown_step(True, 'DoShutdown') and clock() < deadline:
+    sleep(0.1)

openpilot/starpilot/tests/test_jetlink_lifecycle.py

@@ -1,5 +1,4 @@
 """Opt-in process ownership and remote shutdown before manager cleanup."""
-import sys
 from types import SimpleNamespace
 import unittest
 from unittest.mock import Mock, patch
@@ -25,45 +24,54 @@ def _cleanup_child(ready, cleaned, release):
 
 
 class TestJetlinkLifecycle(unittest.TestCase):
-  def setUp(self):
-    hardware = patch('openpilot.common.hardware.HARDWARE.get_device_type', return_value='tizi')
-    hardware.start()
-    self.addCleanup(hardware.stop)
-
-  def test_disabled_or_malformed_mode_never_imports_adapter(self):
-    for mode in (None, 0, -1, 3, True, False, b'1', b'2', b'corrupt'):
-      params = SimpleNamespace(get=lambda key, mode=mode: mode)
-      with self.subTest(mode=mode), patch.dict(sys.modules, {'openpilot.starpilot.models.jetlink_adapter': None}):
-        self.assertFalse(lifecycle.enabled(params))
-
-  def test_selected_owner_requires_available_transport_without_chestnut(self):
-    params = SimpleNamespace(get=lambda key: 1)
+  def jetlink(self, on):
     link = Mock()
-    adapter = SimpleNamespace(get_jetlink=Mock(return_value=link), supported_device=lambda: True)
-    with patch.dict(sys.modules, {'openpilot.starpilot.models.jetlink_adapter': adapter}):
-      for available in (True, False):
-        link.enabled.return_value = available
-        self.assertEqual(lifecycle.enabled(params), available)
-      adapter.get_jetlink.return_value = None
-      self.assertFalse(lifecycle.enabled(params))
+    for name, value in (('hooks', lambda: link), ('link_on', lambda: on)):
+      patcher = patch(f'openpilot.starpilot.models.jetlink_adapter.{name}', value)
+      patcher.start()
+      self.addCleanup(patcher.stop)
+    return link
 
-  def test_only_power_actions_request_bounded_remote_shutdown(self):
-    link = Mock()
-    link.enabled.return_value = True
-    adapter = SimpleNamespace(get_jetlink=lambda: link, supported_device=lambda: True)
-    with patch.dict(sys.modules, {'openpilot.starpilot.models.jetlink_adapter': adapter}):
-      for reason in (None, 'DoShutdown', 'DoReboot', 'DoUninstall'):
-        link.shutdown.reset_mock()
-        params = SimpleNamespace(get=lambda key: 2, get_bool=lambda key, reason=reason: key == reason)
-        lifecycle.shutdown(params)
-        if reason is None:
-          link.shutdown.assert_not_called()
-        else:
-          link.shutdown.assert_called_once_with(reason, timeout=25.0)
-      link.enabled.return_value = False
-      link.shutdown.reset_mock()
+  def test_with_the_link_off_jetlink_is_never_bound(self):
+    from jetlink.openpilot.hooks import Hooks
+    from openpilot.starpilot.models.jetlink_adapter import API
+    self.jetlink(False)
+    factory = Mock()
+    with patch('openpilot.starpilot.models.jetlink_adapter.hooks', lambda: Hooks(API, factory, link_on=lambda: False)):
+      self.assertFalse(lifecycle.enabled(None))
+      lifecycle.shutdown(SimpleNamespace(get_bool=lambda key: key == 'DoShutdown'))
+    factory.assert_not_called()
+
+  def test_the_owner_follows_jetlink(self):
+    link = self.jetlink(True)
+    for available in (True, False):
+      link.should_run.return_value = available
+      self.assertEqual(lifecycle.enabled(None), available)
+
+  def test_only_power_off_asks_the_host_and_is_capped(self):
+    link = self.jetlink(True)
+    for reason in (None, 'DoReboot', 'DoUninstall', 'DoShutdown'):
+      link.shutdown_step.reset_mock()
+      link.shutdown_step.return_value = True
+      params = SimpleNamespace(get_bool=lambda key, reason=reason: key == reason)
       lifecycle.shutdown(params)
-      link.shutdown.assert_not_called()
+      if reason == 'DoShutdown':
+        link.shutdown_step.assert_called_once_with(True, 'DoShutdown')
+      else:
+        link.shutdown_step.assert_not_called()
+
+    # a host that never answers holds the power-off for the cap, no longer
+    now, waits = [0.], []
+
+    def sleep(seconds):
+      waits.append(seconds)
+      if len(waits) > 1000:
+        raise AssertionError("the power-off wait has no cap")
+      now[0] += seconds
+    link.shutdown_step.return_value = False
+    lifecycle.shutdown(params, clock=lambda: now[0], sleep=sleep)
+    self.assertGreaterEqual(now[0], lifecycle.SHUTDOWN_CAP_S)
+    self.assertLess(now[0], lifecycle.SHUTDOWN_CAP_S + 1)
 
   def test_registered_native_int_and_owner_file_reader_agree(self):
     from pathlib import Path
@@ -76,9 +84,11 @@ class TestJetlinkLifecycle(unittest.TestCase):
       params = Params()
       config = owner_config()
       settings = Settings(FileParams(config.params_dir), config.keys)
-      link = Mock()
-      link.enabled.return_value = True
-      with patch('openpilot.starpilot.models.jetlink_adapter.get_jetlink', return_value=link):
+      from openpilot.starpilot.models import jetlink_adapter
+      jetlink_adapter.hooks.cache_clear()
+      self.addCleanup(jetlink_adapter.hooks.cache_clear)
+      with patch('jetlink.openpilot.parts.Parts.chestnut_fitted', return_value=False), \
+           patch('openpilot.common.hardware.HARDWARE.get_device_type', return_value='tizi'):
         for value, mode in ((0, 'off'), (1, 'usb'), (2, 'ios'), (3, 'off'), (-1, 'off')):
           with self.subTest(value=value):
             params.put('JetlinkMode', value, block=True)
@@ -86,46 +96,13 @@ class TestJetlinkLifecycle(unittest.TestCase):
             self.assertEqual(params.get('JetlinkMode'), value)
             self.assertEqual(FileParams(config.params_dir).get_int('JetlinkMode'), value)
             self.assertEqual(settings.mode(), mode)
+            self.assertEqual(jetlink_adapter.link_on(), value in (1, 2))
             self.assertEqual(lifecycle.enabled(params), value in (1, 2))
         params.put('JetlinkMode', 1, block=True)
-        for device, supported in (('tizi', True), ('mici', True), ('tici', False), ('pc', False)):
-          with self.subTest(device=device), patch('openpilot.common.hardware.HARDWARE.get_device_type', return_value=device):
-            self.assertEqual(lifecycle.enabled(params), supported)
         Path(params.get_param_path('JetlinkMode')).write_bytes(b'corrupt')
         self.assertEqual(settings.mode(), 'off')
         self.assertFalse(lifecycle.enabled(params))
 
-  def test_owner_releases_resources_before_blocked_worker_wait(self):
-    import subprocess
-    from openpilot.starpilot.system.jetlink.daemon import owner_class
-    from jetlink.comma import owner as upstream
-    from jetlink.comma import root
-    from unittest.mock import call
-
-    owner = owner_class()([], settings=SimpleNamespace(mode=lambda: 'usb'), chestnut_ids=frozenset())
-    owner.stop = True
-    owner.tuned = 'ios'
-    owner.port = Mock()
-    owner.close_link = Mock()
-    owner.beat = Mock()
-    order = Mock()
-    order.attach_mock(owner.port.off, 'port')
-    order.attach_mock(owner.close_link, 'close')
-    worker = Mock()
-    owner.worker = worker
-    with patch.object(root, 'run', return_value=True) as restore, patch.object(upstream, 'WORKER_GRACE', 0):
-      order.attach_mock(restore, 'restore')
-      order.attach_mock(worker.wait, 'wait')
-      worker.wait.side_effect = subprocess.TimeoutExpired('provision', .5)
-      owner.stop_worker()
-    self.assertEqual(order.mock_calls[:6], [call.port(), call.close(), call.restore('vm', 'restore', timeout=1.0),
-                                         call.restore('udc', 'restore', timeout=1.0), call.restore('draw', 'on', timeout=1.0),
-                                         call.wait(upstream.POLL)])
-    worker.terminate.assert_called_once()
-    worker.kill.assert_called_once()
-    self.assertIsNone(owner.worker)
-    self.assertIsNone(owner.tuned)
-
   def test_process_nonblocking_stop_and_blocking_signal_cleanup(self):
     import multiprocessing
     import time

openpilot/system/manager/process_config.py

@@ -137,7 +137,7 @@ procs = [
                 enabled=platform.system() == "Linux", restart_on_exit=True),
   PythonProcess("galaxy_hotspot", "openpilot.starpilot.galaxy.hotspot_daemon", galaxy_local,
                 enabled=COMMA_HARDWARE and not os.environ.get('OPENPILOT_PREFIX'), restart_on_exit=True),
-  JetlinkProcess("jetlinkd", "openpilot.starpilot.system.jetlink.daemon", jetlink_enabled, enabled=COMMA_HARDWARE),
+  JetlinkProcess("jetlinkd", "openpilot.starpilot.models.jetlink_adapter", jetlink_enabled, enabled=COMMA_HARDWARE),
   PythonProcess("android_autod", "openpilot.starpilot.system.android_auto.daemon", android_auto_enabled, enabled=COMMA_HARDWARE,
                 restart_on_exit=True),  # boot can race bluetoothd (NoSuchUnit)
   # Boot can race the audio device; selfdrived blocks engagement while soundd is down.