Complete G8 fleet source-reference sweep with provenance exceptions

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a070b5-4994-7271-bd8b-7c3dbcedec4b

tools/test_fleet_references.py

@@ -0,0 +1,50 @@
+from pathlib import Path
+import hashlib
+import json
+import subprocess
+import tempfile
+import unittest
+
+from validate_fleet_references import scan, validate
+
+
+class FleetReferenceTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.repo = self.root / 'example'
+        self.repo.mkdir()
+        subprocess.run(['git', 'init', '-q'], cwd=self.repo, check=True)
+        self.legacy = 'identity' + '-canon'
+        self.ledger = self.root / 'review.json'
+        self.ledger.write_text(json.dumps({'repositories': ['example'], 'exceptions': []}))
+
+    def test_tracked_ignored_and_hidden_files_are_scanned(self):
+        (self.repo / '.gitignore').write_text('ignored.md\n')
+        (self.repo / 'ignored.md').write_text(self.legacy)
+        (self.repo / '.settings').write_text(self.legacy)
+        (self.repo / 'source-link').symlink_to('../' + self.legacy)
+        subprocess.run(['git', 'add', '-f', 'ignored.md'], cwd=self.repo, check=True)
+        _, _, found = scan(self.root)
+        self.assertEqual(set(found), {'example/ignored.md', 'example/.settings', 'example/source-link'})
+
+    def test_new_live_reference_is_rejected(self):
+        (self.repo / 'README.md').write_text('Source: ' + self.legacy)
+        result = validate(self.root, self.ledger)
+        self.assertEqual(result['result'], 'fail')
+        self.assertEqual(result['errors'][0]['reason'], 'unreviewed reference')
+
+    def test_review_is_bound_to_exact_reference_content(self):
+        text = 'Historical source: ' + self.legacy
+        (self.repo / 'README.md').write_text(text)
+        data = {'repositories': ['example'], 'exceptions': [{'file': 'example/README.md',
+                'reason': 'Historical source', 'line_sha256': [hashlib.sha256(text.encode()).hexdigest()]}]}
+        self.ledger.write_text(json.dumps(data))
+        self.assertEqual(validate(self.root, self.ledger)['result'], 'pass')
+        (self.repo / 'README.md').write_text(text + '\nCurrent source: ' + self.legacy)
+        self.assertEqual(validate(self.root, self.ledger)['result'], 'fail')
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/validate_fleet_references.py

@@ -0,0 +1,86 @@
+"""Validate reviewed legacy-reference exceptions across adjacent Git repositories.
+
+Scans tracked files plus nonignored untracked files, including hidden files and symlink target paths.
+Matches are fingerprints, not embedded file content. Git internals and ignored untracked
+dependency/cache trees are outside the source scan; live registry/cache
+and service checks are separate evidence.
+"""
+from __future__ import annotations
+import argparse
+from collections import Counter
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+
+PROJECT = Path(__file__).resolve().parents[1]
+PATTERN = re.compile('identity' + r'[-_ ]?canon(?![a-z])', re.I)
+AUDIT_FILES = {
+    'ledger/reference-sweep.json',
+    'docs/evidence/2026-09-06-reference-sweep.json',
+    'docs/evidence/2026-09-06-reference-sweep.md',
+    'docs/evidence/2026-09-06-reuse-registration.json',
+    'docs/evidence/2026-09-06-reuse-composed.json',
+    'docs/evidence/2026-09-06-hub-coordinates.json',
+}
+
+
+def scan(workspace):
+    repos = sorted(p for p in workspace.iterdir() if (p / '.git').exists())
+    matches = {}
+    total_files = 0
+    for repo in repos:
+        paths = subprocess.check_output(['git', 'ls-files', '-z', '--cached', '--others', '--exclude-standard'], cwd=repo)
+        for relative in sorted(set(p.decode() for p in paths.split(b'\0') if p)):
+            if repo.name == PROJECT.name and relative in AUDIT_FILES:
+                continue
+            path = repo / relative
+            if path.is_symlink():
+                # Audit the routing pointer even if its destination is absent.
+                data = os.readlink(path).encode()
+            elif path.is_file():
+                data = path.read_bytes()
+            else:
+                continue
+            total_files += 1
+            if b'\0' in data:
+                continue
+            lines = data.decode('utf-8', errors='replace').splitlines()
+            found = [{'line': i, 'sha256': hashlib.sha256(line.encode()).hexdigest()}
+                     for i, line in enumerate(lines, 1) if PATTERN.search(line)]
+            if found:
+                matches[f'{repo.name}/{relative}'] = found
+    return repos, total_files, matches
+
+
+def validate(workspace, ledger_path):
+    ledger = json.loads(ledger_path.read_text())
+    repos, file_count, found = scan(workspace)
+    accepted = {e['file']: e for e in ledger['exceptions']}
+    errors = []
+    for name, rows in found.items():
+        entry = accepted.get(name)
+        if not entry:
+            errors.append({'file': name, 'reason': 'unreviewed reference', 'lines': [r['line'] for r in rows]})
+        elif Counter(r['sha256'] for r in rows) != Counter(entry['line_sha256']):
+            errors.append({'file': name, 'reason': 'reference content changed; review required'})
+    for name in accepted.keys() - found.keys():
+        errors.append({'file': name, 'reason': 'obsolete exception; remove or review'})
+    missing = set(ledger['repositories']) - {p.name for p in repos}
+    errors.extend({'repo': r, 'reason': 'reviewed checkout missing'} for r in sorted(missing))
+    return {'gate': 'G8-source-scan', 'result': 'pass' if not errors else 'fail',
+            'repositories_scanned': len(repos), 'files_scanned': file_count,
+            'reviewed_exception_files': len(accepted), 'matching_lines': sum(map(len, found.values())),
+            'errors': errors}
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--workspace', type=Path, default=PROJECT.parent)
+    parser.add_argument('--ledger', type=Path, default=PROJECT / 'ledger/reference-sweep.json')
+    args = parser.parse_args()
+    result = validate(args.workspace.resolve(), args.ledger)
+    print(json.dumps(result, indent=2))
+    raise SystemExit(0 if result['result'] == 'pass' else 1)