Record complete corpus provenance distribution and G6 evidence

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

tools/validate_corpus.py

@@ -0,0 +1,75 @@
+"""Verify G6 source coverage, frozen snapshots, and exact destination fragments."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import re
+import subprocess
+
+import jsonschema
+import yaml
+
+
+def validate(project, commerce, info):
+    manifest = json.loads((project / 'ledger/corpus-distribution.json').read_text())
+    roots = {'commerce-canon': commerce, 'info-tech-canon': info}
+    base = 'infospace/assimilation/canon-federation'
+    commit = manifest['source_commit']
+    paths = subprocess.check_output(['git', 'ls-tree', '-r', '--name-only', commit, '--', 'research', 'terminology', 'scenarios'], cwd=commerce, text=True).splitlines()
+    listed = [row['source_path'] for row in manifest['files']]
+    assert set(paths) == set(listed) and len(paths) == len(listed), 'Incomplete or duplicate corpus coverage'
+    original = {}
+    copies = 0
+    for row in manifest['files']:
+        path = row['source_path']
+        data = subprocess.check_output(['git', 'show', commit + ':' + path], cwd=commerce)
+        original[path] = data
+        assert hashlib.sha256(data).hexdigest() == row['sha256'], 'Invalid source digest: ' + path
+        assert (commerce / path).read_bytes() == data, 'Original changed: ' + path
+        assert row['destinations'], 'Unrouted source: ' + path
+        assert row['applicable_models'], 'Missing destination interest: ' + path
+        for target in row['destinations']:
+            assert target['path'] == base + '/source/' + path, target
+            assert (roots[target['repo']] / target['path']).read_bytes() == data, target
+            copies += 1
+    schema = yaml.safe_load((info / 'infospace/schemas/assimilation.schema.yaml').read_text())
+    links = 0
+    for name, repo in roots.items():
+        workspace = repo / base
+        local = json.loads((workspace / 'distribution.json').read_text())
+        expected = [row for row in manifest['files'] if any(d['repo'] == name for d in row['destinations'])]
+        assert local['files'] == expected and local['source_commit'] == commit
+        expected_views = [row for row in manifest['views'] if any(d['repo'] == name for d in row['destinations'])]
+        assert local['views'] == expected_views
+        actual_paths = {str(p.relative_to(workspace / 'source')) for p in (workspace / 'source').rglob('*') if p.is_file()}
+        assert actual_paths == {row['source_path'] for row in expected}, name
+        record = yaml.safe_load((workspace / 'assimilation.yaml').read_text())
+        jsonschema.validate(record, schema)
+        assert record['disposition'] == 'observe' and record['status'] == 'closed'
+        assert set(record['source_files']) == {'source/' + row['source_path'] for row in expected}
+        for path in workspace.rglob('*.md'):
+            if 'source' in path.relative_to(workspace).parts: continue
+            for link in re.findall(r'\]\(([^)]+)\)', path.read_text()):
+                if '://' in link or link.startswith('#'): continue
+                assert (path.parent / link.split('#')[0]).exists(), (path, link)
+                links += 1
+    for row in manifest['views']:
+        source = original[row['source_path']].decode().splitlines(keepends=True)
+        assert 1 <= row['start_line'] <= row['end_line'] <= len(source), row
+        body = ''.join(source[row['start_line'] - 1:row['end_line']])
+        assert hashlib.sha256(body.encode()).hexdigest() == row['sha256'], row
+        for destination in row['destinations']:
+            for model in destination['models']:
+                view = (roots[destination['repo']] / base / 'views' / (model + '.md')).read_text()
+                assert body in view and row['sha256'] in view, (row['title'], model)
+    return {'source_files': len(paths), 'snapshot_copies': copies, 'exact_fragments': len(manifest['views']),
+            'local_links': links, 'source_commit': commit, 'unrouted_sources': 0,
+            'changed_originals': 0, 'status': 'pass'}
+
+
+if __name__ == '__main__':
+    p = argparse.ArgumentParser()
+    p.add_argument('--commerce-repo', type=Path, required=True)
+    p.add_argument('--info-tech-repo', type=Path, required=True)
+    args = p.parse_args()
+    print(json.dumps(validate(Path(__file__).resolve().parents[1], args.commerce_repo.resolve(), args.info_tech_repo.resolve()), indent=2))