The config defines the modes (modes.<name>): the kind of artifact the mode reviews - reply, series or commits, the three shipped partitions; a kind is a mode of its own name - the lenses it runs, in its kind's gate order, and the effort it runs at, else the kind's default (max for a reply or a series, xhigh for a commit range). `run --mode` names any of them, the dump records mode and kind, and everything that decided by the mode's name - the partition and hash, the prompts, the schemas, the disposition and reuse-key rules, what the CLI requires - decides by the kind. Every lens stays off until a mode lists it; lenses.<name>.run is retired with a warning that says so, and a lens of another kind, a name that is no lens, a lens listed twice and an unknown effort are config problems. The defaults run no lens; the template ships the three kinds as modes with every lens of their kind at the kind's effort, to trim (the owner's ruling of 2026-11-01), and init says so as it writes it - a kept config is the repository's own, and only one that lists no lens gets a step. The readers of the retired key follow: the CI smoke config, the shipped skill, the procedure (its description of the config as well as its mechanics), the manifest template, the run's messages and the comments that named the three modes. The run lock, the run numbers and the dump names are the kind's, so one full run per hash holds across the modes of a kind and their rounds count on together; the dump says which mode ran. The database broker starts for a lens the mode runs, as every other reader of the lenses off reads them. A named mode runs through the CLI in the kernel-modes test.
180 lines
11 KiB
Python
180 lines
11 KiB
Python
#!/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]
|
|
# The lenses of each artifact kind (Plan::KindLenses): a mode lists lenses of its kind.
|
|
KIND_LENSES = {'commits': [f for f, _ in REPO_TABLE], 'series': [f for f, _ in TABLE[5:]], 'reply': ['facts.md', 'etiquette.md', 'consistency.md', 'conciseness.md', 'maintainer-persona.md']}
|
|
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')
|
|
# The lenses no mode lists (Config::LensesOff with no mode): a mode's list counts for the lenses of its kind.
|
|
listed = set()
|
|
for mname, mode in (shown.get('modes') or {}).items():
|
|
if not isinstance(mode, dict): continue
|
|
kind = mname if mname in KIND_LENSES else mode.get('kind')
|
|
if kind in KIND_LENSES and isinstance(mode.get('lenses'), list):
|
|
listed |= {n + '.md' for n in mode['lenses'] if isinstance(n, str) and n + '.md' in KIND_LENSES[kind]}
|
|
case['lensesOff'] = [f for f, _ in TABLE if f not in listed]
|
|
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['prompts']['charterSha']['repo-better-solution.md'] = sha((W / 'process/lenses/repo-better-solution.md').read_text())
|
|
G.write_text(json.dumps(g, indent=1, ensure_ascii=False) + '\n', encoding='utf-8')
|
|
print('written', G)
|