Shard complete lifecycle qualification by module

.github/workflows/tests.yaml

@@ -196,17 +196,6 @@ jobs:
       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: Gate shared AOL vehicle lifecycles
-      if: ${{ !cancelled() && steps.build.outcome == 'success' }}
-      timeout-minutes: 10
-      run: python3 tools/ci/run_vehicle_tests.py --suite aol-lifecycle --output ci-results/aol-lifecycle
-    - name: Save AOL lifecycle cases and source/native evidence
-      if: always()
-      uses: actions/upload-artifact@v4
-      with:
-        name: aol-lifecycle
-        path: ci-results/aol-lifecycle/
-        if-no-files-found: warn
     - name: Run unit tests
       if: ${{ !cancelled() && steps.build.outcome == 'success' }}
       timeout-minutes: ${{ contains(runner.name, 'nsc') && 2 || 20 }}
@@ -252,6 +241,57 @@ jobs:
           gm-test-results.json
         if-no-files-found: error
 
+  aol_lifecycles:
+    name: AOL lifecycles (${{ matrix.shard }})
+    runs-on: ubuntu-24.04
+    timeout-minutes: 35
+    permissions:
+      contents: read
+    strategy:
+      fail-fast: false
+      matrix:
+        shard: [0, 1, 2, 3]
+    env:
+      FUZZ_SEED: '0'
+    steps:
+    - uses: actions/checkout@v7
+    - run: ./tools/op.sh setup
+    - name: Build native host dependencies
+      run: scons
+    - name: Gate every assigned lifecycle module serially
+      timeout-minutes: 10
+      run: python3 tools/ci/run_vehicle_tests.py --suite aol-lifecycle --shard-index "${{ matrix.shard }}" --shard-count 4 --output "ci-results/aol-lifecycle-${{ matrix.shard }}"
+    - name: Save every lifecycle outcome and native build
+      if: always()
+      uses: actions/upload-artifact@v4
+      with:
+        name: aol-lifecycle-${{ matrix.shard }}
+        path: ci-results/aol-lifecycle-${{ matrix.shard }}/
+        if-no-files-found: error
+
+  aol_lifecycle_complete:
+    name: complete AOL lifecycle catalog
+    needs: aol_lifecycles
+    if: always()
+    runs-on: ubuntu-24.04
+    permissions:
+      contents: read
+    steps:
+    - uses: actions/checkout@v7
+    - uses: actions/download-artifact@v4
+      with:
+        pattern: aol-lifecycle-*
+        path: ci-results/aol-lifecycles
+    - name: Require every full-catalog case exactly once and passed
+      run: python3 tools/ci/lifecycle_shards.py ci-results/aol-lifecycles --count 4 --output ci-results/aol-lifecycle-complete.json
+    - name: Save complete catalog verdict
+      if: always()
+      uses: actions/upload-artifact@v4
+      with:
+        name: aol-lifecycle-complete
+        path: ci-results/aol-lifecycle-complete.json
+        if-no-files-found: error
+
   map_provider:
     name: map provider
     runs-on: ubuntu-24.04

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "2e1fb0ffe6106943a260fd81df7d79ddba265636641ce911789f8de09d469c60", "schema_version": 1}
+{"protected_inventory_sha256": "53b649eeca854a7437a55e99304597619361bcf9b177854f9307bf6b394c793f", "schema_version": 1}

tools/ci/lifecycle_shards.py

