diff --git a/selfdrive/car/tests/fleet_safety_core.py b/selfdrive/car/tests/fleet_safety_core.py
new file mode 100644
index 0000000000..8f36e1c23b
--- /dev/null
+++ b/selfdrive/car/tests/fleet_safety_core.py
@@ -0,0 +1,197 @@
+"""Offline emitted-CAN audit primitives; not a vehicle or selfdrived acceptance test."""
+
+from collections import Counter
+from dataclasses import dataclass
+
+
+def _integer(value, name, low, high):
+  if isinstance(value, bool) or not isinstance(value, int) or not low <= value <= high:
+    raise ValueError(f"{name} must be an integer in [{low}, {high}]")
+  return value
+
+
+@dataclass(frozen=True)
+class EffectiveSafetyConfig:
+  panda_index: int
+  safety_model: int
+  safety_param: int
+  alternative_experience: int
+
+  def __post_init__(self):
+    _integer(self.panda_index, "panda_index", 0, 63)
+    for name in ("safety_model", "safety_param", "alternative_experience"):
+      _integer(getattr(self, name), name, 0, 65535)
+
+  def local_bus(self, global_bus):
+    """Match pandad's bus-offset routing, without modulo aliasing another panda."""
+    _integer(global_bus, "global_bus", 0, 255)
+    offset = self.panda_index * 4
+    if not offset <= global_bus < offset + 4:
+      raise ValueError(f"global bus {global_bus} does not belong to panda {self.panda_index}")
+    return global_bus - offset
+
+
+def effective_safety_configs(CP, FPCP, *, panda_count=None):
+  """Mirror PandaSafety::setSafetyMode; pass the actual panda count when known.
+
+  Without that count, represent all configured pandas (at least one). Additional
+  physical pandas are SILENT, but still receive indexed StarPilot params and the
+  combined alternativeExperience. StarPilot never selects a safety model here.
+  """
+  configs, extra = tuple(CP.safetyConfigs), tuple(FPCP.safetyConfigs)
+  count = max(len(configs), len(extra), 1) if panda_count is None else panda_count
+  _integer(count, "panda_count", 0, 64)
+  # Both capnp fields are Int16, assigned/ORed into a uint16_t in pandad.
+  alternative = (_integer(CP.alternativeExperience, "CP.alternativeExperience", -32768, 32767) & 65535)
+  alternative |= _integer(FPCP.alternativeExperience, "FPCP.alternativeExperience", -32768, 32767) & 65535
+  result = []
+  for i in range(count):
+    model, param = 0, 0  # cereal::CarParams::SafetyModel::SILENT
+    if i < len(configs):
+      model = getattr(configs[i].safetyModel, "raw", configs[i].safetyModel)
+      param = _integer(configs[i].safetyParam, "CP.safetyParam", 0, 65535)
+    if i < len(extra):
+      param |= _integer(extra[i].safetyParam, "FPCP.safetyParam", 0, 65535)
+    result.append(EffectiveSafetyConfig(i, model, param, alternative))
+  return tuple(result)
+
+
+@dataclass(frozen=True)
+class Permissions:
+  controls_allowed: bool
+  aol_allowed: bool
+  longitudinal_allowed: bool
+
+  @property
+  def active(self):
+    return self.controls_allowed or self.aol_allowed or self.longitudinal_allowed
+
+  @property
+  def label(self):
+    return "+".join(name for name, value in (("controls", self.controls_allowed), ("aol", self.aol_allowed),
+                                           ("long", self.longitudinal_allowed)) if value) or "inactive"
+
+
+@dataclass(frozen=True)
+class TxRecord:
+  addr: int
+  bus: int
+  local_bus: int
+  data: bytes
+  timestamp_ns: int
+  scenario: str
+  permissions: Permissions
+  requested_lat_active: bool | None
+  requested_long_active: bool | None
+  accepted: bool | None
+  expected_block_reason: str | None
+  failure: str | None
+
+  @property
+  def requested_mode(self):
+    # Retain partially unknown states; do not turn unknown into inactive.
+    values = {True: "active", False: "inactive", None: "unknown"}
+    return f"lat:{values[self.requested_lat_active]},long:{values[self.requested_long_active]}"
+
+  @property
+  def requested_active(self):
+    return self.requested_lat_active is True or self.requested_long_active is True
+
+
+class TxAudit:
+  """One safety instance per panda. Caller supplies recorded RX and clock updates.
+
+  check() records rejected packets instead of raising so every emitted packet can
+  be audited. The caller must consume summary()['status']; 'uncovered' is not a
+  pass. Hook/factory exceptions are recorded and re-raised. Expected negative
+  cases require a reason on each packet and fail if the hook accepts that packet.
+  """
+
+  def __init__(self, safety, packet_factory, *, config):
+    self.safety, self.packet_factory, self.config = safety, packet_factory, config
+    self._records = []
+
+  @property
+  def records(self):
+    return tuple(self._records)
+
+  def check(self, addr, bus, data, timestamp_ns, scenario, *, expected_block_reason=None,
+            requested_lat_active=None, requested_long_active=None):
+    _integer(addr, "addr", 0, 0x1FFFFFFF)
+    _integer(timestamp_ns, "timestamp_ns", 0, 2**64 - 1)
+    local_bus = self.config.local_bus(bus)
+    if not isinstance(scenario, str) or not scenario.strip():
+      raise ValueError("scenario must be nonempty")
+    if expected_block_reason is not None and (not isinstance(expected_block_reason, str) or not expected_block_reason.strip()):
+      raise ValueError("expected_block_reason must be a nonempty per-packet reason")
+    for value in (requested_lat_active, requested_long_active):
+      if value is not None and type(value) is not bool:
+        raise ValueError("requested activity must be bool or None")
+    if not isinstance(data, (bytes, bytearray, memoryview)):
+      raise ValueError("data must be bytes-like")
+    payload = bytes(data)
+    permissions = Permissions(bool(self.safety.get_controls_allowed()), bool(self.safety.get_aol_allowed()),
+                              bool(self.safety.get_longitudinal_allowed()))
+    accepted, failure = None, None
+    try:
+      accepted = bool(self.safety.safety_tx_hook(self.packet_factory(addr, local_bus, payload)))
+      if expected_block_reason is None and not accepted:
+        failure = "unexpected rejection"
+      elif expected_block_reason is not None and accepted:
+        failure = "expected block was accepted"
+    except Exception as exc:
+      failure = f"hook/factory error: {type(exc).__name__}: {exc}"
+      raise
+    finally:
+      self._records.append(TxRecord(addr, bus, local_bus, payload, timestamp_ns, scenario, permissions,
+                                   requested_lat_active, requested_long_active, accepted, expected_block_reason, failure))
+    return self._records[-1]
+
+  def summary(self, *, failure_limit=20):
+    _integer(failure_limit, "failure_limit", 0, 2**31 - 1)
+    records = self._records
+    active = sum(r.permissions.active for r in records)
+    requested_active = sum(r.requested_active for r in records)
+    accepted_active = sum(r.accepted is True and r.permissions.active and r.requested_active for r in records)
+    missing = []
+    if not records:
+      missing.append("no emitted TX")
+    if not active:
+      missing.append("no TX with active safety permission")
+    if not requested_active:
+      missing.append("no TX with explicitly active command request")
+    if not accepted_active:
+      missing.append("no accepted TX with active command request and safety permission")
+    failure_count = sum(r.failure is not None for r in records)
+    failures = []
+    for i, r in enumerate(records):
+      if r.failure and len(failures) < failure_limit:
+        failures.append({"index": i, "addr": r.addr, "bus": r.bus, "data_hex": r.data.hex(), "timestamp_ns": r.timestamp_ns,
+                         "scenario": r.scenario, "reason": r.failure})
+    permissions = [r.permissions.label for r in records]
+    requested = [r.requested_mode for r in records]
+
+    def transitions(states):
+      return dict(Counter(f"{a} -> {b}" for a, b in zip(states, states[1:], strict=False) if a != b))
+
+    return {
+      "status": "failed" if failure_count else "uncovered" if missing else "pass",
+      "tx_total": len(records), "tx_accepted": sum(r.accepted is True for r in records),
+      "tx_rejected": sum(r.accepted is False for r in records),
+      "expected_blocks": sum(r.accepted is False and r.expected_block_reason is not None for r in records),
+      "unexpected_rejections": sum(r.accepted is False and r.expected_block_reason is None for r in records),
+      "unexpected_acceptances": sum(r.accepted is True and r.expected_block_reason is not None for r in records),
+      "hook_errors": sum(r.accepted is None for r in records),
+      "active_tx": active, "inactive_tx": len(records) - active,
+      "accepted_active_tx": accepted_active,
+      "aol_only_tx": sum(r.permissions.aol_allowed and not r.permissions.controls_allowed for r in records),
+      "requested_active_tx": requested_active,
+      "requested_inactive_tx": sum(r.requested_lat_active is False and r.requested_long_active is False for r in records),
+      "requested_unknown_tx": sum(r.requested_lat_active is None or r.requested_long_active is None for r in records),
+      "requested_aol_only_tx": sum(r.requested_lat_active is True and r.permissions.aol_allowed and
+                                   not r.permissions.controls_allowed for r in records),
+      "permission_states": dict(Counter(permissions)), "requested_states": dict(Counter(requested)),
+      "permission_transitions": transitions(permissions), "requested_transitions": transitions(requested),
+      "uncovered_reasons": missing, "failure_count": failure_count, "failures": failures,
+      "failures_truncated": failure_count - len(failures),
+    }
diff --git a/selfdrive/car/tests/test_fleet_safety_core.py b/selfdrive/car/tests/test_fleet_safety_core.py
new file mode 100644
index 0000000000..bd14f042f9
--- /dev/null
+++ b/selfdrive/car/tests/test_fleet_safety_core.py
@@ -0,0 +1,217 @@
+import json
+import unittest  # noqa: TID251 - Deliberately stdlib-only offline tests.
+from dataclasses import FrozenInstanceError
+from types import SimpleNamespace
+
+from selfdrive.car.tests.fleet_safety_core import EffectiveSafetyConfig, TxAudit, effective_safety_configs  # noqa: TID251
+
+
+def params(configs=(), alternative=0):
+  return SimpleNamespace(safetyConfigs=list(configs), alternativeExperience=alternative)
+
+
+def config(model, param):
+  return SimpleNamespace(safetyModel=model, safetyParam=param)
+
+
+class FakeSafety:
+  def __init__(self, controls=False, aol=False, longitudinal=False, accepted=True, hook=None):
+    self.controls, self.aol, self.longitudinal = controls, aol, longitudinal
+    self.accepted, self.hook = accepted, hook
+    self.packets = []
+
+  def get_controls_allowed(self):
+    return self.controls
+
+  def get_aol_allowed(self):
+    return self.aol
+
+  def get_longitudinal_allowed(self):
+    return self.longitudinal
+
+  def safety_tx_hook(self, packet):
+    self.packets.append(packet)
+    if self.hook:
+      self.hook(self)
+    return self.accepted
+
+
+def audit(safety=None, panda_index=0):
+  return TxAudit(safety or FakeSafety(), lambda addr, bus, data: SimpleNamespace(addr=addr, bus=bus, data=data),
+                 config=EffectiveSafetyConfig(panda_index, 2, 0, 0))
+
+
+def check(tx, **kwargs):
+  values = dict(addr=0x123, bus=0, data=b"\x01\x02", timestamp_ns=123456, scenario="recorded engagement",
+                requested_lat_active=True, requested_long_active=False)
+  values.update(kwargs)
+  return tx.check(**values)
+
+
+class TestEffectiveSafetyConfigs(unittest.TestCase):
+  def test_panda_models_params_and_alternative_bits(self):
+    cp = params([config(SimpleNamespace(raw=5), 0x01), config(17, 0x08)], 0x01)
+    fpcp = params([config(99, 0x10), config(98, 0x20), config(97, 0x40)], 0x04)
+    result = effective_safety_configs(cp, fpcp, panda_count=4)
+    self.assertEqual(result, (EffectiveSafetyConfig(0, 5, 0x11, 5), EffectiveSafetyConfig(1, 17, 0x28, 5),
+                              EffectiveSafetyConfig(2, 0, 0x40, 5), EffectiveSafetyConfig(3, 0, 0, 5)))
+    with self.assertRaises(FrozenInstanceError):
+      result[0].safety_model = 99
+
+  def test_signed_capnp_alt_matches_uint16_assignment(self):
+    self.assertEqual(effective_safety_configs(params(alternative=-32768), params(alternative=1))[0].alternative_experience, 32769)
+
+  def test_default_count_and_actual_count(self):
+    self.assertEqual(len(effective_safety_configs(params(), params())), 1)
+    self.assertEqual(len(effective_safety_configs(params([config(5, 0)]), params([config(0, 0)] * 3))), 3)
+    self.assertEqual(effective_safety_configs(params(), params(), panda_count=0), ())
+    self.assertEqual(len(effective_safety_configs(params([config(5, 0)] * 2), params(), panda_count=1)), 1)
+
+  def test_global_bus_never_aliases_another_panda(self):
+    second = EffectiveSafetyConfig(1, 5, 0, 0)
+    self.assertEqual(second.local_bus(4), 0)
+    self.assertEqual(second.local_bus(7), 3)
+    for bus in (0, 3, 8, 128, -1, 256, True):
+      with self.subTest(bus=bus), self.assertRaises(ValueError):
+        second.local_bus(bus)
+
+  def test_invalid_config_not_silently_truncated(self):
+    for cp, fp, kwargs in ((params([config(1, 65536)]), params(), {}),
+                           (params(), params([config(1, -1)]), {}),
+                           (params(alternative=65535), params(), {}),
+                           (params(), params(), {"panda_count": -1})):
+      with self.subTest(cp=cp, fp=fp, kwargs=kwargs), self.assertRaises(ValueError):
+        effective_safety_configs(cp, fp, **kwargs)
+
+
+class TestTxAudit(unittest.TestCase):
+  def test_aol_rejection_fails_without_controls(self):
+    tx = audit(FakeSafety(aol=True, accepted=False))
+    record = check(tx)
+    self.assertFalse(record.permissions.controls_allowed)
+    self.assertTrue(record.permissions.aol_allowed)
+    self.assertEqual(tx.summary()["status"], "failed")
+    self.assertEqual(tx.summary()["unexpected_rejections"], 1)
+    self.assertEqual(tx.summary()["requested_aol_only_tx"], 1)
+
+  def test_permissions_are_captured_before_mutating_hook(self):
+    def clear(safety):
+      safety.controls = safety.aol = safety.longitudinal = False
+    safety = FakeSafety(controls=True, aol=True, longitudinal=True, accepted=False, hook=clear)
+    tx = audit(safety)
+    record = check(tx)
+    self.assertEqual((record.permissions.controls_allowed, record.permissions.aol_allowed, record.permissions.longitudinal_allowed),
+                     (True, True, True))
+    self.assertFalse(safety.controls)
+    self.assertEqual(tx.summary()["active_tx"], 1)
+    self.assertEqual(tx.summary()["status"], "failed")
+
+  def test_inactive_rejection_also_fails(self):
+    tx = audit(FakeSafety(accepted=False))
+    check(tx, requested_lat_active=False)
+    self.assertEqual(tx.summary()["status"], "failed")
+    self.assertEqual(tx.summary()["inactive_tx"], 1)
+
+  def test_expected_blocks_are_per_packet_and_keep_reason(self):
+    safety = FakeSafety(aol=True, accepted=False)
+    tx = audit(safety)
+    first = check(tx, expected_block_reason="negative case: excessive torque")
+    self.assertEqual(first.expected_block_reason, "negative case: excessive torque")
+    self.assertIsNone(first.failure)
+    self.assertEqual(tx.summary()["expected_blocks"], 1)
+    self.assertEqual(tx.summary()["status"], "uncovered")
+    check(tx)  # Previous expectation must not leak onto another packet.
+    self.assertEqual(tx.summary()["unexpected_rejections"], 1)
+    self.assertEqual(tx.summary()["status"], "failed")
+
+  def test_expected_block_acceptance_fails(self):
+    tx = audit(FakeSafety(controls=True))
+    check(tx, expected_block_reason="negative case: wrong command bit")
+    self.assertEqual(tx.summary()["status"], "failed")
+    self.assertEqual(tx.summary()["unexpected_acceptances"], 1)
+
+  def test_failure_summary_is_bounded_without_hiding_verdict(self):
+    tx = audit(FakeSafety(aol=True, accepted=False))
+    for _ in range(3):
+      check(tx)
+    summary = tx.summary(failure_limit=1)
+    self.assertEqual(summary["failure_count"], 3)
+    self.assertEqual(summary["failures_truncated"], 2)
+    self.assertEqual(len(summary["failures"]), 1)
+    self.assertEqual(len(tx.records), 3)
+    self.assertEqual(tx.summary(failure_limit=0)["status"], "failed")
+
+  def test_missing_coverage_is_not_a_pass(self):
+    tx = audit(FakeSafety())
+    self.assertEqual(tx.summary()["status"], "uncovered")
+    self.assertIn("no emitted TX", tx.summary()["uncovered_reasons"])
+    check(tx)
+    self.assertEqual(tx.summary()["status"], "uncovered")
+    tx = audit(FakeSafety(controls=True))
+    check(tx, requested_lat_active=None, requested_long_active=None)
+    self.assertEqual(tx.summary()["status"], "uncovered")
+    self.assertEqual(tx.summary()["requested_unknown_tx"], 1)
+    check(tx, requested_lat_active=False)
+    self.assertEqual(tx.summary()["status"], "uncovered")
+    check(tx)
+    self.assertEqual(tx.summary()["status"], "pass")
+
+  def test_states_requests_and_transitions_are_independent(self):
+    safety = FakeSafety()
+    tx = audit(safety)
+    check(tx, requested_lat_active=False)
+    safety.aol = True
+    check(tx)
+    check(tx)
+    safety.controls = safety.longitudinal = True
+    check(tx, requested_long_active=True)
+    safety.aol = safety.controls = safety.longitudinal = False
+    check(tx, requested_lat_active=False)
+    summary = tx.summary()
+    self.assertEqual(summary["tx_total"], 5)
+    self.assertEqual(summary["active_tx"], 3)
+    self.assertEqual(summary["inactive_tx"], 2)
+    self.assertEqual(summary["requested_active_tx"], 3)
+    self.assertEqual(summary["requested_inactive_tx"], 2)
+    self.assertEqual(summary["requested_aol_only_tx"], 2)
+    self.assertEqual(summary["permission_transitions"], {"inactive -> aol": 1, "aol -> controls+aol+long": 1,
+                                                         "controls+aol+long -> inactive": 1})
+    self.assertEqual(sum(summary["requested_transitions"].values()), 3)
+    json.dumps(summary)  # Reports can be serialized without compiled dependencies.
+
+  def test_every_packet_keeps_immutable_original_metadata(self):
+    safety = FakeSafety(longitudinal=True)
+    tx = audit(safety, panda_index=1)
+    data = bytearray(b"\x00\xfe")
+    record = check(tx, bus=6, data=data, requested_lat_active=False, requested_long_active=True)
+    data[0] = 5
+    self.assertEqual((record.addr, record.bus, record.local_bus, record.data, record.timestamp_ns, record.scenario),
+                     (0x123, 6, 2, b"\x00\xfe", 123456, "recorded engagement"))
+    self.assertEqual(safety.packets[0].bus, 2)
+    self.assertIsInstance(tx.records, tuple)
+    with self.assertRaises(FrozenInstanceError):
+      record.accepted = False
+    self.assertEqual(tx.summary()["status"], "pass")
+
+  def test_invalid_per_packet_inputs_rejected(self):
+    for kwargs in ({"expected_block_reason": " "}, {"expected_block_reason": True}, {"requested_lat_active": 1},
+                   {"scenario": ""}, {"timestamp_ns": -1}, {"data": 8}, {"addr": -1}, {"bus": 4}):
+      tx = audit()
+      with self.subTest(kwargs=kwargs), self.assertRaises(ValueError):
+        check(tx, **kwargs)
+      self.assertEqual(tx.records, ())
+
+  def test_hook_exception_is_recorded_and_reraised(self):
+    def broken(_safety):
+      raise RuntimeError("hook crashed")
+    tx = audit(FakeSafety(aol=True, hook=broken))
+    with self.assertRaisesRegex(RuntimeError, "hook crashed"):
+      check(tx)
+    self.assertEqual(tx.summary()["status"], "failed")
+    self.assertEqual(tx.summary()["hook_errors"], 1)
+    self.assertEqual(tx.records[0].data, b"\x01\x02")
+    self.assertTrue(tx.records[0].permissions.aol_allowed)
+
+
+if __name__ == "__main__":
+  unittest.main()