recover-receipts.ts
329 lines · scripts/recover-receipts.ts at 3f79d81
1/**
2 * Cancellation recovery for the PR lane's receipt store (W1c: CEO-14/25/34,
3 * ENG-4/5/7/8).
4 *
5 * A PR run cancelled by the next push never reaches its report, so its
6 * finished receipts never enter the store the report saves. The
7 * `recover-receipts` job in evals.yml runs this script from the base ref
8 * (never PR code) with `actions: read`, before `plan-slices`:
9 *
10 * collect restores nothing itself: it reads the store the job restored,
11 * lists this PR's evals.yml runs newer than that store (plus runs
12 * the checkpoint left unresolved), downloads the receipts of
13 * completed cancelled runs (at most 5, merged oldest first,
14 * negatives first), reconstructs completed behavior panels from
15 * their manifest and slice results to publish panel negatives
16 * (partial panels get no credit), and writes the checkpoint
17 * `recovery.json` into the store.
18 * decide (plan-slices) reads the checkpoint and turns reuse off when
19 * recovery degraded or a cancelled run is still unresolved, so a
20 * FAIL that may exist is never bypassed by an older PASS.
21 *
22 * Recovery never fails the run: a gh error, rate limit, missing artifact or
23 * an exhausted download budget degrades to "no recovered receipts" with a
24 * named reason, and the run executes unreduced.
25 */
26import * as fs from 'node:fs';
27import * as os from 'node:os';
28import * as path from 'node:path';
29import { spawnSync } from 'node:child_process';
30import { mergeReceiptDirs, writeNegativeReceipt } from './e2e-shard-reuse';
31import { loadSliceArtifacts, panelReports, parseRunManifest, shardFile } from './test-paid-shards';
33export const RECOVERY_FILE = 'recovery.json';
34export const MAX_RECOVERED_RUNS = 5;
35const PROCESSED_KEEP = 200;
37export interface RecoveryCheckpoint {
38 schema: 1;
39 updatedAt: number;
40 mode: 'pr' | 'dispatch';
41 /** Run ids whose evidence is merged (or that needed none). */
42 processed: string[];
43 /** Cancelled runs whose evidence could not be read yet; reuse stays off until a later run reconciles them. */
44 unresolved: Array<{ runId: string; reason: string }>;
45 /** Why recovery could not run at all; reuse stays off. */
46 degraded?: string;
47 recovered: number;
48}
50export interface WorkflowRun {
51 id: number;
52 status: string;
53 conclusion: string | null;
54 event: string;
55 head_sha: string;
56 run_attempt?: number;
57 updated_at?: string;
58 head_repository?: { full_name?: string } | null;
59 pull_requests?: Array<{ number: number }>;
60 display_title?: string;
61}
63/** One `gh` invocation; injectable so free tests drive recovery from recorded responses. */
64export type GhRunner = (args: string[], timeoutMs: number) => { status: number | null; stdout: string; stderr: string };
66export const runGh: GhRunner = (args, timeoutMs) => {
67 const result = spawnSync('gh', args, { encoding: 'utf8', timeout: Math.max(1_000, timeoutMs), maxBuffer: 64 * 1024 * 1024 });
68 return { status: result.status, stdout: result.stdout ?? '', stderr: result.error ? String(result.error) : result.stderr ?? '' };
69};
71/** A gh failure as a named reason with the next step (DX-6). */
72export function ghFailureReason(stderr: string): string {
73 if (/rate limit|secondary rate|HTTP 429/i.test(stderr)) {
74 const reset = /reset[^0-9]*(\d{9,})/i.exec(stderr)?.[1];
75 return `GitHub API rate limit${reset ? ` (resets at ${new Date(Number(reset) * 1000).toISOString()})` : ''}; rerun the job after the reset`;
76 }
77 if (/HTTP 40[13]|authentication|auth login/i.test(stderr)) return 'gh is not authorized for actions: read; check the job permissions and `gh auth status`';
78 if (/ETIMEDOUT|timed out/i.test(stderr)) return 'gh timed out inside the 60 s recovery budget; rerun the job';
79 return `gh failed: ${stderr.trim().split('\n')[0]?.slice(0, 160) || 'no output'}`;
80}
82export function readCheckpoint(store: string): RecoveryCheckpoint | null {
83 try {
84 const parsed = JSON.parse(fs.readFileSync(path.join(store, RECOVERY_FILE), 'utf8'));
85 if (parsed?.schema !== 1 || !Array.isArray(parsed.processed) || !Array.isArray(parsed.unresolved)) return null;
86 return parsed as RecoveryCheckpoint;
87 } catch { return null; }
88}
90function writeCheckpoint(store: string, checkpoint: RecoveryCheckpoint): void {
91 fs.mkdirSync(store, { recursive: true });
92 fs.writeFileSync(path.join(store, RECOVERY_FILE), `${JSON.stringify(checkpoint, null, 2)}\n`, { mode: 0o600 });
93}
95/** The run id that saved a restored store, from its cache key (`…-pr-<n>-<run id>-<attempt>-merged`). */
96export function storeRunId(cacheKey: string): number {
97 const match = /-pr-\d+-(\d+)-\d+-(?:merged|plan)$/.exec(cacheKey);
98 return match ? Number(match[1]) : 0;
99}
101/** CEO-34 provenance: same workflow event, PR number and head repository, never this run. */
102export function acceptedRun(run: WorkflowRun, scope: { pr: number; headRepo: string; selfRunId: number }): boolean {
103 return run.id !== scope.selfRunId && run.event === 'pull_request'
104 && run.head_repository?.full_name === scope.headRepo
105 && (run.pull_requests ?? []).some(pr => pr.number === scope.pr);
106}
108/** The PR a dispatched screen names in its run name (evals.yml: `E2E Evals [eval-bg <nonce>] screen #<n>`); null for any other run. */
109export function screenRunPr(run: Pick<WorkflowRun, 'event' | 'display_title'>): number | null {
110 const match = run.event === 'workflow_dispatch' ? /^E2E Evals \[eval-bg [A-Za-z0-9._-]+\] screen #(\d+)$/.exec(run.display_title ?? '') : null;
111 return match ? Number(match[1]) : null;
112}
114/** A dispatched screen of this PR from the base repository (same-repository PRs only), never this run. */
115export function acceptedScreenRun(run: WorkflowRun, scope: { pr: number; repo: string; selfRunId: number }): boolean {
116 return run.id !== scope.selfRunId && screenRunPr(run) === scope.pr && run.head_repository?.full_name === scope.repo;
117}
119function listRuns(gh: GhRunner, repo: string, filter: string[], timeoutMs: number): WorkflowRun[] | { error: string } {
120 const listed = gh(['api', '-X', 'GET', `repos/${repo}/actions/workflows/evals.yml/runs`, ...filter.flatMap(f => ['-f', f]), '-f', 'per_page=50'], timeoutMs);
121 if (listed.status !== 0) return { error: ghFailureReason(listed.stderr) };
122 try { return (JSON.parse(listed.stdout).workflow_runs ?? []) as WorkflowRun[]; } catch { return { error: 'gh returned an unreadable run list' }; }
123}
125/**
126 * The PR's evals.yml runs whose results may carry over: its pull_request runs
127 * and, with `screens`, its dispatched screens (both events, filtered by the PR).
128 */
129export function listGateRuns(gh: GhRunner, o: { repo: string; pr: number; headRepo: string; branch: string; selfRunId: number; screens: boolean },
130 left: () => number): { runs: WorkflowRun[] } | { error: string } {
131 const prRuns = listRuns(gh, o.repo, ['event=pull_request', `branch=${o.branch}`], left());
132 if ('error' in prRuns) return prRuns;
133 const runs = prRuns.filter(run => acceptedRun(run, { pr: o.pr, headRepo: o.headRepo, selfRunId: o.selfRunId }));
134 if (!o.screens) return { runs };
135 const screens = listRuns(gh, o.repo, ['event=workflow_dispatch'], left());
136 if ('error' in screens) return screens;
137 return { runs: [...runs, ...screens.filter(run => acceptedScreenRun(run, { pr: o.pr, repo: o.repo, selfRunId: o.selfRunId }))] };
138}
140/**
141 * Negative receipts a cancelled run's completed work proves (ENG-5): every
142 * completed FAIL panel whose trials share one input identity, and every
143 * completed non-passing single-trial shard. Partial panels publish nothing.
144 */
145export function negativesFromRun(runDir: string, run: Pick<WorkflowRun, 'id' | 'head_sha' | 'updated_at'>, out: string): number {
146 let manifest;
147 try { manifest = parseRunManifest(fs.readFileSync(path.join(runDir, 'paid-plan', 'manifest.json'), 'utf8')); } catch { return 0; }
148 const artifacts = loadSliceArtifacts(runDir);
149 let written = 0;
150 for (const attempt of [...new Set(artifacts.map(a => a.result.attempt ?? 1))]) {
151 const results = artifacts.map(a => a.result).filter(r => (r.attempt ?? 1) === attempt);
152 const completedAt = Math.max(0, ...results.map(r => r.finishedAt ?? 0)) || Date.parse(run.updated_at ?? '') || Date.now();
153 const source = { runId: `${run.id}/${attempt}`, revision: run.head_sha, completedAt };
154 if (!/^[a-f0-9]{40}$/.test(source.revision)) continue;
155 const outcomes = results.flatMap(r => r.outcomes);
156 for (const panel of panelReports(manifest, results, attempt).filter(p => p.status === 'FAIL')) {
157 const keys = outcomes.filter(o => o.trial?.case === panel.case && shardFile(o.files[0] ?? '') === panel.file).map(o => o.reused ? null : o.inputKey ?? null);
158 if (keys.length !== panel.panel.n || keys.some(k => k === null) || new Set(keys).size !== 1) continue;
159 writeNegativeReceipt(out, { schema: 1, key: keys[0]!, source });
160 written++;
161 }
162 for (const outcome of outcomes.filter(o => !o.trial && o.inputKey && !o.reused && o.status !== 'passed' && o.status !== 'skipped-by-diff')) {
163 writeNegativeReceipt(out, { schema: 1, key: outcome.inputKey!, source });
164 written++;
165 }
166 }
167 return written;
168}
170export interface CollectOptions {
171 repo: string;
172 pr: number;
173 headRepo: string;
174 branch: string;
175 selfRunId: number;
176 store: string;
177 /** The restored cache key; recovery reads only runs newer than the run that saved it. */
178 storeKey?: string;
179 /** `dispatch`: read-only receipts of the PR's completed runs for a validation dispatch (ENG-8). */
180 mode?: 'pr' | 'dispatch';
181 budgetMs?: number;
182 pollMs?: number;
183 gh?: GhRunner;
184 now?: () => number;
185 sleep?: (ms: number) => void;
186}
188export function collectRecovery(options: CollectOptions): { checkpoint: RecoveryCheckpoint; lines: string[] } {
189 const gh = options.gh ?? runGh;
190 const now = options.now ?? Date.now;
191 const sleep = options.sleep ?? ((ms: number) => Bun.sleepSync(ms));
192 const mode = options.mode ?? 'pr';
193 const deadline = now() + (options.budgetMs ?? 60_000);
194 const left = () => deadline - now();
195 const previous = readCheckpoint(options.store);
196 const processed = new Set(previous?.processed ?? []);
197 const pending = new Set((previous?.unresolved ?? []).map(entry => entry.runId));
198 const lines: string[] = [];
199 const finish = (checkpoint: Omit<RecoveryCheckpoint, 'schema' | 'updatedAt' | 'mode'>) => {
200 const full: RecoveryCheckpoint = { schema: 1, updatedAt: now(), mode, ...checkpoint, processed: checkpoint.processed.slice(-PROCESSED_KEEP) };
201 writeCheckpoint(options.store, full);
202 return { checkpoint: full, lines };
203 };
204 const degrade = (reason: string) => {
205 lines.push(`receipt recovery degraded: ${reason}; no recovered receipts, this run executes unreduced`);
206 return finish({ processed: [...processed], unresolved: previous?.unresolved ?? [], degraded: reason, recovered: 0 });
207 };
209 const listed = gh(['api', '-X', 'GET', `repos/${options.repo}/actions/workflows/evals.yml/runs`,
210 '-f', 'event=pull_request', '-f', `branch=${options.branch}`, '-f', 'per_page=50'], left());
211 if (listed.status !== 0) return degrade(ghFailureReason(listed.stderr));
212 let runs: WorkflowRun[];
213 try { runs = (JSON.parse(listed.stdout).workflow_runs ?? []) as WorkflowRun[]; } catch { return degrade('gh returned an unreadable run list'); }
215 const floor = mode === 'pr' ? storeRunId(options.storeKey ?? '') : 0;
216 const scope = { pr: options.pr, headRepo: options.headRepo, selfRunId: options.selfRunId };
217 for (const rejected of runs.filter(run => run.id !== options.selfRunId && run.event === 'pull_request' && !acceptedRun(run, scope)
218 && (run.pull_requests ?? []).some(pr => pr.number === options.pr))) {
219 lines.push(`rejected run ${rejected.id}: head repository ${rejected.head_repository?.full_name ?? 'unknown'} is not ${options.headRepo}`);
220 }
221 const older = runs.filter(run => acceptedRun(run, scope) && run.id < options.selfRunId);
222 // A completed run that was not cancelled reached its report, which saved its receipts.
223 const reported = (run: WorkflowRun) => run.status === 'completed' && run.conclusion !== 'cancelled';
224 const fresh = (run: WorkflowRun) => pending.has(String(run.id)) || (run.id > floor && !processed.has(String(run.id)));
225 const candidates = (mode === 'dispatch' ? older.filter(run => run.status === 'completed')
226 : older.filter(run => fresh(run) && !reported(run))).sort((a, b) => b.id - a.id).slice(0, MAX_RECOVERED_RUNS);
227 if (mode === 'pr') for (const run of older.filter(run => fresh(run) && reported(run))) processed.add(String(run.id));
229 const unresolved: RecoveryCheckpoint['unresolved'] = [];
230 const work = fs.mkdtempSync(path.join(os.tmpdir(), 'recover-receipts-'));
231 let recovered = 0;
232 try {
233 // Oldest first, so a newer FAIL always lands after the PASS it blocks.
234 for (const listedRun of [...candidates].sort((a, b) => a.id - b.id)) {
235 const id = String(listedRun.id);
236 let run = listedRun;
237 while (run.status !== 'completed' && left() > (options.pollMs ?? 5_000)) {
238 sleep(options.pollMs ?? 5_000);
239 const polled = gh(['api', `repos/${options.repo}/actions/runs/${id}`], left());
240 if (polled.status !== 0) break;
241 try { run = JSON.parse(polled.stdout) as WorkflowRun; } catch { break; }
242 }
243 if (run.status !== 'completed') {
244 unresolved.push({ runId: id, reason: 'still in progress; its completed FAILs cannot be read yet' });
245 lines.push(`run ${id} is still ${run.status}; reuse stays off until a later run reconciles it`);
246 continue;
247 }
248 if (mode === 'pr' && run.conclusion !== 'cancelled') { processed.add(id); continue; }
249 const dir = path.join(work, id);
250 const names = gh(['api', '-X', 'GET', `repos/${options.repo}/actions/runs/${id}/artifacts`, '-f', 'per_page=100'], left());
251 let artifacts: string[] = [];
252 try {
253 artifacts = names.status === 0 ? (JSON.parse(names.stdout).artifacts ?? [])
254 .filter((a: { name: string; expired?: boolean }) => !a.expired && /^(?:paid-plan|paid-slice-\d+-a\d+|report-verdict-a\d+)$/.test(a.name))
255 .map((a: { name: string }) => a.name) : [];
256 } catch { artifacts = []; }
257 if (names.status !== 0) {
258 unresolved.push({ runId: id, reason: ghFailureReason(names.stderr) });
259 lines.push(`run ${id}: cannot list artifacts (${ghFailureReason(names.stderr)})`);
260 continue;
261 }
262 let failed = '';
263 if (artifacts.length && left() <= 0) failed = 'the 60 s download budget ran out';
264 else if (artifacts.length) {
265 // gh extracts one artifact into -D itself and several into one directory per artifact.
266 const target = artifacts.length === 1 ? path.join(dir, artifacts[0]!) : dir;
267 const got = gh(['run', 'download', id, '-R', options.repo, ...artifacts.flatMap(name => ['-n', name]), '-D', target], left());
268 if (got.status !== 0) failed = ghFailureReason(got.stderr);
269 }
270 if (failed) {
271 unresolved.push({ runId: id, reason: failed });
272 lines.push(`run ${id}: ${failed}; its evidence stays unresolved`);
273 continue;
274 }
275 const negatives = path.join(dir, 'recovered-negatives');
276 const panelNegatives = negativesFromRun(dir, run, negatives);
277 const receiptDirs = [negatives, ...artifacts.filter(name => name !== 'paid-plan').flatMap(name =>
278 [path.join(dir, name, 'receipts'), path.join(dir, name, 'paid-report', 'report-receipts')]),
279 ...(mode === 'dispatch' ? [path.join(dir, 'paid-plan', 'store'), path.join(dir, 'paid-plan', 'receipts')] : [])];
280 const merged = mergeReceiptDirs(options.store, receiptDirs);
281 recovered += merged;
282 processed.add(id);
283 lines.push(`run ${id} (${run.conclusion}): merged ${merged} receipt(s), ${panelNegatives} reconstructed negative(s) from completed panels and shards`);
284 }
285 } finally {
286 fs.rmSync(work, { recursive: true, force: true });
287 }
288 if (!candidates.length) lines.push('no cancelled runs to recover');
289 return finish({ processed: [...processed], unresolved, recovered });
290}
292/** plan-slices: reuse is on only when recovery ran, did not degrade and left nothing unresolved. */
293export function recoveryDecision(store: string): { reuse: boolean; reason: string } {
294 const checkpoint = readCheckpoint(store);
295 if (!checkpoint) return { reuse: false, reason: 'receipt recovery unavailable (no checkpoint from the recover-receipts job); cases execute fresh' };
296 if (checkpoint.degraded) return { reuse: false, reason: `receipt recovery degraded: ${checkpoint.degraded}; cases execute fresh` };
297 if (checkpoint.unresolved.length) {
298 return { reuse: false, reason: `cancelled run(s) ${checkpoint.unresolved.map(u => u.runId).join(', ')} not reconciled (${checkpoint.unresolved[0]!.reason}); cases execute fresh and the next push reconciles them` };
299 }
300 return { reuse: true, reason: `receipt reuse on; ${checkpoint.recovered} receipt(s) recovered from cancelled runs` };
301}
303const USAGE = `usage:
304 bun run scripts/recover-receipts.ts collect --repo <owner/name> --pr <n> --head-repo <owner/name> --branch <head branch>
305 --run-id <this run> --store <dir> [--store-key <restored cache key>] [--mode pr|dispatch] [--budget-seconds 60]
306 bun run scripts/recover-receipts.ts decide <store> [--github-output <file>]
307Recovery never fails: on any gh error it records why and the run executes unreduced.`;
309if (import.meta.main) {
310 const [command, ...rest] = process.argv.slice(2);
311 const flag = (name: string) => { const i = rest.indexOf(`--${name}`); return i >= 0 ? rest[i + 1] : undefined; };
312 if (!command || command === '--help' || command === '-h') { console.log(USAGE); process.exit(command ? 0 : 2); }
313 if (command === 'decide' && rest[0]) {
314 const decision = recoveryDecision(rest[0]);
315 const out = flag('github-output');
316 if (out) fs.appendFileSync(out, `reuse=${decision.reuse ? 'on' : 'off'}\n`);
317 console.log(`- Receipt reuse: ${decision.reason}`);
318 } else if (command === 'collect') {
319 const [store, repo, headRepo, branch] = [flag('store'), flag('repo'), flag('head-repo'), flag('branch')];
320 const [pr, selfRunId] = [Number(flag('pr')), Number(flag('run-id'))];
321 if (!store || !repo || !headRepo || !branch || !Number.isSafeInteger(pr) || !Number.isSafeInteger(selfRunId)) {
322 console.error(USAGE); process.exit(2);
323 }
324 const { lines } = collectRecovery({ repo, pr, headRepo, branch, selfRunId, store, storeKey: flag('store-key'),
325 mode: flag('mode') === 'dispatch' ? 'dispatch' : 'pr', budgetMs: Number(flag('budget-seconds') ?? '60') * 1000 });
326 console.log('### Receipt recovery');
327 for (const line of lines) console.log(`- ${line}`);
328 } else { console.error(USAGE); process.exit(2); }
329}