@@ -0,0 +1,92 @@
+"""Partition whole lifecycle modules and prove complete shard outcomes."""
+import argparse
+import hashlib
+from collections import Counter
+import json
+from pathlib import Path
+
+
+def partition(targets, identities, count):
+  if count < 1 or len(set(targets)) != len(targets) or len(set(identities)) != len(identities):
+    raise ValueError('Invalid lifecycle catalog')
+  owners = {}
+  for identity in identities:
+    matches = [target for target in targets if identity.startswith(target.removesuffix('.py').replace('/', '.') + '.')]
+    if len(matches) != 1:
+      raise ValueError(f'Lifecycle identity needs exactly one module: {identity}')
+    owners[identity] = matches[0]
+  weights = Counter(owners.values())
+  if set(weights) != set(targets):
+    raise ValueError('Lifecycle module collected no cases')
+  shards = [[] for _ in range(count)]
+  loads = [0] * count
+  for target in sorted(targets, key=lambda target: (-weights[target], target)):
+    index = min(range(count), key=lambda index: (loads[index], index))
+    shards[index].append(target)
+    loads[index] += weights[target]
+  return shards
+
+
+def aggregate(directory, count):
+  documents = [json.loads(path.read_text()) for path in sorted(Path(directory).rglob('coverage.json'))]
+  if len(documents) != count:
+    raise ValueError('Missing or duplicate lifecycle shard artifacts')
+  first = documents[0]
+  full = first['full_catalog_test_ids']
+  targets = first['full_catalog_targets']
+  expected = partition(targets, full, count)
+  seen, indices = [], set()
+  for doc in documents:
+    index = doc['shard']['index']
+    if index in indices or not 0 <= index < count or doc['shard']['count'] != count:
+      raise ValueError('Invalid lifecycle shard index')
+    indices.add(index)
+    if doc['suite'] != 'aol-lifecycle' or doc['status'] != 'completed' or doc['exit_code'] != 0:
+      raise ValueError('Lifecycle shard failed')
+    if doc['collection_errors'] or doc['gate_errors']:
+      raise ValueError('Lifecycle shard errors')
+    if doc['full_catalog_test_ids'] != full or doc['full_catalog_targets'] != targets or doc['targets'] != expected[index]:
+      raise ValueError('Lifecycle catalog or partition changed')
+    for key in ('revision', 'inputs', 'manifest_sha256', 'dependencies', 'runner_sha256', 'suite_runner_sha256', 'pytest_runner_sha256', 'shard_runner_sha256'):
+      if doc['source'][key] != first['source'][key]:
+        raise ValueError(f'Lifecycle source disagreement: {key}')
+    if doc['source']['inputs'] != doc['source_after']['inputs']:
+      raise ValueError('Lifecycle source changed during execution')
+    for key in ('safety', 'aol_protocol'):
+      if not doc['native'][key]['sha256']:
+        raise ValueError('Missing genuine native build evidence')
+    ids = doc['selected_test_ids']
+    coverage_path = next(path for path in Path(directory).rglob('coverage.json')
+                         if json.loads(path.read_text())['shard']['index'] == index)
+    result_path = coverage_path.with_name('results.json')
+    if doc['result_sha256'] != hashlib.sha256(result_path.read_bytes()).hexdigest():
+      raise ValueError('Lifecycle outcome artifact changed')
+    result = json.loads(result_path.read_text())
+    records = result['results']
+    if result['exit_code'] != 0 or result['collection_errors'] or result['gate_errors']:
+      raise ValueError('Lifecycle outcome report failed')
+    if result['selected_test_ids'] != ids or Counter(row['id'] for row in records) != Counter(ids) or any(row['status'] != 'passed' for row in records):
+      raise ValueError('Lifecycle missing, duplicate or unsuccessful outcomes')
+    prefixes = tuple(target.removesuffix('.py').replace('/', '.') + '.' for target in expected[index])
+    if ids != [identity for identity in full if identity.startswith(prefixes)]:
+      raise ValueError('Lifecycle shard cases differ from the full catalog')
+    seen.extend(ids)
+  if Counter(seen) != Counter(full) or len(set(full)) != len(full):
+    raise ValueError('Lifecycle aggregate does not cover every case exactly once')
+  return {'status': 'completed', 'shards': count, 'cases_passed': len(full), 'targets': len(targets), 'source_revision': first['source']['revision']}
+
+
+def main():
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument('directory', type=Path)
+  parser.add_argument('--count', type=int, default=4)
+  parser.add_argument('--output', type=Path, required=True)
+  args = parser.parse_args()
+  result = aggregate(args.directory, args.count)
+  args.output.parent.mkdir(parents=True, exist_ok=True)
+  args.output.write_text(json.dumps(result, indent=2) + '\n')
+  print(json.dumps(result))
+
+
+if __name__ == '__main__':
+  main()

tools/ci/run_vehicle_tests.py

@@ -19,7 +19,7 @@ ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT))
 
 from tools import test_runner
