Distribute custom pytest workloads without dropping cases

.github/workflows/tests.yaml

@@ -203,13 +203,6 @@ jobs:
         RAYLIB_BACKEND: headless
         STARPILOT_TEST_DIAGNOSTICS: ci-results/host-test-diagnostics/unittest
       run: tools/op.sh test --json-output unit-test-results.json
-    - name: Test custom module-level pytest cases
-      if: ${{ !cancelled() && steps.build.outcome == 'success' }}
-      timeout-minutes: 15
-      env:
-        RAYLIB_BACKEND: headless
-        STARPILOT_TEST_DIAGNOSTICS: ci-results/host-test-diagnostics/pytest
-      run: python3 tools/ci/run_host_tests.py --pytest
     - name: Test GM vehicle configurations and CAN behavior
       if: ${{ !cancelled() && steps.build.outcome == 'success' }}
       timeout-minutes: 5
@@ -244,6 +237,80 @@ jobs:
           ci-results/host-test-diagnostics/
         if-no-files-found: error
 
+  custom_pytest:
+    name: custom pytest (${{ matrix.shard }})
+    runs-on: ubuntu-24.04
+    timeout-minutes: 35
+    permissions:
+      contents: read
+    strategy:
+      fail-fast: false
+      matrix:
+        shard: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23]
+    steps:
+    - uses: actions/checkout@v7
+    - name: Provide JavaScript runtime for Galaxy contract tests
+      uses: actions/setup-node@v7
+      with:
+        node-version: '24'
+        package-manager-cache: false
+    - name: Provide pinned map provider toolchain
+      uses: actions/setup-go@v7
+      with:
+        go-version: '1.25.1'
+        cache-dependency-path: mapd_repo/go.sum
+    - name: Fetch pinned map provider dependencies
+      working-directory: mapd_repo
+      env:
+        GOTOOLCHAIN: local
+      run: go mod download
+    - run: ./tools/op.sh setup
+      id: setup
+    - name: Build openpilot
+      id: build
+      run: scons
+    - name: Provide H.264 encoder for Android Auto integration tests
+      if: ${{ !cancelled() && steps.build.outcome == 'success' }}
+      run: |
+        uv pip install --python .venv/bin/python --only-binary av av==19.0.1
+        .venv/bin/python -c "import av; assert 'libx264' in av.codecs_available"
+    - name: Test every assigned custom pytest node serially
+      if: ${{ !cancelled() && steps.build.outcome == 'success' }}
+      timeout-minutes: 15
+      env:
+        RAYLIB_BACKEND: headless
+      run: .venv/bin/python -m tools.ci.custom_pytest_shards run --index "${{ matrix.shard }}" --count 24 --output "ci-results/custom-pytest-${{ matrix.shard }}"
+    - name: Save collection, every outcome, JUnit and durable worker diagnostics
+      if: always()
+      uses: actions/upload-artifact@v4
+      with:
+        name: custom-pytest-${{ matrix.shard }}
+        path: ci-results/custom-pytest-${{ matrix.shard }}/
+        if-no-files-found: error
+
+  custom_pytest_complete:
+    name: complete custom pytest catalog
+    needs: custom_pytest
+    if: always()
+    runs-on: ubuntu-24.04
+    permissions:
+      contents: read
+    steps:
+    - uses: actions/checkout@v7
+    - uses: actions/download-artifact@v4
+      with:
+        pattern: custom-pytest-*
+        path: ci-results/custom-pytest-shards
+    - name: Require every case passed or audited capability-unavailable
+      run: python3 -m tools.ci.custom_pytest_shards aggregate --directory ci-results/custom-pytest-shards --count 24 --output ci-results/custom-pytest-complete.json
+    - name: Save complete custom catalog verdict
+      if: always()
+      uses: actions/upload-artifact@v4
+      with:
+        name: custom-pytest-complete
+        path: ci-results/custom-pytest-complete.json
+        if-no-files-found: error
+
   aol_lifecycles:
     name: AOL lifecycles (${{ matrix.shard }})
     runs-on: ubuntu-24.04

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "08228676cd7e8748025e8485f6a751f470d936973d5bf72540a5df5aff3371fc", "schema_version": 1}
+{"protected_inventory_sha256": "dd1da18910ae99b2a3568583cf26b0cf205fc07012c08c24d59e63cddc9e3838", "schema_version": 1}

tools/ci/custom_pytest_shards.py

