tegwick
1d02c83051
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a070b5-4994-7271-bd8b-7c3dbcedec4b
75 lines
4.2 KiB
Python
75 lines
4.2 KiB
Python
"""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))
|