-from tools.ci import vehicle_pytest
+from tools.ci import lifecycle_shards, vehicle_pytest
 
 # Representative production callers plus their source/native refusal cases.
 # The emitted selected_test_ids record exact parametrized configurations.
@@ -559,7 +559,8 @@ def source_provenance():
           "inputs": source_input_snapshot(),
           "manifest_sha256": sha256(manifest), "dependencies": json.loads(manifest.read_text())["dependencies"],
           "runner_sha256": sha256(ROOT / "tools/test_runner.py"), "suite_runner_sha256": sha256(__file__),
-          "pytest_runner_sha256": sha256(ROOT / "tools/ci/vehicle_pytest.py")}
+          "pytest_runner_sha256": sha256(ROOT / "tools/ci/vehicle_pytest.py"),
+          "shard_runner_sha256": sha256(ROOT / "tools/ci/lifecycle_shards.py")}
 
 
 def native_provenance():
@@ -615,7 +616,9 @@ def build_aol_protocol(output):
   return {"command": command, "path": str(executable), "sha256": sha256(executable)}
 
 
-def run(suite, output, collect_only=False):
+def run(suite, output, collect_only=False, shard_index=None, shard_count=4):
+  if shard_index is not None and (suite != "aol-lifecycle" or not 0 <= shard_index < shard_count):
+    raise ValueError("Lifecycle shard index is outside its catalog")
   output = Path(output).resolve()
   output.mkdir(parents=True, exist_ok=True)
   started = time.monotonic()
@@ -654,6 +657,16 @@ def run(suite, output, collect_only=False):
     pytest_nodeids, pytest_errors = vehicle_pytest.collect(pytest_targets)
     errors.extend(pytest_errors)
     ids = [test.id() for test in tests] + [vehicle_pytest.test_id(nodeid) for nodeid in pytest_nodeids]
+    if shard_index is not None and not errors:
+      full_targets = list(plan["targets"])
+      shards = lifecycle_shards.partition(full_targets, ids, shard_count)
+      plan.update(full_catalog_targets=full_targets, full_catalog_test_ids=list(ids),
+                  shard={"index": shard_index, "count": shard_count})
+      plan["targets"] = shards[shard_index]
+      prefixes = tuple(target.removesuffix(".py").replace("/", ".") + "." for target in plan["targets"])
+      tests = [test for test in tests if test.id().startswith(prefixes)]
+      pytest_nodeids = [nodeid for nodeid in pytest_nodeids if vehicle_pytest.test_id(nodeid).startswith(prefixes)]
+      ids = [test.id() for test in tests] + [vehicle_pytest.test_id(nodeid) for nodeid in pytest_nodeids]
     if len(set(ids)) != len(ids):
       errors.append("Vehicle runners selected duplicate test IDs")
     if suite == "interfaces":
@@ -718,11 +731,13 @@ def main():
   parser.add_argument("--suite", choices=("interfaces", "safety-debug", "safety-release", "aol-lifecycle"), required=True)
   parser.add_argument("--output", type=Path, required=True)
   parser.add_argument("--collect-only", action="store_true")
+  parser.add_argument("--shard-index", type=int)
+  parser.add_argument("--shard-count", type=int, default=4)
   args = parser.parse_args()
   os.chdir(ROOT)
   # Keep the same generated examples across collection, execution, and reruns.
   os.environ.setdefault("FUZZ_SEED", "0")
-  return run(args.suite, args.output, args.collect_only)
+  return run(args.suite, args.output, args.collect_only, args.shard_index, args.shard_count)
 
 
 if __name__ == "__main__":

tools/ci/tests/test_lifecycle_shards.py