@@ -0,0 +1,421 @@
+"""Run serial custom pytest shards and require exact passed or audited-unavailable outcomes."""
+import argparse
+from collections import Counter, defaultdict
+import hashlib
+import importlib
+import json
+import math
+import os
+from pathlib import Path
+import subprocess
+import sys
+from xml.etree import ElementTree
+
+from tools.ci.lifecycle_shards import build_identity, same_build_inputs
+
+ROOT = Path(__file__).resolve().parents[2]
+HEAVY_MODULES = {
+  'openpilot/starpilot/car/toyota/tests/test_legacy_advert_callers.py',
+  'openpilot/starpilot/car/toyota/tests/test_camera_2aa_callers.py',
+}
+
+CAPABILITY_NODES = {
+  'openpilot/starpilot/galaxy_bluetooth/vendor/ifaddr/test_ifaddr.py::test_netifaces_compatibility',
+  'openpilot/starpilot/system/android_auto/tests/test_apk_identity.py::test_real_android_auto_17_6_xapk',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[False-32-18-0-0]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[False-32-18-4-2]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[False-32-20-3-5]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[False-40-18-0-6]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[True-32-18-0-0]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[True-32-18-4-2]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[True-32-20-3-5]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_composed_car_frame_converts_to_the_same_picture[True-40-18-0-6]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[False-32-18-0-0]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[False-32-18-4-2]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[False-32-20-3-5]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[True-32-18-0-0]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[True-32-18-4-2]',
+  'openpilot/starpilot/system/android_auto/tests/test_gpu_nv12.py::test_fused_projection_conversion_matches_rgba_composition[True-32-20-3-5]',
+}
+
+
+def checksum(data):
+  return hashlib.sha256(data).hexdigest()
+
+
+def write_json(path, value):
+  path = Path(path)
+  path.parent.mkdir(parents=True, exist_ok=True)
+  temporary = path.with_suffix(path.suffix + '.tmp')
+  temporary.write_text(json.dumps(value, sort_keys=True, indent=2) + '\n')
+  temporary.replace(path)
+
+
+def policy():
+  value = json.loads(Path(__file__).with_name('custom_pytest_workloads.json').read_text())
+  if (value['version'] != 1 or set(value['independent_functions']) != HEAVY_MODULES or
+      set(value['capability_skips']) != CAPABILITY_NODES):
+    raise ValueError('Unreviewed custom pytest grouping policy')
+  if type(value['independent_max_cases']) is not int or not 1 <= value['independent_max_cases'] <= 60:
+    raise ValueError('Invalid independent parameter group bound')
+  numbers = [value['unknown_seconds_per_case'], value['independent_max_seconds']]
+  numbers.extend(row['seconds_per_case'] for row in value['independent_functions'].values())
+  numbers.extend(row['seconds'] for row in value['modules'].values())
+  if any(type(n) not in (int, float) or not math.isfinite(n) or n <= 0 for n in numbers):
+    raise ValueError('Invalid custom pytest workload estimate')
+  if any(type(row['cases']) is not int or row['cases'] < 1 or len(row['blob']) != 40 for row in value['modules'].values()):
+    raise ValueError('Invalid measured custom pytest workload')
+  return value
+
+
+def blob_id(path):
+  data = path.read_bytes()
+  return hashlib.sha1(f'blob {len(data)}\0'.encode() + data).hexdigest()
+
+
+def partition(catalog, count, root=ROOT):
+  """Whole modules, except explicitly reviewed independent heavy parameter cases."""
+  if type(count) is not int or not 1 <= count <= 24 or not catalog or len(catalog) != len(set(catalog)):
+    raise ValueError('Empty, duplicate or invalid custom pytest catalog')
+  grouped = defaultdict(list)
+  for node in catalog:
+    if not isinstance(node, str) or '::' not in node:
+      raise ValueError('Invalid custom pytest node ID')
+    module = node.split('::', 1)[0]
+    if not module.endswith('.py') or Path(module).is_absolute() or '..' in Path(module).parts:
+      raise ValueError('Invalid custom pytest module path')
+    grouped[module].append(node)
+  config = policy()
+  units = []
+  for module, nodes in sorted(grouped.items()):
+    if module in config['independent_functions']:
+      rate = config['independent_functions'][module]['seconds_per_case']
+      functions = defaultdict(list)
+      for node in nodes:
+        function = node.split('::', 1)[1].split('[', 1)[0]
+        if not function.isidentifier() or not function.startswith('test_'):
+          raise ValueError('Heavy grouping requires reviewed top-level pytest functions')
+        functions[function].append(node)
+      limit = min(config['independent_max_cases'], max(1, int(config['independent_max_seconds'] // rate)))
+      for function, function_nodes in sorted(functions.items()):
+        for start in range(0, len(function_nodes), limit):
+          selected = sorted(function_nodes)[start:start + limit]
+          units.append((len(selected) * rate, module + '::' + function + ':' + str(start), selected))
+    else:
+      rate = config['unknown_seconds_per_case']
+      measured = config['modules'].get(module)
+      if measured and measured['cases'] == len(nodes) and blob_id(Path(root) / module) == measured['blob']:
+        rate = measured['seconds'] / measured['cases']
+      units.append((rate * len(nodes), module, nodes))
+  selected, loads = [set() for _ in range(count)], [0.0] * count
+  for cost, _name, nodes in sorted(units, key=lambda unit: (-unit[0], unit[1])):
+    index = min(range(count), key=lambda i: (loads[i], i))
+    selected[index].update(nodes)
+    loads[index] += cost
+  return [[node for node in catalog if node in shard] for shard in selected], loads
+
+
+def source_snapshot(root=ROOT):
+  root = Path(root)
+  return {'build': build_identity(root),
+          'runners': {name: checksum((root / 'tools/ci' / name).read_bytes()) for name in (
+            'run_host_tests.py', 'custom_pytest_shards.py', 'custom_pytest_workloads.json',
+            'pytest_progress.py', 'worker_diagnostics.py', 'lifecycle_shards.py', 'lifecycle_build_outputs.json')}}
+
+
+REAL_XAPK = 'openpilot/.cache/android_auto/android-auto-17-6-663454-release.xapk'
+
+
+def guard_observation(guard, root):
+  if guard == 'gpu_env':
+    return {'STARPILOT_AUTO_GL_TEST': os.environ.get('STARPILOT_AUTO_GL_TEST')}
+  if guard == 'real_xapk':
+    return {'path': REAL_XAPK, 'exists': (Path(root) / REAL_XAPK).exists()}
+  if guard == 'netifaces':
+    try:
+      importlib.import_module('netifaces')
+    except ImportError:
+      return {'importable': False}
+    return {'importable': True}
+  raise ValueError('Unreviewed capability guard')
+
+
+def guard_unavailable(guard, observation):
+  if guard == 'gpu_env' and set(observation) == {'STARPILOT_AUTO_GL_TEST'}:
+    value = observation['STARPILOT_AUTO_GL_TEST']
+    if value is None or isinstance(value, str):
+      return value != '1'
+  if (guard == 'real_xapk' and set(observation) == {'path', 'exists'} and
+      observation['path'] == REAL_XAPK and type(observation['exists']) is bool):
+    return not observation['exists']
+  if guard == 'netifaces' and set(observation) == {'importable'} and type(observation['importable']) is bool:
+    return not observation['importable']
+  raise ValueError('Forged capability condition evidence')
+
+
+def capability_state(selected, root=ROOT):
+  inventory = policy()['capability_skips']
+  state = {}
+  for node in selected:
+    if node in inventory:
+      row = inventory[node]
+      state[node] = {'source_blob': blob_id(Path(root) / node.split('::')[0]),
+                     'guard': row['guard'], 'observation': guard_observation(row['guard'], root)}
+  return state
+
+
+def validate_capabilities(selected, state, root=ROOT):
+  inventory = policy()['capability_skips']
+  if set(state) != set(selected) & set(inventory):
+    raise ValueError('Missing or forged audited capability nodes')
+  for node, record in state.items():
+    row = inventory[node]
+    if set(record) != {'source_blob', 'guard', 'observation'} or record['guard'] != row['guard']:
+      raise ValueError('Forged audited capability guard')
+    if record['source_blob'] != blob_id(Path(root) / node.split('::')[0]):
+      raise ValueError('Audited capability source differs from checkout')
+    guard_unavailable(record['guard'], record['observation'])
+
+
+def validate_outcomes(selected, reports, capabilities=None):
+  capabilities = {} if capabilities is None else capabilities
+  inventory = policy()['capability_skips']
+  if not selected or set(reports) != set(selected):
+    raise ValueError('Missing or unexpected custom pytest outcomes')
+  skipped = []
+  for node in selected:
+    phases = reports[node]
+    if (Counter(row['phase'] for row in phases) == Counter({'setup': 1, 'call': 1, 'teardown': 1}) and
+        all(row['status'] == 'passed' for row in phases)):
+      continue
+    row = inventory.get(node)
+    record = capabilities.get(node)
+    if (row is None or record is None or record['source_blob'] != row['source_blob'] or
+        record['guard'] != row['guard'] or not guard_unavailable(record['guard'], record['observation'])):
+      raise ValueError('Failed, skipped or erroneous nonexempt custom pytest node')
+    if Counter(report['phase'] for report in phases) != Counter({'setup': 1, 'teardown': 1}):
+      raise ValueError('Audited capability skip has wrong phases')
+    setup = next(report for report in phases if report['phase'] == 'setup')
+    teardown = next(report for report in phases if report['phase'] == 'teardown')
+    if setup['status'] != 'skipped' or setup.get('reason') != row['reason'] or teardown['status'] != 'passed':
+      raise ValueError('Audited capability skip has wrong reason or outcome')
+    skipped.append({'id': node, 'status': 'skipped', 'disposition': 'capability-unavailable',
+                    'reason': row['reason'], 'source_blob': row['source_blob'], 'guard': row['guard'],
+                    'observation': record['observation']})
+  return skipped
+
+
+
+def junit_address(node):
+  path, bracket, parameters = node.partition('[')
+  names = path.split('::')
+  names[0] = names[0].replace('/', '.').removesuffix('.py')
+  names[-1] += bracket + parameters
+  return '.'.join(names[:-1]), names[-1]
+
+
+def validate_junit(path, selected, skipped):
+  cases = ElementTree.parse(path).getroot().findall('.//testcase')
+  expected = {junit_address(node): node for node in selected}
+  if len(expected) != len(selected) or Counter((row.get('classname'), row.get('name')) for row in cases) != Counter(expected.keys()):
+    raise ValueError('JUnit must contain every selected node exactly once')
+  unavailable = {row['id']: row for row in skipped}
+  for case in cases:
+    node = expected[(case.get('classname'), case.get('name'))]
+    if case.find('failure') is not None or case.find('error') is not None:
+      raise ValueError('JUnit contains failed or erroneous custom pytest case')
+    skips = case.findall('skipped')
+    if node not in unavailable:
+      if skips:
+        raise ValueError('JUnit contains an unaudited skip')
+    elif (len(skips) != 1 or skips[0].get('type') != 'pytest.skip' or
+          skips[0].get('message') != unavailable[node]['reason'].removeprefix('Skipped: ')):
+      raise ValueError('JUnit audited capability reason differs from actual phase reports')
+
+
+# Explicit pytest plugin. Selection happens after real collection, never from an invented node list.
+def pytest_addoption(parser):
+  group = parser.getgroup('custom-pytest-shards')
+  group.addoption('--custom-shard-index', type=int)
+  group.addoption('--custom-shard-count', type=int)
+  group.addoption('--custom-shard-output')
+
+
+def pytest_configure(config):
+  output = config.getoption('--custom-shard-output')
+  if output:
+    plugin = ShardPlugin(config)
+    config.pluginmanager.register(plugin, 'custom-pytest-shard-evidence')
+
+
+class ShardPlugin:
+  def __init__(self, config):
+    self.index = config.getoption('--custom-shard-index')
+    self.count = config.getoption('--custom-shard-count')
+    self.output = Path(config.getoption('--custom-shard-output'))
+    self.errors = []
+    self.catalog = []
+    self.selected = []
+    self.reports = defaultdict(list)
+    self.capabilities = {}
+
+  def pytest_collectreport(self, report):
+    if report.failed or report.skipped:
+      self.errors.append({'id': report.nodeid, 'failure': str(report.longrepr)[:4096]})
+
+  def pytest_collection_modifyitems(self, session, config, items):
+    self.catalog = [item.nodeid for item in items]
+    assignment, loads = partition(self.catalog, self.count)
+    if type(self.index) is not int or not 0 <= self.index < self.count:
+      raise ValueError('Invalid custom pytest shard index')
+    self.selected = assignment[self.index]
+    self.capabilities = capability_state(self.selected)
+    selected = set(self.selected)
+    rejected = [item for item in items if item.nodeid not in selected]
+    items[:] = [item for item in items if item.nodeid in selected]
+    config.hook.pytest_deselected(items=rejected)
+    write_json(self.output / 'plan.json', {'catalog': self.catalog, 'selected': self.selected,
+               'index': self.index, 'count': self.count, 'estimated_seconds': loads[self.index],
+               'estimate_only': True})
+
+  def pytest_runtest_logreport(self, report):
+    row = {'phase': report.when, 'status': report.outcome, 'duration': report.duration}
+    if report.skipped:
+      row['reason'] = str(report.longrepr[2]) if isinstance(report.longrepr, tuple) and len(report.longrepr) == 3 else str(report.longrepr)
+    self.reports[report.nodeid].append(row)
+
+  def pytest_sessionfinish(self, session, exitstatus):
+    errors = list(self.errors)
+    skipped = []
+    try:
+      validate_capabilities(self.selected, self.capabilities)
+      skipped = validate_outcomes(self.selected, self.reports, self.capabilities)
+    except ValueError as exc:
+      errors.append({'outcomes': str(exc)})
+    value = {'catalog': self.catalog, 'selected': self.selected, 'index': self.index, 'count': self.count,
+             'reports': dict(self.reports), 'capabilities': self.capabilities, 'audited_skips': skipped,
+             'collection_errors': self.errors, 'gate_errors': errors,
+             'pytest_exit': int(exitstatus), 'status': 'completed' if not errors and int(exitstatus) == 0 else 'failed'}
+    write_json(self.output / 'outcomes.json', value)
+    if errors and int(exitstatus) == 0:
+      session.exitstatus = 1
+
+
+def run(index, count, output):
+  output = Path(output).resolve()
+  output.mkdir(parents=True, exist_ok=False)
+  before = source_snapshot()
+  write_json(output / 'source-before.json', before)
+  env = dict(os.environ, STARPILOT_TEST_DIAGNOSTICS=str(output / 'diagnostics'))
+  command = [sys.executable, str(ROOT / 'tools/ci/run_host_tests.py'), '--pytest', '-p',
+             'tools.ci.custom_pytest_shards', f'--custom-shard-index={index}', f'--custom-shard-count={count}',
+             f'--custom-shard-output={output}']
+  child = subprocess.run(command, cwd=ROOT, env=env, check=False)
+  after = source_snapshot()
+  write_json(output / 'source-after.json', after)
+  plan_path = output / 'plan.json'
+  selected = json.loads(plan_path.read_text())['selected'] if plan_path.is_file() else []
+  result = {'capabilities': capability_state(selected), 'index': index, 'count': count, 'child_exit': child.returncode, 'command': command,
+            'source_before': before, 'source_after': after,
+            'outcomes_sha256': checksum((output / 'outcomes.json').read_bytes()) if (output / 'outcomes.json').is_file() else None,
+            'junit_sha256': checksum((output / 'diagnostics/pytest.xml').read_bytes()) if (output / 'diagnostics/pytest.xml').is_file() else None,
+            'plan_sha256': checksum((output / 'plan.json').read_bytes()) if (output / 'plan.json').is_file() else None,
+            'selected_collection_sha256': checksum((output / 'diagnostics/pytest-collection.json').read_bytes())
+            if (output / 'diagnostics/pytest-collection.json').is_file() else None}
+  success = child.returncode == 0 and before == after and all(result[key] for key in (
+    'outcomes_sha256', 'junit_sha256', 'plan_sha256', 'selected_collection_sha256'))
+  result['status'] = 'completed' if success else 'failed'
+  write_json(output / 'receipt.json', result)
+  return 0 if success else 1
+
+
+def aggregate(directory, count, root=ROOT):
+  paths = sorted(Path(directory).rglob('receipt.json'))
+  if len(paths) != count:
+    raise ValueError('Missing or duplicate custom pytest shard jobs')
+  expected_source = source_snapshot(root)
+  full = None
+  seen = []
+  audited_skips = []
+  indices = set()
+  for path in paths:
+    doc = json.loads(path.read_text())
+    index = doc['index']
+    if type(index) is not int or not 0 <= index < count or index in indices or doc['count'] != count:
+      raise ValueError('Invalid or duplicate custom pytest shard index')
+    indices.add(index)
+    if doc['status'] != 'completed' or doc['child_exit'] != 0:
+      raise ValueError('Failed, cancelled or incomplete custom pytest job')
+    before, after = doc['source_before'], doc['source_after']
+    if before != after or before['runners'] != expected_source['runners']:
+      raise ValueError('Source or runner changed during execution')
+    if not same_build_inputs(before['build'], expected_source['build']):
+      raise ValueError('Custom pytest source HEAD, input closure or build recipe mismatch')
+    outcome_path = path.with_name('outcomes.json')
+    junit = path.parent / 'diagnostics/pytest.xml'
+    if not outcome_path.is_file() or checksum(outcome_path.read_bytes()) != doc['outcomes_sha256']:
+      raise ValueError('Missing or changed custom pytest outcomes artifact')
+    if not junit.is_file() or checksum(junit.read_bytes()) != doc['junit_sha256']:
+      raise ValueError('Missing or changed custom pytest JUnit artifact')
+    for name, key in (('plan.json', 'plan_sha256'), ('diagnostics/pytest-collection.json', 'selected_collection_sha256')):
+      artifact = path.parent / name
+      if not artifact.is_file() or checksum(artifact.read_bytes()) != doc[key]:
+        raise ValueError('Missing or changed actual collection witness')
+    plan = json.loads(path.with_name('plan.json').read_text())
+    outcome = json.loads(outcome_path.read_text())
+    if any(plan[key] != outcome[key] for key in ('index', 'count', 'catalog', 'selected')):
+      raise ValueError('Actual collection plan and outcomes disagree')
+    actual_selected = json.loads((path.parent / 'diagnostics/pytest-collection.json').read_text())
+    if actual_selected != outcome['selected']:
+      raise ValueError('Actual pytest progress collection differs from selected outcomes')
+    if outcome['status'] != 'completed' or outcome['pytest_exit'] != 0 or outcome['collection_errors'] or outcome['gate_errors']:
+      raise ValueError('Custom pytest collection, setup or execution failed')
+    if outcome['index'] != index or outcome['count'] != count:
+      raise ValueError('Custom pytest outcome job identity differs')
+    catalog = outcome['catalog']
+    if full is None:
+      full = catalog
+    if catalog != full:
+      raise ValueError('Custom pytest full catalogs disagree')
+    assignment, _ = partition(catalog, count, root)
+    if outcome['selected'] != assignment[index]:
+      raise ValueError('Forged custom pytest case assignment')
+    validate_capabilities(outcome['selected'], outcome['capabilities'], root)
+    if doc['capabilities'] != outcome['capabilities']:
+      raise ValueError('Wrapper and pytest capability observations disagree')
+    skipped = validate_outcomes(outcome['selected'], outcome['reports'], outcome['capabilities'])
+    if skipped != outcome['audited_skips']:
+      raise ValueError('Forged audited capability skip disposition')
+    validate_junit(junit, outcome['selected'], skipped)
+    audited_skips.extend(skipped)
+    seen.extend(outcome['selected'])
+  if not full or len(full) != len(set(full)) or Counter(seen) != Counter(full):
+    raise ValueError('Custom pytest aggregate must cover every collected case exactly once')
+  return {'status': 'completed', 'shards': count, 'cases_passed': len(full) - len(audited_skips),
+          'cases_collected': len(full), 'cases_audited_skipped': len(audited_skips), 'audited_skips': audited_skips,
+          'modules': len({node.split('::', 1)[0] for node in full}),
+          'revision': expected_source['build']['inputs']['revision']}
+
+
+def main():
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('action', choices=('run', 'aggregate'))
+  parser.add_argument('--index', type=int)
+  parser.add_argument('--count', type=int, default=24)
+  parser.add_argument('--output', type=Path, required=True)
+  parser.add_argument('--directory', type=Path)
+  args = parser.parse_args()
+  if args.action == 'run':
+    if args.index is None:
+      parser.error('--index is required')
+    return run(args.index, args.count, args.output)
+  if args.directory is None:
+    parser.error('--directory is required')
+  result = aggregate(args.directory, args.count)
+  write_json(args.output, result)
+  print(json.dumps(result))
+  return 0
+
+
+if __name__ == '__main__':
+  raise SystemExit(main())

tools/ci/tests/test_custom_pytest_shards.py

@@ -0,0 +1,554 @@
+"""Pure tests for custom pytest selection, durable outcomes and mandatory completion."""
+import copy
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+from xml.etree import ElementTree
+
+import pytest
+
+from tools.ci import custom_pytest_shards as shards
+
+
+def source():
+  return {'build': {'inputs': {'revision': 'commit'}, 'declared_outputs': {}}, 'runners': {'runner': 'a' * 64}}
+
+
+def reports(nodes):
+  return {node: [{'phase': phase, 'status': 'passed', 'duration': .01} for phase in ('setup', 'call', 'teardown')] for node in nodes}
+
+
+
+def save_junit(path, selected, skipped):
+  root = ElementTree.Element('testsuites')
+  suite = ElementTree.SubElement(root, 'testsuite')
+  by_node = {row['id']: row for row in skipped}
+  for node in selected:
+    classname, name = shards.junit_address(node)
+    case = ElementTree.SubElement(suite, 'testcase', classname=classname, name=name)
+    if node in by_node:
+      ElementTree.SubElement(case, 'skipped', type='pytest.skip', message=by_node[node]['reason'].removeprefix('Skipped: '))
+  path.parent.mkdir(parents=True, exist_ok=True)
+  ElementTree.ElementTree(root).write(path)
+
+
+def save_job(directory, index, count, catalog, selected):
+  root = directory / str(index)
+  root.mkdir(parents=True)
+  outcome = {'index': index, 'count': count, 'catalog': catalog, 'selected': selected, 'reports': reports(selected),
+             'status': 'completed', 'pytest_exit': 0, 'collection_errors': [], 'gate_errors': [], 'capabilities': {}, 'audited_skips': []}
+  shards.write_json(root / 'outcomes.json', outcome)
+  shards.write_json(root / 'plan.json', {key: outcome[key] for key in ('index', 'count', 'catalog', 'selected')})
+  shards.write_json(root / 'diagnostics/pytest-collection.json', selected)
+  save_junit(root / 'diagnostics/pytest.xml', selected, [])
+  receipt = {'capabilities': {}, 'index': index, 'count': count, 'status': 'completed', 'child_exit': 0,
+             'source_before': source(), 'source_after': source(),
+             'outcomes_sha256': shards.checksum((root / 'outcomes.json').read_bytes()),
+             'junit_sha256': shards.checksum((root / 'diagnostics/pytest.xml').read_bytes()),
+             'plan_sha256': shards.checksum((root / 'plan.json').read_bytes()),
+             'selected_collection_sha256': shards.checksum((root / 'diagnostics/pytest-collection.json').read_bytes())}
+  shards.write_json(root / 'receipt.json', receipt)
+  return root
+
+
+@pytest.fixture
+def jobs(tmp_path, monkeypatch):
+  monkeypatch.setattr(shards, 'source_snapshot', lambda root=shards.ROOT: source())
+  monkeypatch.setattr(shards, 'same_build_inputs', lambda a, b: a == b)
+  catalog = ['a.py::test_a[1]', 'a.py::test_a[10]', 'b.py::test_b']
+  assigned, _ = shards.partition(catalog, 2)
+  paths = [save_job(tmp_path, index, 2, catalog, selected) for index, selected in enumerate(assigned)]
+  return tmp_path, paths, catalog
+
+
+def rewrite(path, name, change):
+  target = path / name
+  value = json.loads(target.read_text())
+  change(value)
+  shards.write_json(target, value)
+  if name == 'outcomes.json':
+    receipt = json.loads((path / 'receipt.json').read_text())
+    receipt['outcomes_sha256'] = shards.checksum(target.read_bytes())
+    shards.write_json(path / 'receipt.json', receipt)
+
+
+def test_complete_exact_catalog(jobs):
+  directory, _, catalog = jobs
+  result = shards.aggregate(directory, 2)
+  assert result['cases_passed'] == len(catalog)
+  assert result['shards'] == 2
+
+
+@pytest.mark.parametrize('mutation', [
+  lambda d: d.update(status='failed'),
+  lambda d: d.update(status='cancelled'),
+  lambda d: d.update(child_exit=1),
+  lambda d: d.update(count=3),
+  lambda d: d.update(index=1),
+  lambda d: d['source_after']['runners'].update(runner='b' * 64),
+  lambda d: d['source_before']['runners'].update(runner='b' * 64),
+  lambda d: d['source_before']['build']['inputs'].update(revision='other'),
+  lambda d: d.update(outcomes_sha256='0' * 64),
+  lambda d: d.update(junit_sha256='0' * 64),
+])
+def test_failed_cancelled_forged_or_changed_job_denied(jobs, mutation):
+  directory, paths, _ = jobs
+  rewrite(paths[0], 'receipt.json', mutation)
+  with pytest.raises(ValueError):
+    shards.aggregate(directory, 2)
+
+
+@pytest.mark.parametrize('name', ['receipt.json', 'outcomes.json', 'diagnostics/pytest.xml', 'plan.json', 'diagnostics/pytest-collection.json'])
+def test_missing_job_or_artifact_denied(jobs, name):
+  directory, paths, _ = jobs
+  (paths[0] / name).unlink()
+  with pytest.raises(ValueError):
+    shards.aggregate(directory, 2)
+
+
+@pytest.mark.parametrize('mutation', [
+  lambda d: d.update(status='failed'),
+  lambda d: d.update(pytest_exit=2),
+  lambda d: d['collection_errors'].append({'id': 'bad.py'}),
+  lambda d: d['gate_errors'].append({'error': 'setup'}),
+  lambda d: d['catalog'].append('invented.py::test_fake'),
+  lambda d: d['catalog'].append(d['catalog'][0]),
+  lambda d: d['selected'].append(d['selected'][0]),
+  lambda d: d.update(selected=[]),
+  lambda d: d['reports'].clear(),
+  lambda d: next(iter(d['reports'].values())).pop(),
+  lambda d: next(iter(d['reports'].values())).append({'phase': 'call', 'status': 'passed'}),
+  lambda d: next(iter(d['reports'].values()))[1].update(status='skipped'),
+  lambda d: next(iter(d['reports'].values()))[1].update(status='failed'),
+  lambda d: next(iter(d['reports'].values()))[0].update(status='failed'),
+  lambda d: next(iter(d['reports'].values()))[2].update(status='failed'),
+])
+def test_collection_assignment_outcome_errors_denied(jobs, mutation):
+  directory, paths, _ = jobs
+  rewrite(paths[0], 'outcomes.json', mutation)
+  with pytest.raises(ValueError):
+    shards.aggregate(directory, 2)
+
+
+def test_duplicate_job_denied(jobs):
+  directory, paths, _ = jobs
+  import shutil
+  shutil.copytree(paths[0], directory / 'duplicate')
+  with pytest.raises(ValueError):
+    shards.aggregate(directory, 2)
+
+
+def test_changed_source_in_all_jobs_denied_against_checkout(jobs):
+  directory, paths, _ = jobs
+  for path in paths:
+    def change(d):
+      for name in ('source_before', 'source_after'):
+        d[name]['build']['inputs']['revision'] = 'forged'
+    rewrite(path, 'receipt.json', change)
+  with pytest.raises(ValueError, match='source HEAD'):
+    shards.aggregate(directory, 2)
+
+
+def test_exact_parameter_nodes_whole_other_modules_and_bounded_heavy_groups():
+  heavy = 'openpilot/starpilot/car/toyota/tests/test_legacy_advert_callers.py'
+  catalog = [f'{heavy}::test_independent[{i}]' for i in range(126)] + ['other.py::test_case[1]', 'other.py::test_case[10]']
+  assigned, loads = shards.partition(catalog, 24)
+  assert sorted(node for group in assigned for node in group) == sorted(catalog)
+  assert max(loads) <= 240
+  assert any('other.py::test_case[1]' in group and 'other.py::test_case[10]' in group for group in assigned)
+  assert [sum(node.startswith(heavy) for node in group) for group in assigned].count(60) == 2
+  assert shards.partition(catalog, 24) == (assigned, loads)
+
+
+@pytest.mark.parametrize('catalog', [[], ['a.py::test_a', 'a.py::test_a'], ['../a.py::test_a'], ['/a.py::test_a'], ['a.py']])
+def test_invalid_catalog_denied(catalog):
+  with pytest.raises(ValueError):
+    shards.partition(catalog, 24)
+
+
+def test_unreviewed_heavy_class_not_split():
+  with pytest.raises(ValueError, match='top-level'):
+    shards.partition(['openpilot/starpilot/car/toyota/tests/test_legacy_advert_callers.py::TestShared::test_x'], 24)
+
+
+def test_changed_source_invalidates_measured_duration(tmp_path, monkeypatch):
+  p = tmp_path / 'a.py'
+  p.write_text('original')
+  config = copy.deepcopy(shards.policy())
+  config['modules'] = {'a.py': {'cases': 1, 'seconds': 99., 'blob': shards.blob_id(p)}}
+  monkeypatch.setattr(shards, 'policy', lambda: config)
+  assert shards.partition(['a.py::test_a'], 1, tmp_path)[1] == [99.]
+  p.write_text('changed')
+  assert shards.partition(['a.py::test_a'], 1, tmp_path)[1] == [1.]
+
+
+@pytest.mark.parametrize('body,expected', [
+  ('def test_a():\n assert True\ndef test_b():\n assert True\n', 'completed'),
+  ('def test_a():\n assert False\ndef test_b():\n assert True\n', 'failed'),
+  ('import pytest\n@pytest.mark.skip(reason="visible skip")\ndef test_a():\n pass\ndef test_b():\n assert True\n', 'failed'),
+  ('raise RuntimeError("collection failure")\n', 'failed'),
+])
+def test_real_pytest_plugin_durable_results(tmp_path, body, expected):
+  (tmp_path / 'test_cases.py').write_text(body)
+  output = tmp_path / 'evidence'
+  env = dict(os.environ, PYTHONPATH=str(shards.ROOT), PYTEST_DISABLE_PLUGIN_AUTOLOAD='1',
+             STARPILOT_TEST_DIAGNOSTICS=str(output / 'diagnostics'))
+  command = [sys.executable, '-m', 'pytest', '-q', '-p', 'tools.ci.custom_pytest_shards',
+             '-p', 'tools.ci.pytest_progress', '--custom-shard-index=0', '--custom-shard-count=1',
+             f'--custom-shard-output={output}', f'--junitxml={output}/diagnostics/pytest.xml', 'test_cases.py']
+  child = subprocess.run(command, cwd=tmp_path, env=env, capture_output=True, timeout=20, check=False)
+  outcome = json.loads((output / 'outcomes.json').read_text())
+  assert outcome['status'] == expected, child.stdout.decode() + child.stderr.decode()
+  assert (child.returncode == 0) == (expected == 'completed')
+  assert (output / 'diagnostics/pytest.xml').is_file()
+  if expected == 'completed':
+    assert len(outcome['catalog']) == len(outcome['selected']) == 2
+    assert list((output / 'diagnostics').glob('worker-*.jsonl'))
+
+
+def test_real_child_death_leaves_failed_receipt_and_durable_node(tmp_path, monkeypatch):
+  wrapper = tmp_path / 'tools/ci/run_host_tests.py'
+  wrapper.parent.mkdir(parents=True)
+  wrapper.write_bytes((Path(__file__).parents[1] / 'run_host_tests.py').read_bytes())
+  module = tmp_path / 'openpilot/starpilot/tests/test_signal.py'
+  module.parent.mkdir(parents=True)
+  module.write_text('import os, signal\ndef test_child_signal():\n os.kill(os.getpid(), signal.SIGTERM)\n')
+  monkeypatch.setattr(shards, 'ROOT', tmp_path)
+  monkeypatch.setattr(shards, 'source_snapshot', lambda: source())
+  monkeypatch.setenv('PYTHONPATH', str(Path(__file__).parents[3]))
+  output = tmp_path / 'evidence'
+  assert shards.run(0, 1, output) == 1
+  receipt = json.loads((output / 'receipt.json').read_text())
+  assert receipt['status'] == 'failed'
+  assert receipt['child_exit'] == 143
+  assert receipt['outcomes_sha256'] is None
+  workers = list((output / 'diagnostics').glob('worker-*.json'))
+  assert workers
+  assert any(json.loads(path.read_text())['active_test'].endswith('::test_child_signal') for path in workers)
+
+
+def test_real_plugin_exact_parameter_assignment_across_shards(tmp_path):
+  path = tmp_path / 'openpilot/starpilot/car/toyota/tests/test_legacy_advert_callers.py'
+  path.parent.mkdir(parents=True)
+  path.write_text('import pytest\n@pytest.mark.parametrize("value", range(126))\ndef test_independent(value):\n assert 0 <= value < 126\n')
+  outcomes = []
+  for index in range(3):
+    output = tmp_path / f'shard-{index}'
+    env = dict(os.environ, PYTHONPATH=str(shards.ROOT), PYTEST_DISABLE_PLUGIN_AUTOLOAD='1')
+    command = [sys.executable, '-m', 'pytest', '-q', '-p', 'tools.ci.custom_pytest_shards',
+               f'--custom-shard-index={index}', '--custom-shard-count=3', f'--custom-shard-output={output}', str(path)]
+    child = subprocess.run(command, cwd=tmp_path, env=env, capture_output=True, timeout=20, check=False)
+    assert child.returncode == 0, child.stdout.decode() + child.stderr.decode()
+    outcomes.append(json.loads((output / 'outcomes.json').read_text()))
+  seen = [node for outcome in outcomes for node in outcome['selected']]
+  assert len(seen) == len(set(seen)) == 126
+  assert set(seen) == set(outcomes[0]['catalog'])
+  for outcome in outcomes:
+    shards.validate_outcomes(outcome['selected'], outcome['reports'])
+
+
+def test_forged_catalog_in_all_outcomes_denied_by_actual_collection_witness(jobs):
+  directory, paths, _ = jobs
+  for path in paths:
+    rewrite(path, 'outcomes.json', lambda d: d['catalog'].append('invented.py::test_fake'))
+  with pytest.raises(ValueError, match='collection plan'):
+    shards.aggregate(directory, 2)
+
+
+def test_changed_progress_collection_denied(jobs):
+  directory, paths, _ = jobs
+  shards.write_json(paths[0] / 'diagnostics/pytest-collection.json', ['invented.py::test_fake'])
+  with pytest.raises(ValueError, match='collection witness'):
+    shards.aggregate(directory, 2)
+
+
+def test_real_wrapper_shards_and_aggregate_keep_every_case(tmp_path, monkeypatch):
+  wrapper = tmp_path / 'tools/ci/run_host_tests.py'
+  wrapper.parent.mkdir(parents=True)
+  wrapper.write_bytes((Path(__file__).parents[1] / 'run_host_tests.py').read_bytes())
+  tests = tmp_path / 'openpilot/starpilot/tests'
+  tests.mkdir(parents=True)
+  for name in ('a', 'b'):
+    (tests / f'test_{name}.py').write_text(f'def test_{name}():\n assert True\n')
+  monkeypatch.setattr(shards, 'ROOT', tmp_path)
+  monkeypatch.setattr(shards, 'source_snapshot', lambda root=tmp_path: source())
+  monkeypatch.setattr(shards, 'same_build_inputs', lambda a, b: a == b)
+  monkeypatch.setenv('PYTHONPATH', str(Path(__file__).parents[3]))
+  for index in range(2):
+    assert shards.run(index, 2, tmp_path / 'jobs' / str(index)) == 0
+  result = shards.aggregate(tmp_path / 'jobs', 2)
+  assert result['cases_passed'] == 2
+  assert result['status'] == 'completed'
+
+
+def test_real_module_collection_skip_denied_even_when_other_cases_pass(tmp_path):
+  (tmp_path / 'test_skip.py').write_text('import pytest\npytest.skip("module skip must remain failure", allow_module_level=True)\n')
+  (tmp_path / 'test_pass.py').write_text('def test_pass():\n assert True\n')
+  output = tmp_path / 'evidence'
+  env = dict(os.environ, PYTHONPATH=str(shards.ROOT), PYTEST_DISABLE_PLUGIN_AUTOLOAD='1')
+  child = subprocess.run([sys.executable, '-m', 'pytest', '-q', '-p', 'tools.ci.custom_pytest_shards',
+                          '--custom-shard-index=0', '--custom-shard-count=1', f'--custom-shard-output={output}',
+                          'test_skip.py', 'test_pass.py'], cwd=tmp_path, env=env, capture_output=True, timeout=20, check=False)
+  outcome = json.loads((output / 'outcomes.json').read_text())
+  assert child.returncode != 0
+  assert outcome['status'] == 'failed'
+  assert outcome['collection_errors'][0]['id'] == 'test_skip.py'
+  assert len(outcome['selected']) == 1
+  shards.validate_outcomes(outcome['selected'], outcome['reports'])
+
+
+
+def unavailable_observation(guard):
+  return {'gpu_env': {'STARPILOT_AUTO_GL_TEST': None}, 'netifaces': {'importable': False},
+          'real_xapk': {'path': shards.REAL_XAPK, 'exists': False}}[guard]
+
+
+def skipped_reports(node, reason):
+  return {node: [{'phase': 'setup', 'status': 'skipped', 'duration': .01, 'reason': reason},
+                 {'phase': 'teardown', 'status': 'passed', 'duration': .01}]}
+
+
+@pytest.fixture
+def capability_guards(monkeypatch):
+  inventory = shards.policy()['capability_skips']
+  blobs = {node.split('::')[0]: row['source_blob'] for node, row in inventory.items()}
+  monkeypatch.setattr(shards, 'blob_id', lambda path: blobs[path.relative_to(shards.ROOT).as_posix()])
+  monkeypatch.setattr(shards, 'guard_observation', lambda guard, root: unavailable_observation(guard))
+  return inventory
+
+
+@pytest.mark.parametrize('node', sorted(shards.CAPABILITY_NODES))
+def test_each_original_capability_skip_remains_unavailable_never_passed(capability_guards, node):
+  row = capability_guards[node]
+  state = shards.capability_state([node])
+  shards.validate_capabilities([node], state)
+  result = shards.validate_outcomes([node], skipped_reports(node, row['reason']), state)
+  assert result == [{'id': node, 'status': 'skipped', 'disposition': 'capability-unavailable',
+                     'reason': row['reason'], 'source_blob': row['source_blob'], 'guard': row['guard'],
+                     'observation': unavailable_observation(row['guard'])}]
+  assert shards.validate_outcomes([node], reports([node]), state) == []
+
+
+@pytest.mark.parametrize('node', sorted(shards.CAPABILITY_NODES))
+def test_each_capability_wrong_reason_or_source_denied(capability_guards, node):
+  row = capability_guards[node]
+  state = shards.capability_state([node])
+  with pytest.raises(ValueError):
+    shards.validate_outcomes([node], skipped_reports(node, row['reason'] + ' changed'), state)
+  state[node]['source_blob'] = '0' * 40
+  with pytest.raises(ValueError):
+    shards.validate_outcomes([node], skipped_reports(node, row['reason']), state)
+  with pytest.raises(ValueError):
+    shards.validate_capabilities([node], state)
+
+
+@pytest.mark.parametrize('node', sorted(shards.CAPABILITY_NODES))
+def test_each_available_capability_cannot_be_skipped(capability_guards, node):
+  row = capability_guards[node]
+  state = shards.capability_state([node])
+  observation = {'gpu_env': {'STARPILOT_AUTO_GL_TEST': '1'}, 'netifaces': {'importable': True},
+                 'real_xapk': {'path': shards.REAL_XAPK, 'exists': True}}[row['guard']]
+  state[node]['observation'] = observation
+  with pytest.raises(ValueError):
+    shards.validate_outcomes([node], skipped_reports(node, row['reason']), state)
+
+
+def test_forged_capability_node_and_condition_denied(capability_guards):
+  node = sorted(capability_guards)[0]
+  state = shards.capability_state([node])
+  with pytest.raises(ValueError):
+    shards.validate_outcomes([node + '-forged'], skipped_reports(node + '-forged', capability_guards[node]['reason']), state)
+  state[node]['observation'] = {'unavailable': True}
+  with pytest.raises(ValueError):
+    shards.validate_capabilities([node], state)
+
+
+def test_audited16_counts_are_skipped_not_passed_in_exact_aggregate(tmp_path, monkeypatch, capability_guards):
+  monkeypatch.setattr(shards, 'source_snapshot', lambda root=shards.ROOT: source())
+  monkeypatch.setattr(shards, 'same_build_inputs', lambda a, b: a == b)
+  catalog = sorted(capability_guards) + ['required.py::test_required']
+  assigned, _ = shards.partition(catalog, 2)
+  for index, selected in enumerate(assigned):
+    path = save_job(tmp_path, index, 2, catalog, selected)
+    state = shards.capability_state(selected)
+    phase_reports = reports(selected)
+    for node in state:
+      phase_reports.update(skipped_reports(node, capability_guards[node]['reason']))
+    skipped = shards.validate_outcomes(selected, phase_reports, state)
+    rewrite(path, 'outcomes.json', lambda d, phases=phase_reports, caps=state, skips=skipped:
+            d.update(reports=phases, capabilities=caps, audited_skips=skips))
+    save_junit(path / 'diagnostics/pytest.xml', selected, skipped)
+    junit_hash = shards.checksum((path / 'diagnostics/pytest.xml').read_bytes())
+    rewrite(path, 'receipt.json', lambda d, caps=state, digest=junit_hash: d.update(capabilities=caps, junit_sha256=digest))
+  result = shards.aggregate(tmp_path, 2)
+  assert result['cases_collected'] == 17
+  assert result['cases_passed'] == 1
+  assert result['cases_audited_skipped'] == 16
+  assert len(result['audited_skips']) == 16
+  assert all(row['status'] == 'skipped' for row in result['audited_skips'])
+
+
+def test_actual_capability_guard_conditions(tmp_path, monkeypatch):
+  monkeypatch.delenv('STARPILOT_AUTO_GL_TEST', raising=False)
+  assert shards.guard_unavailable('gpu_env', shards.guard_observation('gpu_env', tmp_path))
+  monkeypatch.setenv('STARPILOT_AUTO_GL_TEST', '1')
+  assert not shards.guard_unavailable('gpu_env', shards.guard_observation('gpu_env', tmp_path))
+  assert shards.guard_unavailable('real_xapk', shards.guard_observation('real_xapk', tmp_path))
+  apk = tmp_path / shards.REAL_XAPK
+  apk.parent.mkdir(parents=True)
+  apk.write_bytes(b'presence-only guard witness; not a real APK')
+  assert not shards.guard_unavailable('real_xapk', shards.guard_observation('real_xapk', tmp_path))
+  def unavailable(name):
+    assert name == 'netifaces'
+    raise ImportError('optional dependency absent')
+  monkeypatch.setattr(shards.importlib, 'import_module', unavailable)
+  assert shards.guard_unavailable('netifaces', shards.guard_observation('netifaces', tmp_path))
+  monkeypatch.setattr(shards.importlib, 'import_module', lambda name: object())
+  assert not shards.guard_unavailable('netifaces', shards.guard_observation('netifaces', tmp_path))
+
+
+
+@pytest.mark.parametrize('change', ['missing', 'duplicate', 'skip', 'failure', 'error'])
+def test_junit_false_success_and_unaudited_skips_denied(jobs, change):
+  directory, paths, _ = jobs
+  path = paths[0] / 'diagnostics/pytest.xml'
+  tree = ElementTree.parse(path)
+  suite = tree.getroot().find('testsuite')
+  assert suite is not None
+  case = suite.find('testcase')
+  assert case is not None
+  if change == 'missing':
+    suite.remove(case)
+  elif change == 'duplicate':
+    suite.append(copy.deepcopy(case))
+  else:
+    ElementTree.SubElement(case, 'skipped' if change == 'skip' else change)
+  tree.write(path)
+  rewrite(paths[0], 'receipt.json', lambda d: d.update(junit_sha256=shards.checksum(path.read_bytes())))
+  with pytest.raises(ValueError, match='JUnit'):
+    shards.aggregate(directory, 2)
+
+
+def test_real_pytest_all16_audited_guards_have_skipped_junit_not_passes(tmp_path):
+  """Real pytest phases/JUnit with controlled guard witnesses; no graphics or private APK is tested."""
+  import shutil
+  package = tmp_path / 'tools/ci'
+  package.mkdir(parents=True)
+  (tmp_path / 'tools/__init__.py').touch()
+  (package / '__init__.py').touch()
+  for name in ('custom_pytest_shards.py', 'lifecycle_shards.py', 'pytest_progress.py', 'worker_diagnostics.py'):
+    shutil.copyfile(shards.ROOT / 'tools/ci' / name, package / name)
+  config = shards.policy()
+  by_module = {}
+  for node in sorted(shards.CAPABILITY_NODES):
+    by_module.setdefault(node.split('::')[0], []).append(node)
+  for module, nodes in by_module.items():
+    source_path = tmp_path / module
+    source_path.parent.mkdir(parents=True, exist_ok=True)
+    lines = ['import pytest', 'from unittest import SkipTest']
+    groups = {}
+    for node in nodes:
+      tail = node.split('::', 1)[1]
+      function, bracket, parameter = tail.partition('[')
+      groups.setdefault(function, []).append(parameter.removesuffix(']') if bracket else None)
+    for function, parameters in groups.items():
+      reason = config['capability_skips'][nodes[0]]['reason'].removeprefix('Skipped: ')
+      if parameters != [None]:
+        lines.append(f'@pytest.mark.parametrize("case", range({len(parameters)}), ids={parameters!r})')
+        signature = 'case'
+      else:
+        signature = ''
+      lines.extend([f'def {function}({signature}):', f'  raise SkipTest({reason!r})'])
+    # The original guards skip during setup; a fixture models that exact phase.
+    lines = ['import pytest', 'from unittest import SkipTest', '@pytest.fixture(autouse=True)',
+             'def original_capability_guard():', f'  raise SkipTest({reason!r})'] + [
+               line.replace(f'  raise SkipTest({reason!r})', '  assert False, "body must not run when original guard is unavailable"')
+               for line in lines[2:]]
+    source_path.write_text('\n'.join(lines) + '\n')
+    for node in nodes:
+      config['capability_skips'][node]['source_blob'] = shards.blob_id(source_path)
+  shards.write_json(package / 'custom_pytest_workloads.json', config)
+  (tmp_path / 'guard_witness.py').write_text('''def pytest_configure(config):
+  from tools.ci import custom_pytest_shards as s
+  def observe(guard, root):
+    return {'gpu_env': {'STARPILOT_AUTO_GL_TEST': None}, 'netifaces': {'importable': False},
+            'real_xapk': {'path': s.REAL_XAPK, 'exists': False}}[guard]
+  s.guard_observation = observe
+''')
+  output = tmp_path / 'evidence'
+  env = dict(os.environ, PYTHONPATH=str(tmp_path), PYTEST_DISABLE_PLUGIN_AUTOLOAD='1',
+             STARPILOT_TEST_DIAGNOSTICS=str(output / 'diagnostics'))
+  child = subprocess.run([sys.executable, '-m', 'pytest', '-q', '-p', 'tools.ci.custom_pytest_shards',
+                          '-p', 'tools.ci.pytest_progress', '-p', 'guard_witness', '--custom-shard-index=0',
+                          '--custom-shard-count=1', f'--custom-shard-output={output}',
+                          f'--junitxml={output}/diagnostics/pytest.xml', *by_module], cwd=tmp_path,
+                         env=env, capture_output=True, timeout=20, check=False)
+  outcome = json.loads((output / 'outcomes.json').read_text())
+  assert child.returncode == 0, child.stdout.decode() + child.stderr.decode()
+  assert outcome['status'] == 'completed'
+  assert len(outcome['catalog']) == len(outcome['audited_skips']) == 16
+  assert all(row['status'] == 'skipped' for row in outcome['audited_skips'])
+  shards.validate_junit(output / 'diagnostics/pytest.xml', outcome['selected'], outcome['audited_skips'])
+  progress = json.loads((output / 'diagnostics/pytest-collection.json').read_text())
+  assert progress == outcome['selected']
+
+
+def test_workflow_uses_test_interpreter_for_helper_and_actual_wrapper(tmp_path):
+  """Two real Python environments; optional module exists only in configured test environment."""
+  import venv
+  project = tmp_path / 'project'
+  test_env = project / '.venv'
+  parent_env = tmp_path / 'parent-python'
+  builder = venv.EnvBuilder(with_pip=False, symlinks=True)
+  builder.create(test_env)
+  builder.create(parent_env)
+  if os.name == 'nt':
+    pytest.fail('The hosted custom shard workflow requires Linux Python paths')
+  test_python = test_env / 'bin/python'
+  parent_python = parent_env / 'bin/python'
+  probe_file = tmp_path / 'worker-observation.json'
+  for interpreter in (test_python, parent_python):
+    result = subprocess.check_output([str(interpreter), '-c', 'import sysconfig; print(sysconfig.get_path("purelib"))'], text=True)
+    site = Path(result.strip())
+    site.mkdir(parents=True, exist_ok=True)
+    # This probe module records interpreter/capability only; it does not claim to execute pytest cases.
+    (site / 'pytest.py').write_text('''import json, os, sys
+from pathlib import Path
+from tools.ci import custom_pytest_shards as s
+Path(os.environ['OPTIONAL_PROBE']).write_text(json.dumps({
+  'python': sys.executable, 'guard': s.guard_observation('netifaces', Path.cwd())}))
+''')
+    if interpreter == test_python:
+      (site / 'netifaces.py').write_text('"""Pure optional-dependency witness; no native package imported."""\n')
+  wrapper = project / 'tools/ci/run_host_tests.py'
+  wrapper.parent.mkdir(parents=True)
+  wrapper.write_bytes((shards.ROOT / 'tools/ci/run_host_tests.py').read_bytes())
+  test_file = project / 'openpilot/starpilot/tests/test_probe.py'
+  test_file.parent.mkdir(parents=True)
+  test_file.write_text('def test_probe():\n assert True\n')
+  probe = tmp_path / 'parent-probe.py'
+  probe.write_text('''import json, os, subprocess, sys
+from pathlib import Path
+from tools.ci import custom_pytest_shards as s
+project = Path(sys.argv[1])
+parent = {'python': sys.executable, 'guard': s.guard_observation('netifaces', project)}
+subprocess.run([sys.executable, str(project / 'tools/ci/run_host_tests.py'), '--pytest'], cwd=project, check=True)
+worker = json.loads(Path(os.environ['OPTIONAL_PROBE']).read_text())
+print(json.dumps({'parent': parent, 'worker': worker}))
+''')
+  environment = dict(os.environ, PYTHONPATH=str(shards.ROOT), OPTIONAL_PROBE=str(probe_file))
+  environment.pop('PYTHONHOME', None)
+  wrong = json.loads(subprocess.check_output([str(parent_python), str(probe), str(project)], env=environment, text=True, timeout=20))
+  assert wrong['parent']['guard'] == {'importable': False}
+  workflow = (shards.ROOT / '.github/workflows/tests.yaml').read_text()
+  run_line = next(line for line in workflow.splitlines() if 'tools.ci.custom_pytest_shards run ' in line)
+  interpreter = run_line.split('run: ', 1)[1].split()[0]
+  assert interpreter == '.venv/bin/python'
+  chosen = json.loads(subprocess.check_output([str(project / interpreter), str(probe), str(project)], env=environment, text=True, timeout=20))
+  assert chosen['parent']['guard'] == chosen['worker']['guard'] == {'importable': True}
+  assert chosen['parent']['python'] == chosen['worker']['python'] == str(test_python)
+  aggregate_line = next(line for line in workflow.splitlines() if 'tools.ci.custom_pytest_shards aggregate ' in line)
+  assert aggregate_line.split('run: ', 1)[1].startswith('python3 -m ')
+  assert 'passed or audited capability-unavailable' in workflow