Crafter.Review/tests/Goldens/derive.py

#!/usr/bin/env python3
# SPDX-License-Identifier: AGPL-3.0-only
# SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
"""Re-derive the goldens that follow from the lens table and the config defaults.

tests/Goldens/fixture/pipeline.json holds the recorded answers the suite
compares against. Where an entry follows from the lens table or the config
defaults, this script derives it from the rules instead of a hand edit: the
config cases through the binary's own `config` command (what Load, Validate,
LensesOff and Environments answer), the commit-truth lens's plans, the
not-reviewed lists and the estimates from the recorded units by the lens's
rules, and the gates from the table. Before it writes, it asserts that the
recorded table still reproduces the recorded entries, so a drift in the
derivation itself shows up as a failure here.

    python3 -I tests/Goldens/derive.py <checkout> <crafter-review binary>

Run from anywhere; review the diff of pipeline.json. The suite never runs it.
"""
import sys, json, hashlib, subprocess, pathlib, tempfile, os, shutil

W = pathlib.Path(sys.argv[1]); EXE = sys.argv[2]
G = W / 'tests/Goldens/fixture/pipeline.json'
g = json.loads(G.read_text())
TABLE = [('repo-better-solution.md', 'none'), ('repo-functional.md', 'browser'), ('repo-bug-prover.md', 'none'), ('repo-security.md', 'none'), ('repo-facts.md', 'none'),
         ('better-solution.md', 'none'), ('code-quality.md', 'none'), ('minimalism.md', 'none'), ('bug-prover.md', 'none'), ('facts.md', 'none'), ('consistency.md', 'none'),
         ('process-docs.md', 'none'), ('etiquette.md', 'none'), ('conciseness.md', 'none'), ('maintainer-persona.md', 'none')]
# The push gates' lenses: the plans and estimates of the series goldens are theirs.
REPO_TABLE = TABLE[:5]
GATES = [('Gate 1 — Approach', ['repo-better-solution.md']), ('Gate 2 — Functional', ['repo-functional.md']),
         ('Gate 3 — Proof', ['repo-bug-prover.md', 'repo-security.md']), ('Gate 4 — Commit truth', ['repo-facts.md'])]
name_of = lambda f: f.replace('.md', '')
sha = lambda s: hashlib.sha256(s.encode()).hexdigest()

# ---- config: every case through the binary --------------------------------------------
c = g['config']
scratch = pathlib.Path(tempfile.mkdtemp(prefix='goldens-patch-'))
def run_config(repo, extra=()):
    r = subprocess.run([EXE, *extra, 'config'], cwd=repo, capture_output=True, text=True, env={**os.environ, 'GIT_CONFIG_GLOBAL': '/dev/null', 'GIT_CONFIG_SYSTEM': '/dev/null'})
    lines = r.stdout.split('\n'); body, warnings, problems = [], [], []
    tail = False
    for l in lines:
        if l.startswith('warning: '): warnings.append(l[9:]); tail = True
        elif l.startswith('problem: '): problems.append(l[9:]); tail = True
        elif not tail: body.append(l)
    shown = json.loads('\n'.join(body))
    return shown, warnings, problems, r.returncode
def spelled(text, repo):
    text = text.replace(str(repo), '<repo>')
    return text.replace(f'`{EXE} slots`', '`<invocation> slots`')
# LensEntry: the config's entry for the lens, else its default one - the
# defaults as the binary answers them, so this holds whatever they are.
default_dir = scratch / 'defaults'; default_dir.mkdir()
DEFAULT_LENSES = run_config(default_dir)[0]['lenses']
def entry_of(lenses, name):
    e = lenses.get(name) if isinstance(lenses, dict) else None
    return e if isinstance(e, dict) else DEFAULT_LENSES[name]