@@ -0,0 +1,91 @@
+import copy
+import hashlib
+import importlib.util
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+spec = importlib.util.spec_from_file_location('private_lifecycle_shards', Path(__file__).parents[1] / 'lifecycle_shards.py')
+assert spec is not None and spec.loader is not None
+helper = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(helper)
+
+
+class LifecycleShardTests(unittest.TestCase):
+  def catalog(self):
+    targets = ['a.py', 'b.py', 'c.py', 'd.py']
+    identities = [f'{name}.Test.test_{i}' for name in 'abcd' for i in range(3)]
+    return targets, identities
+
+  def test_partition_preserves_whole_modules_and_all_cases(self):
+    targets, identities = self.catalog()
+    shards = helper.partition(targets, identities, 4)
+    self.assertEqual(sorted(target for shard in shards for target in shard), targets)
+    self.assertEqual([len(shard) for shard in shards], [1] * 4)
+    self.assertEqual(shards, helper.partition(list(reversed(targets)), identities, 4))
+
+  def test_invalid_catalogs_are_rejected(self):
+    targets, identities = self.catalog()
+    for bad_targets, bad_ids, count in [(targets, identities, 0), (targets + targets, identities, 4),
+                                        (targets, identities + identities, 4), (targets, ['foreign.test'], 4),
+                                        (targets + ['empty.py'], identities, 4)]:
+      with self.subTest(targets=bad_targets, count=count), self.assertRaises(ValueError):
+        helper.partition(bad_targets, bad_ids, count)
+
+  def documents(self):
+    targets, full = self.catalog()
+    partition = helper.partition(targets, full, 4)
+    source = dict.fromkeys(('revision', 'inputs', 'manifest_sha256', 'dependencies', 'runner_sha256',
+                                    'suite_runner_sha256', 'pytest_runner_sha256', 'shard_runner_sha256'), 'same')
+    docs = []
+    for index, assigned in enumerate(partition):
+      ids = [identity for identity in full if identity.startswith(assigned[0].removesuffix('.py') + '.')]
+      doc = {'suite': 'aol-lifecycle', 'status': 'completed', 'exit_code': 0, 'shard': {'index': index, 'count': 4},
+             'targets': assigned, 'selected_test_ids': ids, 'full_catalog_targets': targets, 'full_catalog_test_ids': full,
+             'collection_errors': [], 'gate_errors': [], 'source': source, 'source_after': source,
+             'native': {'safety': {'sha256': '1' * 64}, 'aol_protocol': {'sha256': '2' * 64}}}
+      result = {'selected_test_ids': ids, 'exit_code': 0, 'collection_errors': [], 'gate_errors': [],
+                'results': [{'id': identity, 'status': 'passed'} for identity in ids]}
+      docs.append((doc, result))
+    return docs
+
+  def check(self, docs):
+    with tempfile.TemporaryDirectory() as directory:
+      for index, (doc, result) in enumerate(docs):
+        path = Path(directory) / str(index)
+        path.mkdir()
+        (path / 'results.json').write_text(json.dumps(result))
+        doc['result_sha256'] = hashlib.sha256((path / 'results.json').read_bytes()).hexdigest()
+        (path / 'coverage.json').write_text(json.dumps(doc))
+      return helper.aggregate(directory, 4)
+
+  def test_complete_catalog_passes(self):
+    self.assertEqual(self.check(self.documents())['cases_passed'], 12)
+
+  def test_bad_outcomes_and_provenance_are_denied(self):
+    mutations = [lambda d: d.pop(),
+                 lambda d: d[0][0]['shard'].update(index=1),
+                 lambda d: d[0][0].update(exit_code=1),
+                 lambda d: d[0][0].update(gate_errors=['failed gate']),
+                 lambda d: d[0][0]['full_catalog_test_ids'].pop(),
+                 lambda d: d[0][0].update(targets=['foreign.py']),
+                 lambda d: d[0][0]['source'].update(revision='foreign'),
+                 lambda d: d[0][0]['source_after'].update(inputs='changed'),
+                 lambda d: d[0][0]['native']['safety'].update(sha256=''),
+                 lambda d: d[0][1]['results'].pop(),
+                 lambda d: d[0][1]['results'].append(d[0][1]['results'][0]),
+                 lambda d: d[0][1]['results'][0].update(status='skipped'),
+                 lambda d: d[0][1]['results'][0].update(status='failed'),
+                 lambda d: d[0][1].update(exit_code=1)]
+    for index, mutate in enumerate(mutations):
+      docs = copy.deepcopy(self.documents())
+      # Break intentional shared catalog/source aliases to exercise genuine cross-shard disagreement.
+      docs = json.loads(json.dumps(docs))
+      mutate(docs)
+      with self.subTest(index=index), self.assertRaises((ValueError, KeyError, StopIteration)):
+        self.check(docs)
+
+
+if __name__ == '__main__':
+  unittest.main()