judge-calibration.ts
386 lines · scripts/judge-calibration.ts at 3f79d81
1#!/usr/bin/env -S bun --no-env-file --config=/dev/null
2/**
3 * W2 judge calibration harness (comparison 1: schema transport, prose unchanged).
4 *
5 * For every corpus item of a request configuration it draws three independent
6 * 3-sample panels: two with the current request ("old-a", "old-b") and one with
7 * the schema-transport request ("new"). Each sample is one direct callJudge
8 * call. Calibration never goes through judgePanel or the workflow passing-panel
9 * cache: a replayed panel would make old-vs-old flips read zero.
10 *
11 * Each sample records a verdict-bearing value or an error kind (refusal,
12 * truncation, parse, schema, transport), its latency and its priced usage.
13 * Errors never count as verdicts; flip rates use items where both panels
14 * completed (all three samples valid).
15 *
16 * Usage:
17 * bun run scripts/judge-calibration.ts manifest [--budget 90]
18 * bun run scripts/judge-calibration.ts run --config <id> --budget <usd> [--concurrency 6]
19 * bun run scripts/judge-calibration.ts resume --config <id> --budget <usd> (only never-dispatched planned calls)
20 * bun run scripts/judge-calibration.ts analyze --config <id>
21 */
22import * as fs from 'node:fs';
23import * as path from 'node:path';
24import { createHash } from 'node:crypto';
25import Anthropic from '@anthropic-ai/sdk';
26import { callJudge, JudgeRefusalError, type CallJudgeOptions, type JudgeResponseMeta } from '../test/helpers/llm-judge';
27import { JUDGE_MS } from '../test/helpers/eval-budgets';
28import { estimateCostUsd } from '../test/helpers/pricing';
30export const PANEL_SAMPLES = 3;
31export const PHASES = ['old-a', 'old-b', 'new'] as const;
32export type Phase = typeof PHASES[number];
33export type Verdict = 'pass' | 'fail';
34export type ErrorKind = 'refusal' | 'truncation' | 'parse' | 'schema' | 'transport';
35export const LATENCY_CEILING_MS = 0.8 * JUDGE_MS;
36export const CORPUS_ROOT = path.join(import.meta.dir, '..', 'test', 'fixtures', 'judge-calibration');
38export interface CorpusItem {
39 id: string;
40 category: 'pass' | 'fail' | 'adjacent';
41 expected: Verdict;
42 split: 'dev' | 'heldout';
43 rationale: string;
44 inputs: Record<string, string>;
45}
47export interface CalibrationConfig {
48 id: string;
49 /** Dispatch order when the priced manifest exceeds the budget (armJudge, inline judges, workflow by case count). */
50 priority: number;
51 /** Eval cases that send this request configuration. */
52 cases: string[];
53 model: string;
54 build(inputs: Record<string, string>): string;
55 /** Today's request options. */
56 oldOptions: CallJudgeOptions;
57 /** Comparison (1): today's options plus the JSON schema; prose unchanged. */
58 newOptions: CallJudgeOptions;
59 /** Throws when production would reject the sample's shape. */
60 validate(value: unknown): void;
61 verdict(samples: unknown[]): Verdict;
62 /** Manifest pricing estimate per call (output includes adaptive thinking). */
63 estimatedOutputTokens: number;
64}
66export interface SampleRecord {
67 config: string;
68 item: string;
69 split: CorpusItem['split'];
70 phase: Phase;
71 sample: number;
72 status: 'ok' | 'error';
73 error_kind?: ErrorKind;
74 error?: string;
75 value?: unknown;
76 latency_ms: number;
77 stop_reason: string | null;
78 usage: { input_tokens: number; output_tokens: number } | null;
79 cost_usd: number;
80}
82export type JudgeCall = (prompt: string, model: string, opts: CallJudgeOptions) => Promise<unknown>;
84export function loadCorpus(configId: string, root = CORPUS_ROOT): CorpusItem[] {
85 const dir = path.join(root, configId);
86 const corpus = JSON.parse(fs.readFileSync(path.join(dir, 'corpus.json'), 'utf8'));
87 return corpus.items.map((item: any) => ({
88 ...item,
89 inputs: Object.fromEntries(Object.entries(item.inputs as Record<string, string | { file: string }>).map(([key, value]) =>
90 [key, typeof value === 'string' ? value : fs.readFileSync(path.join(dir, value.file), 'utf8')])),
91 }));
92}
94export function classifyError(error: unknown): ErrorKind {
95 if (error instanceof JudgeRefusalError) return 'refusal';
96 const message = error instanceof Error ? error.message : String(error);
97 if (/truncated at max_tokens|stop_reason=max_tokens/.test(message)) return 'truncation';
98 if (error instanceof SyntaxError || /non-JSON|did not complete/.test(message)) return 'parse';
99 return 'transport';
100}
102/** The fixed request identity: what must not move between dispatch and analysis. */
103export function requestHash(config: CalibrationConfig, phase: Phase): string {
104 const { onResponse: _ignored, signal: _signal, ...options } = phase === 'new' ? config.newOptions : config.oldOptions;
105 return createHash('sha256').update(JSON.stringify({ model: config.model, options })).digest('hex');
106}
108export interface PlannedCall { item: CorpusItem; phase: Phase; sample: number }
110/** Every paid call for one configuration, interleaved by a fixed seed so phases share time-of-day drift. */
111export function planCalls(items: CorpusItem[], split: CorpusItem['split']): PlannedCall[] {
112 const calls = items.filter(item => item.split === split)
113 .flatMap(item => PHASES.flatMap(phase => Array.from({ length: PANEL_SAMPLES }, (_, sample) => ({ item, phase, sample }))));
114 let seed = 0x5eed;
115 const random = () => (seed = (seed * 1103515245 + 12345) % 2 ** 31) / 2 ** 31;
116 for (let i = calls.length - 1; i > 0; i--) {
117 const j = Math.floor(random() * (i + 1));
118 [calls[i], calls[j]] = [calls[j]!, calls[i]!];
119 }
120 return calls;
121}
123/** Planned calls never dispatched (a budget or infrastructure stop); completing them is not a resample. */
124export function undispatched(calls: PlannedCall[], records: SampleRecord[]): PlannedCall[] {
125 const done = new Set(records.map(record => `${record.item}/${record.phase}/${record.sample}`));
126 return calls.filter(call => !done.has(`${call.item.id}/${call.phase}/${call.sample}`));
127}
129export async function runSample(config: CalibrationConfig, planned: PlannedCall, call: JudgeCall = callJudge as JudgeCall,
130 clock: () => number = () => performance.now()): Promise<SampleRecord> {
131 let meta: JudgeResponseMeta | null = null;
132 const options = { ...(planned.phase === 'new' ? config.newOptions : config.oldOptions), onResponse: (m: JudgeResponseMeta) => { meta = m; } };
133 const base = { config: config.id, item: planned.item.id, split: planned.item.split, phase: planned.phase, sample: planned.sample };
134 const started = clock();
135 let value: unknown;
136 let failure: { kind: ErrorKind; message: string } | null = null;
137 try {
138 value = await call(config.build(planned.item.inputs), config.model, options);
139 try { config.validate(value); }
140 catch (error) { failure = { kind: 'schema', message: error instanceof Error ? error.message : String(error) }; }
141 } catch (error) {
142 failure = { kind: classifyError(error), message: error instanceof Error ? error.message : String(error) };
143 }
144 const latency_ms = Math.round(clock() - started);
145 const response = meta as JudgeResponseMeta | null;
146 const usage = response?.usage ?? null;
147 const cost_usd = usage ? estimateCostUsd({ input: usage.input_tokens, output: usage.output_tokens }, config.model) : 0;
148 const shared = { latency_ms, stop_reason: response?.stop_reason ?? null, usage, cost_usd };
149 return failure
150 ? { ...base, status: 'error', error_kind: failure.kind, error: failure.message.slice(0, 500), ...shared }
151 : { ...base, status: 'ok', value, ...shared };
152}
154export interface RunOptions {
155 budgetUsd: number;
156 concurrency?: number;
157 call?: JudgeCall;
158 clock?: () => number;
159 onRecord?: (record: SampleRecord) => void;
160}
162/**
163 * Dispatch planned calls with bounded concurrency. Stops dispatching when the
164 * actual spend plus one in-flight estimate per open slot would pass the budget,
165 * or after two consecutive transport failures (infrastructure, not a verdict).
166 */
167export async function runCalls(config: CalibrationConfig, calls: PlannedCall[], opts: RunOptions): Promise<{ records: SampleRecord[]; stopped: string | null }> {
168 const records: SampleRecord[] = [];
169 const perCall = estimateCostUsd({ input: 0, output: config.estimatedOutputTokens }, config.model);
170 let spent = 0, next = 0, consecutiveTransport = 0, inFlight = 0;
171 let stopped: string | null = null;
172 const worker = async () => {
173 while (!stopped && next < calls.length) {
174 if (spent + (inFlight + 1) * perCall * 2 > opts.budgetUsd) { stopped = `budget: spent $${spent.toFixed(2)} of $${opts.budgetUsd.toFixed(2)}`; break; }
175 const planned = calls[next++]!;
176 inFlight++;
177 const record = await runSample(config, planned, opts.call, opts.clock);
178 inFlight--;
179 spent += record.cost_usd;
180 consecutiveTransport = record.error_kind === 'transport' ? consecutiveTransport + 1 : 0;
181 records.push(record);
182 opts.onRecord?.(record);
183 if (consecutiveTransport >= 2) stopped = `infrastructure: two consecutive transport failures (${record.error})`;
184 }
185 };
186 await Promise.all(Array.from({ length: Math.max(1, opts.concurrency ?? 6) }, worker));
187 return { records, stopped };
188}
190export interface PanelResult { verdict: Verdict | null; errors: number }
192export function panels(config: CalibrationConfig, records: SampleRecord[]): Map<string, Record<Phase, PanelResult>> {
193 const byItem = new Map<string, Record<Phase, PanelResult>>();
194 for (const item of new Set(records.map(record => record.item))) {
195 const entry = {} as Record<Phase, PanelResult>;
196 for (const phase of PHASES) {
197 const samples = records.filter(record => record.item === item && record.phase === phase);
198 const ok = samples.filter(record => record.status === 'ok');
199 entry[phase] = { verdict: ok.length === PANEL_SAMPLES && samples.length === PANEL_SAMPLES ? config.verdict(ok.map(record => record.value)) : null,
200 errors: samples.length - ok.length };
201 }
202 byItem.set(item, entry);
203 }
204 return byItem;
205}
207function p95(values: number[]): number | null {
208 if (values.length === 0) return null;
209 const sorted = [...values].sort((a, b) => a - b);
210 return sorted[Math.min(sorted.length - 1, Math.ceil(0.95 * sorted.length) - 1)]!;
211}
213export interface ComparisonSummary {
214 items: number;
215 old_flips: { flips: number; compared: number };
216 new_flips: { flips: number; compared: number };
217 agreement: Record<Phase, number>;
218 false_passes: Record<Phase, string[]>;
219 new_false_passes: string[];
220}
222export function compare(config: CalibrationConfig, items: CorpusItem[], records: SampleRecord[]): ComparisonSummary {
223 const byItem = panels(config, records);
224 const flips = (a: Phase, b: Phase) => {
225 const both = items.map(item => byItem.get(item.id)).filter(entry => entry?.[a].verdict && entry[b].verdict);
226 return { flips: both.filter(entry => entry![a].verdict !== entry![b].verdict).length, compared: both.length };
227 };
228 const agreement = Object.fromEntries(PHASES.map(phase =>
229 [phase, items.filter(item => byItem.get(item.id)?.[phase].verdict === item.expected).length])) as Record<Phase, number>;
230 const falsePasses = Object.fromEntries(PHASES.map(phase =>
231 [phase, items.filter(item => item.expected === 'fail' && byItem.get(item.id)?.[phase].verdict === 'pass').map(item => item.id)])) as Record<Phase, string[]>;
232 return { items: items.length, old_flips: flips('old-a', 'old-b'), new_flips: flips('new', 'old-a'), agreement,
233 false_passes: falsePasses, new_false_passes: falsePasses.new.filter(id => !falsePasses['old-a'].includes(id)) };
234}
236export interface CalibrationResult {
237 config: string;
238 model: string;
239 request_hash: Record<'old' | 'new', string>;
240 complete: boolean;
241 stopped: string | null;
242 calls: number;
243 cost_usd: number;
244 errors: Record<'old' | 'new', Record<ErrorKind, number> & { samples: number; rate: number }>;
245 latency_p95_ms: Record<'old' | 'new', number | null>;
246 all: ComparisonSummary;
247 heldout: ComparisonSummary;
248 landing: { lands: boolean; reasons: string[]; strict_no_flip_increase: boolean };
249 prose_step_triggered: { triggered: boolean; reasons: string[] };
250}
252export function analyze(config: CalibrationConfig, items: CorpusItem[], records: SampleRecord[], stopped: string | null = null): CalibrationResult {
253 const prompt = (phase: Phase) => phase === 'new' ? 'new' : 'old';
254 const errors = Object.fromEntries((['old', 'new'] as const).map(which => {
255 const samples = records.filter(record => prompt(record.phase) === which);
256 const counts = Object.fromEntries((['refusal', 'truncation', 'parse', 'schema', 'transport'] as const)
257 .map(kind => [kind, samples.filter(record => record.error_kind === kind).length])) as Record<ErrorKind, number>;
258 const errored = samples.filter(record => record.status === 'error').length;
259 return [which, { ...counts, samples: samples.length, rate: samples.length ? errored / samples.length : 0 }];
260 })) as CalibrationResult['errors'];
261 const latency = Object.fromEntries((['old', 'new'] as const).map(which =>
262 [which, p95(records.filter(record => prompt(record.phase) === which && record.status === 'ok').map(record => record.latency_ms))])) as CalibrationResult['latency_p95_ms'];
263 const expectedCalls = items.length * PHASES.length * PANEL_SAMPLES;
264 const complete = records.length === expectedCalls && !stopped;
265 const all = compare(config, items, records);
266 const heldout = compare(config, items.filter(item => item.split === 'heldout'), records);
268 const reasons: string[] = [];
269 if (!complete) reasons.push(`incomplete: ${records.length}/${expectedCalls} calls${stopped ? ` (${stopped})` : ''}`);
270 if (all.new_flips.flips > all.old_flips.flips + 1) reasons.push(`new flips ${all.new_flips.flips}/${all.new_flips.compared} > old flips ${all.old_flips.flips}/${all.old_flips.compared} + 1`);
271 if (all.agreement.new < all.agreement['old-a']) reasons.push(`expected-verdict agreement dropped: new ${all.agreement.new} < old ${all.agreement['old-a']}`);
272 if (all.new_false_passes.length > 0) reasons.push(`new false passes: ${all.new_false_passes.join(', ')}`);
273 if (latency.new !== null && latency.new > LATENCY_CEILING_MS) reasons.push(`new p95 latency ${latency.new} ms > ${LATENCY_CEILING_MS} ms (80% of JUDGE_MS)`);
275 const trigger: string[] = [];
276 if (records.some(record => record.phase === 'new' && record.status === 'error')) trigger.push('new prompt has errored samples');
277 if (reasons.length > 0) trigger.push('a landing condition failed');
278 if (all.old_flips.flips >= 3) trigger.push(`old-vs-old flips ${all.old_flips.flips} >= 3`);
280 return {
281 config: config.id, model: config.model,
282 request_hash: { old: requestHash(config, 'old-a'), new: requestHash(config, 'new') },
283 complete, stopped, calls: records.length,
284 cost_usd: +records.reduce((sum, record) => sum + record.cost_usd, 0).toFixed(4),
285 errors, latency_p95_ms: latency, all, heldout,
286 landing: { lands: reasons.length === 0, reasons, strict_no_flip_increase: all.new_flips.flips <= all.old_flips.flips },
287 prose_step_triggered: { triggered: trigger.length > 0, reasons: trigger },
288 };
289}
291export interface ManifestEntry {
292 config: string;
293 priority: number;
294 cases: number;
295 model: string;
296 items: number;
297 calls: number;
298 input_tokens: number;
299 estimated_output_tokens_per_call: number;
300 estimated_cost_usd: number;
301}
303/** Price every planned call before dispatch; count_tokens is free. */
304export async function priceManifest(configs: CalibrationConfig[], countTokens: (model: string, prompt: string) => Promise<number>,
305 root = CORPUS_ROOT): Promise<ManifestEntry[]> {
306 const entries: ManifestEntry[] = [];
307 for (const config of [...configs].sort((a, b) => a.priority - b.priority)) {
308 if (!fs.existsSync(path.join(root, config.id, 'corpus.json'))) continue;
309 const items = loadCorpus(config.id, root);
310 let input = 0;
311 for (const item of items) input += await countTokens(config.model, config.build(item.inputs)) * PHASES.length * PANEL_SAMPLES;
312 const calls = items.length * PHASES.length * PANEL_SAMPLES;
313 entries.push({ config: config.id, priority: config.priority, cases: config.cases.length, model: config.model, items: items.length, calls,
314 input_tokens: input, estimated_output_tokens_per_call: config.estimatedOutputTokens,
315 estimated_cost_usd: estimateCostUsd({ input, output: calls * config.estimatedOutputTokens }, config.model) });
316 }
317 return entries;
318}
320function flag(args: string[], name: string): string | undefined {
321 const index = args.indexOf(`--${name}`);
322 return index >= 0 ? args[index + 1] : undefined;
323}
325/** The W2 per-judge table plus the arm benchmark's paired judge (run by --config only; the manifest prices the W2 table). */
326export async function loadCalibrationConfigs(): Promise<{ w2: CalibrationConfig[]; all: CalibrationConfig[] }> {
327 // The per-judge table (builders, schemas, verdict rules) is committed separately from this harness.
328 const { CALIBRATION_CONFIGS } = await import(path.join(import.meta.dir, 'judge-calibration-configs.ts')) as { CALIBRATION_CONFIGS: CalibrationConfig[] };
329 const { ARM_PAIR_CALIBRATION } = await import(path.join(import.meta.dir, '..', 'test', 'helpers', 'arm-pair-judge.ts')) as { ARM_PAIR_CALIBRATION: CalibrationConfig };
330 return { w2: CALIBRATION_CONFIGS, all: [...CALIBRATION_CONFIGS, ARM_PAIR_CALIBRATION] };
331}
333async function main(args: string[]): Promise<void> {
334 const { w2: CALIBRATION_CONFIGS, all: configs } = await loadCalibrationConfigs();
335 const command = args[0];
336 if (command === 'manifest') {
337 const client = new Anthropic();
338 const entries = await priceManifest(CALIBRATION_CONFIGS, async (model, prompt) =>
339 (await client.messages.countTokens({ model, messages: [{ role: 'user', content: prompt }] })).input_tokens);
340 const budget = Number(flag(args, 'budget') ?? 90);
341 let running = 0;
342 const manifest = { generated_at: new Date().toISOString(), budget_usd: budget,
343 entries: entries.map(entry => ({ ...entry, cumulative_usd: +(running += entry.estimated_cost_usd).toFixed(4), within_budget: running <= budget })) };
344 fs.writeFileSync(path.join(CORPUS_ROOT, 'manifest.json'), JSON.stringify(manifest, null, 2) + '\n');
345 console.table(manifest.entries);
346 return;
347 }
348 const config = configs.find(candidate => candidate.id === flag(args, 'config'));
349 if (!config) throw new Error(`unknown --config; expected one of ${configs.map(candidate => candidate.id).join(', ')}`);
350 const items = loadCorpus(config.id);
351 const runsDir = path.join(CORPUS_ROOT, config.id, 'runs');
352 const samplesPath = path.join(runsDir, 'comparison-1.samples.jsonl');
353 const resultPath = path.join(runsDir, 'comparison-1.result.json');
354 if (command === 'run' || command === 'resume') {
355 // resume dispatches only planned calls a stop left undispatched; recorded samples are never redrawn.
356 if (command === 'run' && fs.existsSync(samplesPath)) throw new Error(`${samplesPath} exists; calibration runs are never repeated`);
357 if (command === 'resume' && !fs.existsSync(samplesPath)) throw new Error(`${samplesPath} is missing; nothing to resume`);
358 const prior: SampleRecord[] = command === 'resume' ? fs.readFileSync(samplesPath, 'utf8').trim().split('\n').map(line => JSON.parse(line)) : [];
359 fs.mkdirSync(runsDir, { recursive: true });
360 const budgetUsd = Number(flag(args, 'budget'));
361 if (!Number.isFinite(budgetUsd) || budgetUsd <= 0) throw new Error('--budget <usd> is required');
362 const concurrency = Number(flag(args, 'concurrency') ?? 6);
363 const onRecord = (record: SampleRecord) => fs.appendFileSync(samplesPath, JSON.stringify(record) + '\n');
364 let spent = 0, stopped: string | null = null;
365 // Development items first; the held-out third is dispatched only after, with the same frozen requests.
366 for (const split of ['dev', 'heldout'] as const) {
367 const outcome = await runCalls(config, undispatched(planCalls(items, split), prior), { budgetUsd: budgetUsd - spent, concurrency, onRecord });
368 spent += outcome.records.reduce((sum, record) => sum + record.cost_usd, 0);
369 console.log(`[calibration] ${config.id} ${split}: ${outcome.records.length} calls, $${spent.toFixed(2)} cumulative${outcome.stopped ? `, stopped: ${outcome.stopped}` : ''}`);
370 if ((stopped = outcome.stopped)) break;
371 }
372 fs.writeFileSync(path.join(runsDir, 'comparison-1.stop.json'), JSON.stringify({ stopped }) + '\n');
373 } else if (command !== 'analyze') {
374 throw new Error(`unknown command ${command}; expected manifest, run, resume or analyze`);
375 }
376 const records: SampleRecord[] = fs.readFileSync(samplesPath, 'utf8').trim().split('\n').map(line => JSON.parse(line));
377 const stopPath = path.join(runsDir, 'comparison-1.stop.json');
378 const stopped = fs.existsSync(stopPath) ? JSON.parse(fs.readFileSync(stopPath, 'utf8')).stopped : null;
379 const result = analyze(config, items, records, stopped);
380 fs.writeFileSync(resultPath, JSON.stringify(result, null, 2) + '\n');
381 console.log(JSON.stringify(result, null, 2));
382}
384if (import.meta.main) {
385 main(process.argv.slice(2)).catch(error => { console.error(error); process.exit(1); });
386}