changed_problems = {}
for case in c['cases']:
    repo = scratch / case['name']; repo.mkdir(parents=True)
    if case['body'] is not None:
        (repo / '.review-gates').mkdir()
        (repo / '.review-gates' / 'config.json').write_text(case['body'], encoding='utf-8', newline='')
    shown, warnings, problems, code = run_config(repo)
    path = shown.pop('_path')
    old_problems = case['problems']
    case['problems'] = [spelled(x, repo) for x in problems]
    case['warnings'] = [spelled(x, repo) for x in warnings]
    case['config'] = json.loads(spelled(json.dumps(shown), repo))
    case['path'] = None if path is None else spelled(path, repo)
    lenses = shown.get('lenses')
    case['lensesOff'] = [f for f, _ in TABLE if entry_of(lenses, name_of(f)).get('run') is False]
    case['environments'] = {f: entry_of(lenses, name_of(f)).get('environment', 'none') for f, _ in TABLE}
    if old_problems != case['problems']: changed_problems[case['name']] = (old_problems, case['problems'])
missing = scratch / 'missing'; missing.mkdir()
_, _, mprob, _ = run_config(missing, ['--config', str(missing / 'none.json')])
c['missing'] = [spelled(x, missing) for x in mprob]
# The defaults as the binary answers them: the 'none' case is Defaults() with
# _path null and the database password masked, which the record keeps unmasked.
none = next(case for case in c['cases'] if case['name'] == 'none')
defaults = json.loads(json.dumps(none['config']))
defaults['supabase']['dbUrl'] = c['defaults']['supabase']['dbUrl']
c['defaults'] = defaults
c['example'] = json.loads((W / 'process/templates/config.json').read_text())
ex = scratch / 'example'; ex.mkdir()
_, _, eprob, _ = run_config(ex, ['--config', str(W / 'process/templates/config.json')])
c['exampleProblems'] = [spelled(x, ex) for x in eprob]
shutil.rmtree(scratch)
print('config cases with changed problems:', len(changed_problems))
for n, (o, new) in changed_problems.items(): print('  ', n, '\n     -', o, '\n     +', new)

# ---- series: the facts lens's plans, not_reviewed, the estimate ------------------------
OPUS = 'claude-opus-5-5'
def facts_plan(entry):
    inv = entry['inventory']['units']; info = entry['units']
    blocks = []
    for u, meta in zip(inv, info):
        uid = u['id']; kind = meta['kind']
        if kind == 'code' and meta['doc']: kind = 'doc'
        common = {'lens': 'repo-facts', 'lensFile': 'repo-facts.md'}
        if kind == 'slot':
            if uid not in ('whole:order', 'whole:absent'): continue
            blocks.append({'label': f'repo-facts:slot:{uid}', **common, 'kind': 'slot', 'unit': uid, 'unitKind': 'slot', 'expectedIds': [uid], 'unitSha': sha(meta['text']),
                           'model': OPUS, 'maxTurns': 50, 'timeoutMin': 35})
        elif kind == 'doc':
            if uid.endswith('#hdr'):
                blocks.append({'label': f'repo-facts:na:{uid}', **common, 'kind': 'auto-na', 'unit': uid, 'expectedIds': [uid], 'unitSha': sha(meta['text']), 'basis': 'mechanical: diff header of a documentation file'})
            elif not any(l[:1] in ('+', '-') and l[1:].strip() for l in meta['norm'].split('\n')):
                blocks.append({'label': f'repo-facts:na:{uid}', **common, 'kind': 'auto-na', 'unit': uid, 'expectedIds': [uid], 'unitSha': sha(meta['text']), 'basis': 'mechanical: context only - the commit adds and removes no line in this part'})
            else:
                blocks.append({'label': f'repo-facts:block:{uid}', **common, 'kind': 'block', 'unit': uid, 'unitKind': 'doc', 'expectedIds': [uid], 'unitSha': sha(meta['text']), 'model': OPUS, 'maxTurns': 60, 'timeoutMin': 40})
        elif kind == 'text':
            blocks.append({'label': f'repo-facts:block:{uid}', **common, 'kind': 'block', 'unit': uid, 'unitKind': 'text', 'expectedIds': [uid], 'unitSha': sha(meta['text']), 'model': OPUS, 'maxTurns': 60, 'timeoutMin': 40})
    return {'whole': None, 'blocks': blocks}
