src/bench.ts

Functions

Imports

It imports src/agents-schema.ts, src/agents.ts, src/auth.ts, src/bench-requests.ts, src/bench-run-key.ts, src/encoding.ts, src/env.ts, src/http.ts, src/job-breaker.ts, src/jobs-seat.ts, src/jobs-transition.ts, src/log.ts, src/signed-requests.ts, src/store-guards.ts. Imported by src/routes.ts.

Source

1import { bytesToHex } from "./encoding";
2import type { Env } from "./env";
3import { claimJti, MAX_REPORT_BYTES, readBoundedText, verifyBenchRequest } from "./signed-requests";
4import { breakerRefusal, breakerState } from "./job-breaker";
5import { logEvent } from "./log";
6import { BENCH_NAMESPACE, BENCH_REPRO_ACTION, BENCH_STALE_MS, newRunKey, readReproReceipt, runJobTitle } from "./bench-run-key";
7import { sha256Hex } from "./auth";
8import type { Agent } from "./agents";
9import { noFlags } from "./agents-schema";
10import { adminFailJob, supersedeJob, workRecorded } from "./jobs-seat";
11import { readJob } from "./jobs-transition";
12import { auditStatement, isMissingRowAbort, requireBenchRunRunning } from "./store-guards";
13import { textResponse } from "./http";
14import { type Arm, type Refusal, JTI_SCOPE, parseTaskRequest, parseDrawRequest, type BenchRegistration, parseStartRequest, parseReproRequest, parseScoreRequest, type TreatmentJob, runOutcome } from "./bench-requests";
15
16// THE METHODS PAPER'S PAIRED COMPARISON, the Worker's half (job_ecb1b5f80c2d;
17// capsid/research/design-paired-comparison.md, section 2; migrations/0047_bench.sql).
18//
19// The capsid-bench repo runs one task per GitHub Actions job per arm and talks to Capsid
20// through four endpoints, signed like the improve scorer's (its own key, the Worker
21// secret BENCH_SECRET, the same timestamp window, the same replay cache under its own
22// scope) and not under /ops/ for the scorer's reason:
23//
24//   /bench/tasks  register a task and its hidden test counts, once
25//   /bench/draw   record the registration and append the drawn task order
26//   /bench/start  open a run, behind the off switch and the concurrency cap
27//   /bench/repro  the harness's receipt for a treatment run's reproduction test
28//   /bench/score  close it: the Worker computes resolved and the false success claim
29//
30// ONE CLAIM RULE FOR EVERY ARM. Each arm's claim is the claim file its run wrote with
31// the bench's claim tool, read by capsid-bench's one parser and reported as claimed_done
32// (the methods paper's protocol, section 3). The treatment's Capsid job is still checked
33// to be this run's, but its status is not the claim.
34
35export const BENCH_TASKS_PATH = "/bench/tasks";
36export const BENCH_DRAW_PATH = "/bench/draw";
37export const BENCH_START_PATH = "/bench/start";
38export const BENCH_SCORE_PATH = "/bench/score";
39export const BENCH_REPRO_PATH = "/bench/repro";
40
41// The off switch: off unless it reads exactly "on". Absent, any other value or a read
42// that fails is off, so a run never starts by default or by outage.
43export const BENCH_ENABLED_KEY = "bench:enabled";
44// At most this many runs at once, so a pilot never starves Dustin's own subscription use.
45export const BENCH_MAX_RUNNING = 4;
46// A running row older than BENCH_STALE_MS (src/bench-run-key.ts) no longer holds a
47// slot: its job hit GitHub's own timeout without reporting. The row stays 'running' as
48// the record of that.
49// The design's k=3 repeats (D17): each model arm gets this many slots per task. A slot
50// may be started twice, and the second start only when the first came back unjudged (a
51// machine failure); a scored or running attempt holds its slot. Enforced here, not in the
52// hand-dispatched workflow, so a run whose outcome was seen cannot be replaced by another
53// start. A running row past BENCH_STALE_MS counts as unjudged.
54export const BENCH_REPEATS = 3;
55// THE ERROR STOP AND THE OUTAGE HOUR (the methods paper's protocol, section 3, revised
56// 2026-10-22). A model run that ends with Claude Code's "error during execution" (a usage
57// limit and a transport error look the same, and neither is classified) is scored from
58// its diff and claim file the first time, as a clock kill is, and fills its slot. Except
59// in an outage hour: a UTC hour in which at least BENCH_OUTAGE_MIN_ENDED model runs ended
60// and at least half of them stopped this way. Such a stop counts as no attempt at all, so
61// its slot is started once more, and no model run starts while the current hour already
62// meets the rule.
63//
64// THE ORDER (section 5, "Order", decided 2026-10-22). A drawn task's six slots run in the
65// fixed order baseline, treatment, baseline, treatment, baseline, treatment, each ended
66// before the next starts, and a drawn task's first slot starts only when every task
67// before it on the draw has ended all six.
68const BENCH_OUTAGE_MIN_ENDED = 4;
69const MODEL_ARMS: readonly Arm[] = ["baseline", "treatment"];
70// The shared front of all three endpoints: bounded body, signature, then the parse,
71// then the jti. The jti is spent only once the body is known good, as in the improve
72// scorer's credential mint.
73async function signedBody<T extends { jti: string }>(
74  request: Request,
75  env: Env,
76  event: string,
77  parse: (body: string) => ({ ok: true } & T) | Refusal
78): Promise<{ ok: true; value: T } | { ok: false; response: Response }> {
79  const bounded = await readBoundedText(request, MAX_REPORT_BYTES);
80  if (!bounded.ok) return { ok: false, response: textResponse(`request too large: exceeds ${MAX_REPORT_BYTES} bytes`, 413) };
81  const verdict = await verifyBenchRequest(
82    env,
83    {
84      timestamp: request.headers.get("X-Bench-Timestamp") ?? "",
85      signature: request.headers.get("X-Bench-Signature") ?? "",
86      body: bounded.text,
87    },
88    new Date()
89  );
90  if (!verdict.ok) {
91    logEvent("error", event, { message: `${event} ${verdict.refusal}` });
92    return { ok: false, response: textResponse(verdict.refusal, verdict.status) };
93  }
94  const parsed = parse(bounded.text);
95  if (!parsed.ok) return { ok: false, response: textResponse(parsed.refusal, 400) };
96  const claim = await claimJti(env.DB, JTI_SCOPE, parsed.jti);
97  if (!claim.ok) return { ok: false, response: textResponse(claim.refusal, claim.status) };
98  return { ok: true, value: parsed };
99}
100
101const TASK_FIELDS = ["source", "split", "repo", "issue_created_at", "fail_to_pass_count", "pass_to_pass_count"] as const;
102
103export async function handleBenchTasks(request: Request, env: Env): Promise<Response> {
104  const front = await signedBody(request, env, "BENCH_TASKS_REJECTED", parseTaskRequest);
105  if (!front.ok) return front.response;
106  const t = front.value.task;
107  const inserted = await env.DB.prepare(
108    `INSERT INTO bench_tasks (task_id, source, split, repo, issue_created_at, fail_to_pass_count, pass_to_pass_count, registered_at)
109     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
110     ON CONFLICT(task_id) DO NOTHING
111     RETURNING task_id`
112  )
113    .bind(t.task_id, t.source, t.split, t.repo, t.issue_created_at, t.fail_to_pass_count, t.pass_to_pass_count, new Date().toISOString())
114    .all<{ task_id: string }>();
115  if (inserted.results.length === 1) return Response.json({ ok: true, task_id: t.task_id, registered: true });
116  // Registered before. The same registration again is a no-op, so the loader can be
117  // rerun; a different one is refused, because the registered list is the one scored.
118  const existing = await env.DB.prepare(
119    "SELECT source, split, repo, issue_created_at, fail_to_pass_count, pass_to_pass_count FROM bench_tasks WHERE task_id = ?1"
120  )
121    .bind(t.task_id)
122    .first<Record<(typeof TASK_FIELDS)[number], string | number>>();
123  const differs = existing ? TASK_FIELDS.filter((f) => existing[f] !== t[f]) : [];
124  if (differs.length === 0) return Response.json({ ok: true, task_id: t.task_id, registered: false });
125  return textResponse(`task ${t.task_id} is already registered with a different ${differs.join(", ")}; a registered task never changes`, 409);
126}
127
128// THE DRAW (the methods paper's protocol, section 4). The first call records the
129// registration; every call must carry that same registration. tasks are the draw from
130// position `from` on: positions already recorded must hold the same tasks, and the rest
131// are appended in order, so a batch can be sent again and the draw only grows. A drawn
132// task must be registered in the full split. One batch, so a refused task appends none.
133export async function handleBenchDraw(request: Request, env: Env): Promise<Response> {
134  const front = await signedBody(request, env, "BENCH_DRAW_REJECTED", parseDrawRequest);
135  if (!front.ok) return front.response;
136  const { registration, from, tasks } = front.value;
137  const now = new Date().toISOString();
138  await env.DB.prepare(
139    `INSERT INTO bench_registration (id, registration_sha256, frame_sha256, seed, recorded_at) VALUES (1, ?1, ?2, ?3, ?4)
140     ON CONFLICT(id) DO NOTHING`
141  )
142    .bind(registration.registration_sha256, registration.frame_sha256, registration.seed, now)
143    .run();
144  const recorded = await env.DB.prepare("SELECT registration_sha256, frame_sha256, seed FROM bench_registration WHERE id = 1").first<BenchRegistration>();
145  const differs = recorded ? (["registration_sha256", "frame_sha256", "seed"] as const).filter((f) => recorded[f] !== registration[f]) : [];
146  if (!recorded || differs.length > 0) {
147    return textResponse(`a different registration is recorded (${differs.join(", ")}); a re-plan is a new registration, recorded by the seat before any confirmatory run`, 409);
148  }
149  const { results: drawn } = await env.DB.prepare("SELECT position, task_id FROM bench_draw WHERE position >= ?1 ORDER BY position")
150    .bind(from)
151    .all<{ position: number; task_id: string }>();
152  const size = await env.DB.prepare("SELECT COUNT(*) AS n FROM bench_draw").first<{ n: number }>();
153  const count = size?.n ?? 0;
154  if (from > count + 1) return textResponse(`the draw holds ${count} tasks, so a batch starts at position ${count + 1} at the latest, not ${from}`, 409);
155  for (const row of drawn) {
156    const sent = tasks[row.position - from];
157    if (sent !== undefined && sent !== row.task_id) return textResponse(`draw position ${row.position} holds ${row.task_id}, not ${sent}; the draw never changes`, 409);
158  }
159  const fresh = tasks.slice(count + 1 - from);
160  if (fresh.length > 0) {
161    const placeholders = fresh.map((_, i) => `?${i + 1}`).join(", ");
162    const { results: full } = await env.DB.prepare(`SELECT task_id FROM bench_tasks WHERE split = 'full' AND task_id IN (${placeholders})`)
163      .bind(...fresh)
164      .all<{ task_id: string }>();
165    const known = new Set(full.map((t) => t.task_id));
166    const missing = fresh.filter((t) => !known.has(t));
167    if (missing.length > 0) return textResponse(`not registered in the full split, so not drawn: ${missing.slice(0, 5).join(", ")}${missing.length > 5 ? ", ..." : ""}`, 409);
168    try {
169      await env.DB.batch(fresh.map((task_id, i) => env.DB.prepare("INSERT INTO bench_draw (position, task_id, recorded_at) VALUES (?1, ?2, ?3)").bind(count + 1 + i, task_id, now)));
170    } catch (err) {
171      return textResponse(`the draw was not extended (${err instanceof Error ? err.message : String(err)}); a task already drawn, or another batch landed at the same time`, 409);
172    }
173  }
174  logEvent("log", "BENCH_DRAW_RECORDED", { message: `BENCH_DRAW_RECORDED added=${fresh.length} size=${count + fresh.length}` });
175  return Response.json({ ok: true, added: fresh.length, size: count + fresh.length });
176}
177
178// THE REGISTRATION LOCK, for a model arm. A full-split task starts only once the
179// registration is recorded and the task is on the draw. A pilot or excluded task starts
180// only before it: the pilot checks feasibility and may change the budgets, and excluded
181// (pre-cutoff) tasks are where the template is tuned (protocol section 2), so both
182// precede the hash. Neither is sealed or analysed.
183// The planted arms call no model and are not locked.
184async function drawRefusal(db: D1Database, taskId: string, split: string): Promise<string | null> {
185  const registered = await db.prepare("SELECT 1 AS one FROM bench_registration WHERE id = 1").first<{ one: number }>();
186  if (split === "pilot" || split === "excluded") return registered ? `task ${taskId} is a ${split} task, and ${split} runs end when the registration is recorded` : null;
187  if (!registered) return `no registration is recorded; record it and the draw with ${BENCH_DRAW_PATH} before a model run on a full-split task`;
188  const drawn = await db.prepare("SELECT 1 AS one FROM bench_draw WHERE task_id = ?1").bind(taskId).first<{ one: number }>();
189  return drawn ? null : `task ${taskId} is not on the draw`;
190}
191
192export interface PriorRun {
193  status: string;
194  started_at: string;
195  finished_at?: string | null;
196  stopped_on_error?: number | boolean | null;
197}
198
199/** Whether a UTC hour with this many ended model runs, this many of them stopped on an error, is an outage hour. */
200export function isOutageHour(ended: number, errors: number): boolean {
201  return ended >= BENCH_OUTAGE_MIN_ENDED && errors * 2 >= ended;
202}
203
204// The outage hours as of `at`, from the runs that had ended by then, as the first 13
205// characters of an ISO time ("2026-10-19T02").
206async function outageHours(db: D1Database, at: string): Promise<Set<string>> {
207  const { results } = await db
208    .prepare(
209      `SELECT substr(finished_at, 1, 13) AS hour, COUNT(*) AS ended,
210         SUM(CASE WHEN json_extract(report, '$.stopped_on_error') = 1 THEN 1 ELSE 0 END) AS errors
211       FROM bench_runs WHERE arm IN ('baseline', 'treatment') AND finished_at IS NOT NULL AND finished_at <= ?1 GROUP BY hour`
212    )
213    .bind(at)
214    .all<{ hour: string; ended: number; errors: number }>();
215  return new Set(results.filter((h) => isOutageHour(h.ended, h.errors)).map((h) => h.hour));
216}
217
218/** The runs that count toward slots: an error stop in an outage hour is no attempt at all. */
219export function countedRuns<T extends PriorRun>(runs: readonly T[], outage: ReadonlySet<string>): T[] {
220  return runs.filter((r) => !(r.stopped_on_error && r.finished_at && outage.has(r.finished_at.slice(0, 13))));
221}
222
223// A task and arm's runs in start order, for the start's slot and the order rule.
224async function priorRuns(db: D1Database, taskId: string, arm: string): Promise<(PriorRun & { id: string })[]> {
225  const { results } = await db
226    .prepare(
227      `SELECT id, status, started_at, finished_at, json_extract(report, '$.stopped_on_error') AS stopped_on_error
228       FROM bench_runs WHERE task_id = ?1 AND arm = ?2 ORDER BY started_at, rowid`
229    )
230    .bind(taskId, arm)
231    .all<PriorRun & { id: string }>();
232  return results;
233}
234
235// Which slot and attempt the next start of a model arm on a task takes, from that task
236// and arm's earlier runs in start order, or null when every slot is used. Slots fill in
237// order, so a slot is identified by position: an unjudged first attempt leaves its slot
238// waiting for its one rerun, and the next start fills that slot before opening another.
239export function nextSlot(runs: readonly PriorRun[], staleBefore: string, repeats = BENCH_REPEATS): { slot: number; attempt: 1 | 2 } | null {
240  let slot = 0;
241  let waiting = false;
242  for (const run of runs) {
243    const unjudged = run.status === "unjudged" || (run.status === "running" && run.started_at <= staleBefore);
244    if (waiting) {
245      waiting = false;
246      continue;
247    }
248    slot++;
249    waiting = unjudged;
250  }
251  if (waiting) return { slot, attempt: 2 };
252  return slot < repeats ? { slot: slot + 1, attempt: 1 } : null;
253}
254
255/** The slot and attempt a model start at `at` takes, from the task and arm's runs started
256 *  before it, each read as it stood at `at`: a run that ended later reads as running. Pure. */
257export function slotAt(runs: readonly PriorRun[], outage: ReadonlySet<string>, at: string): { slot: number; attempt: 1 | 2 } | null {
258  const stood = runs.map((r) => (r.finished_at && r.finished_at <= at ? r : { ...r, status: "running", finished_at: null, stopped_on_error: null }));
259  return nextSlot(countedRuns(stood, outage), new Date(Date.parse(at) - BENCH_STALE_MS).toISOString());
260}
261
262/** How many of a task and arm's slots have ended: scored, or unjudged on both attempts.
263 *  A running attempt, or an unjudged one still owed its rerun, has not. Pure. */
264export function slotsDone(runs: readonly PriorRun[], staleBefore: string): number {
265  let done = 0;
266  let waiting = false;
267  for (const run of runs) {
268    const stale = run.status === "running" && run.started_at <= staleBefore;
269    const running = run.status === "running" && !stale;
270    if (waiting) {
271      waiting = false;
272      if (!running) done++;
273      continue;
274    }
275    if (run.status === "unjudged" || stale) waiting = true;
276    else if (!running) done++;
277  }
278  return done;
279}
280
281/** Why a model start on a drawn task is out of order, or null. `slot` is the slot the
282 *  start takes; `done` counts each arm's ended slots on this task; `earlierOpen` is the
283 *  first earlier drawn task that has not ended all six, if any. Pure. */
284export function orderRefusal(
285  arm: Arm,
286  slot: number,
287  done: Record<"baseline" | "treatment", number>,
288  running: Record<"baseline" | "treatment", boolean>,
289  earlierOpen: string | null
290): string | null {
291  if (arm === "baseline" && slot === 1 && earlierOpen) return `the draw runs in order: ${earlierOpen} has not ended its six slots`;
292  const before = arm === "baseline" ? { arm: "treatment" as const, need: slot - 1 } : { arm: "baseline" as const, need: slot };
293  if (running[before.arm] || done[before.arm] !== before.need) {
294    return `the slots run baseline, treatment, baseline, treatment, baseline, treatment: ${arm} slot ${slot} starts when ${before.arm} has ended ${before.need} and none is running (it has ended ${done[before.arm]})`;
295  }
296  return null;
297}
298
299// The order rule's facts for a start, read from the database. Tasks off the draw (the
300// pilot) have no order.
301async function orderOf(db: D1Database, taskId: string, arm: Arm, slot: number, outage: ReadonlySet<string>, stale: string): Promise<string | null> {
302  const row = await db.prepare("SELECT position FROM bench_draw WHERE task_id = ?1").bind(taskId).first<{ position: number }>();
303  if (!row) return null;
304  const state = async (id: string) => {
305    const done = { baseline: 0, treatment: 0 };
306    const running = { baseline: false, treatment: false };
307    for (const a of MODEL_ARMS) {
308      const runs = countedRuns(await priorRuns(db, id, a), outage);
309      done[a as "baseline" | "treatment"] = slotsDone(runs, stale);
310      running[a as "baseline" | "treatment"] = runs.some((r) => r.status === "running" && r.started_at > stale);
311    }
312    return { done, running };
313  };
314  let earlierOpen: string | null = null;
315  if (arm === "baseline" && slot === 1) {
316    const { results: earlier } = await db.prepare("SELECT task_id FROM bench_draw WHERE position < ?1 ORDER BY position").bind(row.position).all<{ task_id: string }>();
317    for (const { task_id: id } of earlier) {
318      const s = await state(id);
319      if (s.done.baseline < BENCH_REPEATS || s.done.treatment < BENCH_REPEATS) {
320        earlierOpen = id;
321        break;
322      }
323    }
324  }
325  const own = await state(taskId);
326  return orderRefusal(arm, slot, own.done, own.running, earlierOpen);
327}
328
329export async function handleBenchStart(request: Request, env: Env): Promise<Response> {
330  const front = await signedBody(request, env, "BENCH_START_REJECTED", parseStartRequest);
331  if (!front.ok) return front.response;
332  const { task_id, arm, gh_run } = front.value;
333  let enabled: string | null;
334  try {
335    enabled = await env.APP_KV.get(BENCH_ENABLED_KEY);
336  } catch (err) {
337    return textResponse(`could not read ${BENCH_ENABLED_KEY} (${err instanceof Error ? err.message : String(err)}); a bench run starts only when it reads "on"`, 503);
338  }
339  if (enabled !== "on") return textResponse(`bench runs are off: ${BENCH_ENABLED_KEY} reads ${JSON.stringify(enabled)}, not "on"`, 403);
340  const task = await env.DB.prepare("SELECT split FROM bench_tasks WHERE task_id = ?1").bind(task_id).first<{ split: string }>();
341  if (!task) return textResponse(`no task ${task_id} is registered; register it with ${BENCH_TASKS_PATH} first`, 404);
342  const now = new Date();
343  if (arm === "treatment") {
344    const breaker = await breakerState(env, BENCH_NAMESPACE, now);
345    if (breaker.open) return textResponse(`no treatment run starts while the bench job breaker is open: ${breakerRefusal(breaker)}`, 409);
346  }
347  const id = `bench_${bytesToHex(crypto.getRandomValues(new Uint8Array(6)))}`;
348  const stale = new Date(now.getTime() - BENCH_STALE_MS).toISOString();
349  let slot: { slot: number; attempt: 1 | 2 } | null = null;
350  let prior = -1;
351  if (MODEL_ARMS.includes(arm)) {
352    const locked = await drawRefusal(env.DB, task_id, task.split);
353    if (locked) return textResponse(locked, 409);
354    // A stale running row is closed as unjudged before its rerun can open, so its job can
355    // no longer post a late score: /bench/score refuses a row that is not running.
356    // A treatment row's job is closed with it (finishRun); one that moved in between is
357    // left running, and the slot refusal below says so.
358    const { results: staleRuns } = await env.DB.prepare(
359      "SELECT id FROM bench_runs WHERE task_id = ?1 AND arm = ?2 AND status = 'running' AND started_at <= ?3"
360    )
361      .bind(task_id, arm, stale)
362      .all<{ id: string }>();
363    for (const { id: staleId } of staleRuns ?? []) {
364      await finishRun(env, { id: staleId, task_id, arm }, now, [
365        env.DB.prepare(
366          `UPDATE bench_runs SET status = 'unjudged', finished_at = ?2, unjudged_reason = 'no report within the stale window; closed when the next start was asked for'
367           WHERE id = ?1 AND status = 'running'`
368        ).bind(staleId, now.toISOString()),
369      ]);
370    }
371    const outage = await outageHours(env.DB, now.toISOString());
372    if (outage.has(now.toISOString().slice(0, 13))) {
373      return textResponse(`no model run starts in an outage hour: at least half of this hour's ended runs stopped with an error during execution; start again next hour`, 409);
374    }
375    const results = await priorRuns(env.DB, task_id, arm);
376    slot = slotAt(results, outage, now.toISOString());
377    if (!slot) {
378      return textResponse(
379        `task ${task_id} has used its ${arm} runs: ${BENCH_REPEATS} slots, each started at most twice and the second time only after an unjudged first attempt`,
380        409
381      );
382    }
383    const order = await orderOf(env.DB, task_id, arm, slot.slot, outage, stale);
384    if (order) return textResponse(order, 409);
385    prior = results.length;
386  }
387  // A treatment run's own key (src/bench-run-key.ts), returned once and stored only as
388  // its hash on the row it opens.
389  const runKey = arm === "treatment" ? newRunKey() : null;
390  const runKeyHash = runKey ? await sha256Hex(runKey) : null;
391  // One statement, so two starts racing for the last running slot cannot both take it,
392  // and a model start counted against the rows read above fails if another start landed
393  // since. The planted arms call no model and have no per-task slots.
394  const opened = await env.DB.prepare(
395    `INSERT INTO bench_runs (id, task_id, arm, gh_run, status, started_at, run_key_hash)
396     SELECT ?1, ?2, ?3, ?4, 'running', ?5, ?9
397     WHERE (SELECT COUNT(*) FROM bench_runs WHERE status = 'running' AND started_at > ?6) < ?7
398       AND (?8 < 0 OR (SELECT COUNT(*) FROM bench_runs WHERE task_id = ?2 AND arm = ?3) = ?8)
399     RETURNING id`
400  )
401    .bind(id, task_id, arm, gh_run, now.toISOString(), stale, BENCH_MAX_RUNNING, prior, runKeyHash)
402    .all<{ id: string }>();
403  if (opened.results.length !== 1) {
404    if (prior >= 0) {
405      const now2 = await env.DB.prepare("SELECT COUNT(*) AS n FROM bench_runs WHERE task_id = ?1 AND arm = ?2").bind(task_id, arm).first<{ n: number }>();
406      if ((now2?.n ?? prior) !== prior) return textResponse(`another ${arm} start for task ${task_id} landed at the same time; start again`, 409);
407    }
408    return textResponse(`${BENCH_MAX_RUNNING} bench runs are already running; try again when one finishes`, 429);
409  }
410  const at = slot ? ` slot=${slot.slot} attempt=${slot.attempt}` : "";
411  logEvent("log", "BENCH_RUN_STARTED", { message: `BENCH_RUN_STARTED ${id} task=${task_id} arm=${arm} gh_run=${gh_run}${at}` });
412  return Response.json({ ok: true, run_id: id, ...(slot ? { slot: slot.slot, attempt: slot.attempt } : {}), ...(runKey ? { run_key: runKey } : {}) });
413}
414
415// THE JOB CLOSE (the methods paper's protocol, section 4, "The job"; decided 2026-10-19).
416// The score step finishes the run and closes that run's job in one batch, so the run is
417// not finished unless the close commits with it, and the run key stops resolving in the
418// same step (src/bench-run-key.ts reads a running row only). The close is a supersede by
419// the harness with the same reason on every treatment run; a job with work recorded on
420// it, which supersede refuses, is failed by the seat path with that reason, no claim and
421// no outcome row, since it is not the holder's fail. Neither the run key nor the driver
422// key that posted the job can supersede it. A stale run is closed by the same rule when
423// the next start marks it unjudged. The analysis reads the claim file, never job status.
424export const BENCH_CLOSE_REASON = "closed by the harness; the claim is the claim file";
425const BENCH_HARNESS: Agent = {
426  id: "bench-harness",
427  name: "bench-harness",
428  kind: "seat",
429  actor: "bench-harness",
430  scopes: { namespaces: [BENCH_NAMESPACE], repos: [], tools: ["capsid_work_jobs"], grants: ["read", "write"], flags: { ...noFlags(), can_merge: true } },
431  admin: false,
432  row: null,
433};
434
435/** Commits `rowClose` only while the run is running, with its job's close for a treatment run. False when nothing was written. */
436async function finishRun(env: Env, run: { id: string; task_id: string; arm: Arm }, now: Date, rowClose: D1PreparedStatement[]): Promise<boolean> {
437  const statements = [requireBenchRunRunning(env.DB, run.id), ...rowClose];
438  if (run.arm === "treatment") {
439    const open = await env.DB.prepare("SELECT id FROM jobs WHERE namespace = ?1 AND title = ?2 AND status IN ('queued', 'claimed', 'blocked')")
440      .bind(BENCH_NAMESPACE, runJobTitle(run.task_id, run.id))
441      .first<{ id: string }>();
442    const job = open ? await readJob(env.DB, open.id) : null;
443    if (job) {
444      const failed = job.status === "blocked" || (job.status === "claimed" && workRecorded(job) !== null);
445      const result = failed
446        ? await adminFailJob(env, BENCH_HARNESS, now, job.id, BENCH_CLOSE_REASON, { extra: statements, outcome: false })
447        : await supersedeJob(env, BENCH_HARNESS, now, job.id, { reason: BENCH_CLOSE_REASON }, statements);
448      return result.ok;
449    }
450  }
451  try {
452    await env.DB.batch(statements);
453    return true;
454  } catch (err) {
455    if (!isMissingRowAbort(err)) throw err;
456    return false;
457  }
458}
459
460// The receipt is written once, for a running treatment run, and never changes: the
461// harness checks once at the phase break (src/bench-run-key.ts, "THE REPRODUCTION RECEIPT").
462export async function handleBenchRepro(request: Request, env: Env): Promise<Response> {
463  const front = await signedBody(request, env, "BENCH_REPRO_REJECTED", parseReproRequest);
464  if (!front.ok) return front.response;
465  const { run_id, passed, reason } = front.value;
466  const run = await env.DB.prepare("SELECT arm, status FROM bench_runs WHERE id = ?1").bind(run_id).first<{ arm: Arm; status: string }>();
467  if (!run) return textResponse(`no bench run ${run_id}`, 404);
468  if (run.arm !== "treatment") return textResponse(`bench run ${run_id} is a ${run.arm} run; only a treatment run has a reproduction receipt`, 409);
469  if (run.status !== "running") return textResponse(`bench run ${run_id} is already closed (${run.status})`, 409);
470  if (await readReproReceipt(env.DB, run_id)) return textResponse(`bench run ${run_id} already has its reproduction receipt; it is written once`, 409);
471  await auditStatement(env.DB, "system:bench-harness", BENCH_REPRO_ACTION, BENCH_NAMESPACE, run_id, { passed, reason }).run();
472  return Response.json({ ok: true, run_id, passed });
473}
474
475export async function handleBenchScore(request: Request, env: Env): Promise<Response> {
476  const front = await signedBody(request, env, "BENCH_SCORE_REJECTED", parseScoreRequest);
477  if (!front.ok) return front.response;
478  const report = front.value.report;
479  const run = await env.DB.prepare(
480    `SELECT r.id, r.task_id, r.arm, r.status, t.split, t.fail_to_pass_count, t.pass_to_pass_count
481     FROM bench_runs r JOIN bench_tasks t ON t.task_id = r.task_id WHERE r.id = ?1`
482  )
483    .bind(report.run_id)
484    .first<{ id: string; task_id: string; arm: Arm; status: string; split: string; fail_to_pass_count: number; pass_to_pass_count: number }>();
485  if (!run) return textResponse(`no bench run ${report.run_id}`, 404);
486  if (run.status !== "running") return textResponse(`bench run ${run.id} is already closed (${run.status})`, 409);
487  if (report.diff !== null && run.split !== "excluded") return textResponse(`task ${run.task_id} is a ${run.split} task; a diff is kept only for an excluded (practice) task`, 409);
488  const job =
489    run.arm === "treatment" && report.capsid_job_id
490      ? await env.DB.prepare("SELECT namespace, title, status FROM jobs WHERE id = ?1").bind(report.capsid_job_id).first<TreatmentJob>()
491      : null;
492  const outcome = runOutcome(report, run, run, job);
493  const scored = outcome.status === "scored";
494  const flag = (v: boolean | null | undefined) => (v === undefined || v === null ? null : v ? 1 : 0);
495  const u = report.usage;
496  const finishedAt = new Date();
497  const closeRow = env.DB.prepare(
498    `UPDATE bench_runs SET status = ?2, finished_at = ?3, capsid_job_id = ?4, claimed_done = ?5, resolved = ?6, false_claim = ?7,
499       fail_to_pass_passed = ?8, pass_to_pass_passed = ?9, unjudged_reason = ?10, model = ?11, claude_code_version = ?12,
500       cost_usd = ?13, turns = ?14, duration_ms = ?15, input_tokens = ?16, output_tokens = ?17, cache_read_tokens = ?18,
501       cache_creation_tokens = ?19, image_bytes = ?20, report = ?21
502     WHERE id = ?1 AND status = 'running'`
503  ).bind(
504      run.id,
505      outcome.status,
506      finishedAt.toISOString(),
507      report.capsid_job_id,
508      flag(scored ? outcome.claimed_done : undefined),
509      flag(scored ? outcome.resolved : undefined),
510      flag(scored ? outcome.false_claim : undefined),
511      report.hidden?.fail_to_pass.passed ?? null,
512      report.hidden?.pass_to_pass.passed ?? null,
513      scored ? null : outcome.reason,
514      u.model,
515      u.claude_code_version,
516      u.cost_usd,
517      u.turns,
518      u.duration_ms,
519      u.input_tokens,
520      u.output_tokens,
521      u.cache_read_tokens,
522      u.cache_creation_tokens,
523      report.image_bytes,
524      JSON.stringify(report)
525    );
526  if (!(await finishRun(env, run, finishedAt, [closeRow]))) {
527    // Two reports for one run racing: the first closed it, the second matched nothing.
528    const after = await env.DB.prepare("SELECT status FROM bench_runs WHERE id = ?1").bind(run.id).first<{ status: string }>();
529    if (after?.status !== "running") return textResponse(`bench run ${run.id} was closed by another report`, 409);
530    return textResponse(`bench run ${run.id}'s job changed while it was being closed, so nothing was scored; send the report again`, 409);
531  }
532  logEvent("log", "BENCH_RUN_SCORED", { message: `BENCH_RUN_SCORED ${run.id} ${outcome.status}` });
533  return Response.json({ ok: true, run_id: run.id, ...outcome });
534}