tegwick
7295172f1e
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e241-8285-7a63-8c0c-51c9cb824dc3
242 lines
15 KiB
Python
242 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate curated evolution records and render their deterministic report (stdlib)."""
|
|
import argparse
|
|
from collections import Counter, defaultdict
|
|
from datetime import date
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DATA = Path('research/evolution')
|
|
MATURITIES = {'hypothesis', 'emerging', 'established', 'canonical'}
|
|
DISPOSITIONS = {'revision', 'corroboration', 'rejection', 'merge', 'inconclusive'}
|
|
EFFORT = {'research_minutes', 'execution_minutes', 'reported_tokens', 'cost_usd'}
|
|
|
|
|
|
def require(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def load(root=ROOT):
|
|
return {name: json.loads((root / DATA / f'{name}.json').read_text())
|
|
for name in ('profiles', 'ledger', 'context')}
|
|
|
|
|
|
def ref_check(root, ref):
|
|
path = (root / ref).resolve()
|
|
require(path.is_relative_to(root.resolve()) and path.is_file(), f'Missing/unsafe reference: {ref}')
|
|
|
|
|
|
def validate(data, root=ROOT):
|
|
for name, record in data.items():
|
|
require(record['schema_version'] == 1, f'Unsupported schema: {name}')
|
|
profiles = data['profiles']['profiles']
|
|
ids = [p['id'] for p in profiles]
|
|
require(len(ids) == len(set(ids)), 'Duplicate candidate identity')
|
|
paths = [p['path'] for p in profiles]
|
|
require(len(paths) == len(set(paths)), 'Duplicate candidate path')
|
|
require(set(paths) == {str(p.relative_to(root)) for p in (root / 'research/candidates').glob('*.md')}, 'Candidate inventory mismatch')
|
|
for p in profiles:
|
|
require(re.fullmatch(r'CAND-\d{4}', p['id']), 'Invalid research identity')
|
|
ref_check(root, p['path'])
|
|
content = (root / p['path']).read_bytes()
|
|
require(hashlib.sha256(content).hexdigest() == p['sha256'], f'Stale profile: {p["path"]}')
|
|
for field in ('title', 'status', 'maturity'):
|
|
require(f'{field}: {p[field]}\n' in content.decode(), f'Metadata mismatch: {p["id"]} {field}')
|
|
require(p['maturity'] in MATURITIES, 'Invalid maturity')
|
|
require(p['decision'] and p['gaps'] and p['evidence'], 'Missing decision, limits or evidence')
|
|
require(p['review']['role'] in {'author-model', 'independent-model', 'human'}, 'Invalid reviewer role')
|
|
require(not p['review']['independent_human'] or p['review']['role'] == 'human', 'Model review cannot be human validation')
|
|
date.fromisoformat(p['review']['assessed_at'])
|
|
if p['review']['last_substantive_review_date'] is not None:
|
|
review = p['review']
|
|
require(date.fromisoformat(review['last_substantive_review_date']) <= date.fromisoformat(review['assessed_at']), 'Review date exceeds assessment')
|
|
require(review.get('substantive_review_ref') and review.get('substantive_review_scope', '').strip(), 'Substantive review needs provenance and scope')
|
|
ref_check(root, review['substantive_review_ref'])
|
|
require(p['independent_full_contract_uses'] is None or type(p['independent_full_contract_uses']) is int and p['independent_full_contract_uses'] >= 0, 'Invalid independent-use count')
|
|
require(set(p['superseded_by']) <= set(ids) - {p['id']}, 'Unknown/self successor')
|
|
for e in p['evidence']:
|
|
require(e['kind'] in {'documentary', 'static-code', 'runtime'}, 'Invalid evidence kind')
|
|
require(e['family'] and e['scope'], 'Unscoped evidence')
|
|
ref_check(root, e['ref'])
|
|
seen = set()
|
|
for e in data['ledger']['events']:
|
|
require(e['id'] not in seen, 'Duplicate learning identity')
|
|
require(re.fullmatch(r'LEARN-\d{4}', e['id']), 'Invalid learning identity')
|
|
require(set(e['supersedes']) <= seen, 'Supersession must reference earlier entries')
|
|
seen.add(e['id'])
|
|
date.fromisoformat(e['date'])
|
|
require(e['disposition'] in DISPOSITIONS, 'Invalid learning disposition')
|
|
require(e['recording'] in {'retrospective', 'prospective'}, 'Invalid recording mode')
|
|
require(e['stream'] in {'A', 'B', 'C', 'D', 'E', 'F', 'cross-stream'}, 'Invalid stream')
|
|
require(set(e['affected_patterns']) <= set(ids), 'Unknown affected pattern')
|
|
require(e['question'] and e['previous_belief'] and e['conclusion'] and e['remaining_uncertainty'] and e['evidence'], 'Incomplete learning entry')
|
|
require(not e['reviewer']['independent_human'] or e['reviewer']['role'] == 'human', 'Model review cannot be human validation')
|
|
require(set(e['effort']) == EFFORT, 'Effort fields mismatch')
|
|
for value in e['effort'].values():
|
|
require(value is None or type(value) in (int, float) and value >= 0 and value < float('inf'), 'Invalid effort')
|
|
for ref in e['evidence']:
|
|
ref_check(root, ref)
|
|
if e['recording'] == 'prospective':
|
|
require(e['comparison_group'] and e.get('protocol_ref'), 'Prospective cycle needs a comparison group and preregistered protocol')
|
|
ref_check(root, e['protocol_ref'])
|
|
cases = data['context']['cases']
|
|
require({c['id'] for c in cases} == {f'VC-{n:02d}' for n in range(1, 11)} and len(cases) == 10, 'Case coverage inventory mismatch')
|
|
for c in cases:
|
|
require(c['state'] in {'planned', 'synthetic draft', 'provisional', 'reviewed'}, 'Invalid case state')
|
|
ref_check(root, c['ref'])
|
|
for ref in data['context']['open_work']:
|
|
ref_check(root, ref)
|
|
u = data['context']['usefulness']
|
|
ref_check(root, u['ref'])
|
|
require(u['denominator'] > 0 and all(0 <= u[k] <= u['denominator'] for k in ('named_score', 'prose_score')), 'Invalid usefulness result')
|
|
|
|
|
|
def link(ref, label=None):
|
|
# Report is two directories below repository root.
|
|
return f'[{label or Path(ref).stem}](../../{ref})'
|
|
|
|
|
|
def cell(value):
|
|
return str(value).replace('|', '\\|').replace('\n', ' ')
|
|
|
|
|
|
def render(data, baseline=None):
|
|
profiles = data['profiles']['profiles']
|
|
events = data['ledger']['events']
|
|
context = data['context']
|
|
maturity = Counter(p['maturity'] for p in profiles)
|
|
kinds = Counter(k for p in profiles for k in {e['kind'] for e in p['evidence']})
|
|
lines = ['# Language evolution report', '',
|
|
'Generated by `make evolution-report`; curated assessment date: ' + data['profiles']['assessed_at'] + '.', '',
|
|
'This is an evidence profile, not a completion percentage. Evidence presence does not establish full-contract recurrence or production fitness.', '',
|
|
'The baseline profiles are author-model assessments by Codex; they do not confer independent validation.', '',
|
|
'## Current corpus', '',
|
|
f'{len(profiles)} candidates: ' + ', '.join(f'{n} {m}' for m, n in sorted(maturity.items())) + '.', '',
|
|
'| Evidence attached to a candidate | Candidates |', '| --- | ---: |']
|
|
lines += [f'| {kind} | {kinds[kind]} |' for kind in ('documentary', 'static-code', 'runtime')]
|
|
lines += ['', 'Categories overlap. Runtime includes authored fixtures and negative controls. Family labels group related evidence; they are not counts of independent applications.', '',
|
|
f'Independent human profile reviews: {sum(p["review"]["independent_human"] for p in profiles)}/{len(profiles)}. '
|
|
f'Full-contract independent-use counts unassessed: {sum(p["independent_full_contract_uses"] is None for p in profiles)}/{len(profiles)}.', '']
|
|
if baseline:
|
|
old = {p['id']: p for p in baseline['assessment']['profiles']['profiles']}
|
|
now = {p['id']: p for p in profiles}
|
|
shared = old.keys() & now.keys()
|
|
lines += ['## Change since baseline', '',
|
|
f'Baseline {baseline["id"]}; source revision `{baseline["source_revision"]}`. '
|
|
f'Added: {len(now.keys()-old.keys())}; removed: {len(old.keys()-now.keys())}; '
|
|
f'shared candidates with changed content: {sum(old[i]["sha256"] != now[i]["sha256"] for i in shared)}; '
|
|
f'changed evidence/review assessments: {sum(old[i] != now[i] for i in shared)}; '
|
|
f'maturity transitions: {sum(old[i]["maturity"] != now[i]["maturity"] for i in shared)}.', '',
|
|
'The shared cohort is compared separately from additions. Changed text or evidence counts are not learning gains.', '']
|
|
lines += ['## Applicability and usefulness', '', '| Case | State |', '| --- | --- |']
|
|
lines += [f'| {link(c["ref"], c["id"])} | {c["state"]} |' for c in context['cases']]
|
|
u = context['usefulness']
|
|
lines += ['', f'Frozen pilot after adjudication: named {u["named_score"]}/{u["denominator"]}; prose {u["prose_score"]}/{u["denominator"]}. '
|
|
f'Initial unsupported-claim count: {u["initial_unsupported_claims_each"]} per condition (historical only); adjudicated flags remain qualitative, with no revised numeric count. '
|
|
f'Independent human validation: {"recorded" if u["independent_human"] else "pending"}. '
|
|
f'Decision time: {"unknown" if u["decision_time_minutes"] is None else str(u["decision_time_minutes"])+" minutes"}. '
|
|
+ link(u['ref'], 'Adjudication') + '.', '', u['limit'], '',
|
|
'## Learning and effort', '', data['ledger']['coverage'], '',
|
|
'| Outcome | Entries (not independent discoveries) |', '| --- | ---: |']
|
|
counts = Counter(e['disposition'] for e in events)
|
|
lines += [f'| {k} | {counts[k]} |' for k in sorted(DISPOSITIONS)]
|
|
lines += ['', f'Cycles with recorded research effort: {sum(e["effort"]["research_minutes"] is not None for e in events)}/{len(events)}. '
|
|
'Unknown effort is neither zero nor estimated from commits, tokens or assertion counts.', '']
|
|
groups = defaultdict(list)
|
|
for e in events:
|
|
if e['recording'] == 'prospective':
|
|
groups[(e['stream'], e['comparison_group'])].append(e)
|
|
if not groups:
|
|
lines += ['No prospective comparison windows yet. Learning rate and diminishing returns: **not measurable from this baseline**.', '']
|
|
for (stream, group), entries in sorted(groups.items()):
|
|
entries = sorted(entries, key=lambda e: (e['date'], e['id']))
|
|
lines += [f'### Stream {stream} / {cell(group)}', '',
|
|
'| Cycle | Outcome | Research minutes | Consequential outcome/hour |', '| --- | --- | ---: | ---: |']
|
|
for e in entries:
|
|
minutes = e['effort']['research_minutes']
|
|
consequential = e['disposition'] in {'revision', 'rejection', 'merge'}
|
|
rate = f'{60/minutes:.2f}' if consequential and minutes else ('0.00' if minutes else 'unknown')
|
|
lines.append(f'| {e["id"]} | {e["disposition"]} | {minutes if minutes is not None else "unknown"} | {rate} |')
|
|
lines += ['', 'One bounded question counts at most one consequential outcome. Corroboration and inconclusive work remain visible separately. '
|
|
+ ('Fewer than three comparable cycles: no trend assessment.' if len(entries) < 3 else 'At least three cycles: eligible for a qualitative review of returns; this does not demonstrate a plateau.'), '']
|
|
lines += ['## Coherence and maintainability', '',
|
|
'Profile hashes and local references passed validation when this report was generated. That checks traceability, not source truth or external URL freshness.', '',
|
|
f'Profiles with unknown last substantive review date: {sum(p["review"]["last_substantive_review_date"] is None for p in profiles)}. '
|
|
'Assessment dates record this metadata review, not a new substantive validation. Recorded substantive dates identify scoped author-model reviews, not renewed empirical or external-source freshness.', '',
|
|
'Coherence metrics (ambiguous selections, unjustified relations and overlap resolved): not measured. Conflict Copy / Explicit Reconciliation overlap remains an explicit open question.', '',
|
|
'Review backlog and evidence gaps remain in ' + ', '.join(link(r) for r in context['open_work']) + '.', '',
|
|
'## Pattern profiles', '']
|
|
for p in profiles:
|
|
lines += [f'### {p["id"]} — {p["title"]}', '', link(p['path']) + f' · {p["status"]} / {p["maturity"]}', '', p['decision'], '',
|
|
'| Evidence | Family | Scope | Review |', '| --- | --- | --- | --- |']
|
|
lines += [f'| {e["kind"]} | {cell(e["family"])} | {cell(e["scope"])} | {link(e["ref"])} |' for e in p['evidence']]
|
|
lines += ['', 'Open limits: ' + ' '.join(p['gaps']), '',
|
|
'Superseded by: ' + (', '.join(p['superseded_by']) or 'none recorded') + '.', '']
|
|
lines += ['## Recorded learning', '']
|
|
for e in events:
|
|
lines += [f'### {e["id"]} — {e["question"]}', '',
|
|
f'{e["date"]} · stream {e["stream"]} · {e["recording"]} · {e["disposition"]}', '',
|
|
'Before: ' + e['previous_belief'], '', 'Conclusion: ' + e['conclusion'], '',
|
|
'Evidence: ' + ', '.join(link(r) for r in e['evidence']) + '.', '',
|
|
'Remaining uncertainty: ' + ' '.join(e['remaining_uncertainty']), '',
|
|
'Supersedes: ' + (', '.join(e['supersedes']) or 'none recorded') + '.', '']
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def snapshot(data, name, root=ROOT):
|
|
require(re.fullmatch(r'[a-z0-9-]+', name), 'Unsafe snapshot name')
|
|
revision = subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=root, text=True).strip()
|
|
for p in data['profiles']['profiles']:
|
|
blob = subprocess.check_output(['git', 'show', f'{revision}:{p["path"]}'], cwd=root)
|
|
require(hashlib.sha256(blob).hexdigest() == p['sha256'], 'Commit candidate changes before taking snapshot')
|
|
result = dict(schema_version=1, id=name, source_revision=revision, assessment=data)
|
|
path = root / DATA / 'snapshots' / (name + '.json')
|
|
with path.open('x') as f: # Never overwrite a historical snapshot.
|
|
f.write(json.dumps(result, indent=2, ensure_ascii=False) + '\n')
|
|
return path
|
|
|
|
|
|
|
|
def validate_snapshot(baseline, root=ROOT):
|
|
require(baseline['schema_version'] == 1, 'Unsupported snapshot schema')
|
|
require(re.fullmatch(r'[0-9a-f]{40}', baseline['source_revision']), 'Invalid snapshot revision')
|
|
for p in baseline['assessment']['profiles']['profiles']:
|
|
blob = subprocess.check_output(
|
|
['git', 'show', f'{baseline["source_revision"]}:{p["path"]}'], cwd=root)
|
|
require(hashlib.sha256(blob).hexdigest() == p['sha256'], 'Snapshot does not match its pinned source')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('command', choices=['generate', 'check', 'snapshot'])
|
|
parser.add_argument('--name')
|
|
parser.add_argument('--baseline', default='baseline-2026-09-26')
|
|
args = parser.parse_args()
|
|
try:
|
|
data = load()
|
|
validate(data)
|
|
if args.command == 'snapshot':
|
|
require(args.name, 'snapshot requires --name')
|
|
print(snapshot(data, args.name))
|
|
return
|
|
baseline = json.loads((ROOT / DATA / 'snapshots' / (args.baseline + '.json')).read_text())
|
|
validate_snapshot(baseline)
|
|
output = render(data, baseline)
|
|
path = ROOT / DATA / 'EvolutionReport.md'
|
|
if args.command == 'check':
|
|
require(path.read_text() == output, 'Stale report: run make evolution-report')
|
|
print('Evolution records and generated report are consistent.')
|
|
else:
|
|
path.write_text(output)
|
|
print(path.relative_to(ROOT))
|
|
except (ValueError, KeyError, OSError) as exc:
|
|
parser.exit(1, f'Evolution validation failed: {exc}\n')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|