def estimate(jobs, prefix_bytes=123457, lens_layer_bytes={'repo-better-solution': 4100, 'repo-functional': 9000}, tail_bytes=1800, tpb=3.7, sys_tokens=32000, out_per_turn=1200):
    P = prefix_bytes / tpb + sys_tokens
    total = {'cacheRead': 0.0, 'cacheWrite': 0.0, 'output': 0.0, 'processes': 0}
    seen = set()
    for j in jobs:
        if j['kind'] == 'auto-na': continue
        L = lens_layer_bytes.get(j['lens'], 3000) / tpb; T = tail_bytes / tpb
        k = 40 if j['kind'] == 'whole' else 10 if j['kind'] == 'slot' else 25 if j['lensFile'] == 'repo-functional.md' else 10 if j.get('unitKind') == 'code' else (12 if j['lensFile'] == 'repo-facts.md' else 8)
        key = (j['lens'], j['kind'])
        if key not in seen: seen.add(key); total['cacheWrite'] += P + L
        total['cacheWrite'] += T + k * 3000
        total['cacheRead'] += sum(P + L + T + 3000 * i for i in range(k))
        total['output'] += k * out_per_turn
        total['processes'] += 1
    total['costUnits'] = 1.25 * total['cacheWrite'] + 0.1 * total['cacheRead'] + 5 * total['output']
    total['listUsd'] = round((6.25 * total['cacheWrite'] + 0.5 * total['cacheRead'] + 25 * total['output']) / 1e6, 2)
    return total
for name, entry in g['series'].items():
    if 'plans' not in entry: continue
    jobs = []
    recorded = []
    for f, _ in REPO_TABLE:
        plan = entry['plans'].get(f)
        if plan is None: continue
        here = ([plan['whole']] if plan['whole'] is not None else []) + plan['blocks']
        recorded += here
        if f != 'repo-facts.md': jobs += here
    assert estimate(recorded) == entry['estimate'], f'{name}: the estimate formula does not reproduce the golden'
    fp = facts_plan(entry)
    entry['plans']['repo-facts.md'] = fp
    entry['plans']['repo-facts.md:exclude'] = fp
    covered = {b['unit'] for b in fp['blocks'] if b['kind'] != 'auto-na'}
    nr = entry['notReviewed']
    nr['all'] = [i for i in nr['all'] if i not in covered]
    nr['static'] = [i for i in nr['static'] if i not in covered]
    entry['estimate'] = estimate(jobs + fp['blocks'])
    print(f"series {name}: facts plan {len(fp['blocks'])} jobs ({sum(b['kind']=='auto-na' for b in fp['blocks'])} n/a), not reviewed all -> {nr['all']}")

# ---- gates ---------------------------------------------------------------------------------
def gates_for(table, only, without, off):
    out = []
    for title, lenses in table:
        kept = [l for l in lenses if (not only or l in only) and l not in without and l not in off]
        if kept: out.append([title, kept])
    return out
for args, old in g['gates']:
    only, without, off = args.get('only', []), args.get('without', []), args.get('off', [])
    assert old in (gates_for(GATES[:3], only, without, off), gates_for(GATES, only, without, off)), f'gates {args}: neither table reproduces the golden'
new_gates = [[args, gates_for(GATES, args.get('only', []), args.get('without', []), args.get('off', []))] for args, _ in g['gates']]
print('gates entries changed:', sum(1 for (a, o), (_, n) in zip(g['gates'], new_gates) if o != n), 'of', len(new_gates))
g['gates'] = new_gates

# ---- the charter ----------------------------------------------------------------------------
g['prompts']['charterSha']['repo-facts.md'] = sha((W / 'process/lenses/repo-facts.md').read_text())
G.write_text(json.dumps(g, indent=1, ensure_ascii=False) + '\n', encoding='utf-8')
print('written', G)