diff --git a/openpilot/selfdrive/modeld/modeld.py b/openpilot/selfdrive/modeld/modeld.py
index 034a5d7422..0fd459c9b1 100755
--- a/openpilot/selfdrive/modeld/modeld.py
+++ b/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,
diff --git a/openpilot/starpilot/analytics/runtime.py b/openpilot/starpilot/analytics/runtime.py
index ed10566763..aca44d5d44 100644
--- a/openpilot/starpilot/analytics/runtime.py
+++ b/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()
diff --git a/openpilot/starpilot/analytics/tests/test_runtime.py b/openpilot/starpilot/analytics/tests/test_runtime.py
index 432f250c6a..5c4bdace5b 100644
--- a/openpilot/starpilot/analytics/tests/test_runtime.py
+++ b/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) == {}
diff --git a/openpilot/starpilot/models/jetlink_adapter.py b/openpilot/starpilot/models/jetlink_adapter.py
index 56fd8748d1..6b6b15455a 100644
--- a/openpilot/starpilot/models/jetlink_adapter.py
+++ b/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
diff --git a/openpilot/starpilot/models/jetlink_settings.py b/openpilot/starpilot/models/jetlink_settings.py
index f3ce3d72df..6c356f7f5e 100644
--- a/openpilot/starpilot/models/jetlink_settings.py
+++ b/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", ""))
diff --git a/openpilot/starpilot/models/runner.py b/openpilot/starpilot/models/runner.py
index eef59bdd2d..3f6a99f8f3 100644
--- a/openpilot/starpilot/models/runner.py
+++ b/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()
diff --git a/openpilot/starpilot/models/tests/test_jetlink_adapter.py b/openpilot/starpilot/models/tests/test_jetlink_adapter.py
index e40a65dc66..2fe6263936 100644
--- a/openpilot/starpilot/models/tests/test_jetlink_adapter.py
+++ b/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()
diff --git a/openpilot/starpilot/models/tests/test_modeld_startup_fallback.py b/openpilot/starpilot/models/tests/test_modeld_startup_fallback.py
index eb3b104a48..2865c9d9af 100644
--- a/openpilot/starpilot/models/tests/test_modeld_startup_fallback.py
+++ b/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
diff --git a/openpilot/starpilot/system/jetlink/daemon.py b/openpilot/starpilot/system/jetlink/daemon.py
deleted file mode 100644
index 9f0779cd2b..0000000000
--- a/openpilot/starpilot/system/jetlink/daemon.py
+++ /dev/null
@@ -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()
diff --git a/openpilot/starpilot/system/jetlink/lifecycle.py b/openpilot/starpilot/system/jetlink/lifecycle.py
index 4c71dca138..fecdce915d 100644
--- a/openpilot/starpilot/system/jetlink/lifecycle.py
+++ b/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)
diff --git a/openpilot/starpilot/tests/test_jetlink_lifecycle.py b/openpilot/starpilot/tests/test_jetlink_lifecycle.py
index 8facdd4272..9b4a52fce3 100644
--- a/openpilot/starpilot/tests/test_jetlink_lifecycle.py
+++ b/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
diff --git a/openpilot/system/manager/process_config.py b/openpilot/system/manager/process_config.py
index f408555752..b69338fefe 100644
--- a/openpilot/system/manager/process_config.py
+++ b/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.