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 comparison_windows(data):
    """Audit declared prospective groups; labels do not prove comparability."""
    groups = defaultdict(list)
    for event in data['ledger']['events']:
        if event['recording'] == 'prospective':
            groups[(event['stream'], event['comparison_group'])].append(event)
    windows = []
    for (stream, group), entries in sorted(groups.items()):
        entries = sorted(entries, key=lambda e: (e['date'], e['id']))
        unknown = [e['id'] for e in entries if e['effort']['research_minutes'] is None]
        zero = [e['id'] for e in entries if e['effort']['research_minutes'] == 0]
        measured = [e['id'] for e in entries
                    if e['effort']['research_minutes'] is not None and e['effort']['research_minutes'] > 0]
        due = len(entries) >= 3
        windows.append({
            'stream': stream,
            'comparison_group': group,
            'cycle_ids': [e['id'] for e in entries],
            'protocol_refs': list(dict.fromkeys(e['protocol_ref'] for e in entries)),
            'cycles_until_review_trigger': max(0, 3 - len(entries)),
            'qualitative_review_due': due,
            'positive_effort_cycle_ids': measured,
            'unknown_effort_cycle_ids': unknown,
            'zero_effort_cycle_ids': zero,
            'effort_complete': not unknown and not zero,
            'effort_comparison_ready': due and not unknown and not zero,
            'outcomes': {kind: sum(e['disposition'] == kind for e in entries)
                         for kind in sorted(DISPOSITIONS)},
        })
    return {
        'schema_version': 1,
        'basis': 'Declared prospective groups; manually review shared stream, evidence tier, question granularity, evaluation and effort accounting before comparing returns.',
        'review_trigger_cycles': 3,
        'limits': [
            'Group labels and cycle counts do not establish protocol comparability or a learning trend.',
            'Unknown or zero research time prevents effort comparison of the complete declared window; never estimate missing time or silently omit its cycle.',
            'A qualitative review may retain cycles with unknown effort; corroboration and inconclusive results remain visible.',
            'Readiness is a review trigger, not statistical evidence, a plateau declaration or a maturity judgment.',
        ],
        'windows': windows,
    }


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.', '']
    audit = comparison_windows(data)
    if not audit['windows']:
        lines += ['No prospective comparison windows yet. Learning rate and diminishing returns: **not measurable from this baseline**.', '']
    else:
        lines += ['### Comparison readiness', '',
                  'Declared groups require a review of shared evidence tier, question granularity, evaluation and effort accounting. A matching label is not proof of comparability. '
                  + link('research/evolution/ComparisonWindows.json', 'Machine-readable window audit') + '.', '',
                  '| Stream / declared group | Cycles | Positive effort | Unknown / zero effort | Qualitative review trigger | Complete-window effort comparison |',
                  '| --- | ---: | ---: | ---: | --- | --- |']
        for window in audit['windows']:
            lines.append(f'| {window["stream"]} / {cell(window["comparison_group"])} | {len(window["cycle_ids"])} | '
                         f'{len(window["positive_effort_cycle_ids"])} | {len(window["unknown_effort_cycle_ids"])} / {len(window["zero_effort_cycle_ids"])} | '
                         f'{"due" if window["qualitative_review_due"] else "pending"} | {"ready for protocol review" if window["effort_comparison_ready"] else "not ready"} |')
        lines += ['', 'Unknown or zero effort is not dropped to create a favorable window. Qualitative review can retain those cycles, but effort comparisons need an explicitly reviewed window with measured positive research time. No trend or plateau is inferred.', '']
    by_id = {e['id']: e for e in events}
    for window in audit['windows']:
        entries = [by_id[ident] for ident in window['cycle_ids']]
        lines += [f'### Stream {window["stream"]} / {cell(window["comparison_group"])}', '',
                  '| Cycle | Protocol | 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"]} | {link(e["protocol_ref"], "protocol")} | {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 declared cycles: qualitative review trigger pending; no trend assessment.' if not window['qualitative_review_due'] else 'At least three declared cycles: qualitative review trigger reached, subject to protocol comparability review; 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'
        audit_path = ROOT / DATA / 'ComparisonWindows.json'
        audit_output = json.dumps(comparison_windows(data), indent=2, ensure_ascii=False) + '\n'
        if args.command == 'check':
            require(path.read_text() == output, 'Stale report: run make evolution-report')
            require(audit_path.read_text() == audit_output, 'Stale comparison windows: run make evolution-report')
            print('Evolution records and generated report are consistent.')
        else:
            path.write_text(output)
            audit_path.write_text(audit_output)
            print(path.relative_to(ROOT))
            print(audit_path.relative_to(ROOT))
    except (ValueError, KeyError, OSError) as exc:
        parser.exit(1, f'Evolution validation failed: {exc}\n')


if __name__ == '__main__':
    main()