frontend-patterns/scripts/research_evolution.py

#!/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()