src/jobs-transition.ts

Functions

Imports

It imports src/agent-record.ts, src/agents.ts, src/env.ts, src/job-outcomes.ts, src/job-overlaps.ts, src/jobs-claim.ts, src/jobs-edit.ts, src/jobs-mirror.ts, src/jobs-schema.ts, src/log.ts, src/store-guards.ts. Imported by src/bench.ts, src/canon.ts, src/controls-perform.ts, src/controls-plan.ts, src/jobs-begin.ts, src/jobs-claim.ts, src/jobs-edit.ts, src/jobs-holder.ts, src/jobs-park.ts, src/jobs-seat.ts, src/jobs.ts, src/merge-resume.ts, src/routine-fire.ts, src/shadow.ts, src/watcher.ts.

Source

1import type { Env } from "./env";
2import type { Agent } from "./agents";
3import { missingForRecord, type JobRow, type StartedNotice } from "./jobs-schema";
4import { isMissingRowAbort, requireJobUnchanged } from "./store-guards";
5import { loadRecordRows, recordFor } from "./agent-record";
6import type { JobOutcomeRow } from "./job-outcomes";
7import type { JobListRow } from "./jobs-claim";
8import type { OverlapReport } from "./job-overlaps";
9import type { JobVersion } from "./jobs-edit";
10import { jobAudit, mirrorStatements, type ResumeNote } from "./jobs-mirror";
11import { logEvent } from "./log";
12
13// What every job transition shares: the result shape, the refusal, the row read, the
14// guarded batch every transition commits through, and the checks more than one of
15// claim, holder and seat make. The queue's rules are stated in src/jobs.ts.
16
17// claimed_by carries the same shape as audit_log.actor (migrations/0006), so one
18// query joins a job to what its driver did. A minted agent speaks that vocabulary as
19// `agent:<name>` (agentActor in src/agents-schema.ts), and the name is unique in the
20// agents table and never reused, so the string identifies exactly one credential. The
21// admin is `access:<email>` on the MCP login since it moved to Cloudflare Access
22// (docs/auth.md, 2026-09-27) and `github:<login>` on rows the old console wrote before it moved to Access.
23const ACTOR_SHAPE = /^(access:|github:|opkey:|agent:)/;
24
25export function actorShapeRefusal(action: string, actor: string): JobResult | null {
26  if (ACTOR_SHAPE.test(actor)) return null;
27  return refuse(
28    action,
29    `'${actor}' is not a caller identity this queue can hold a claim for. A claim is recorded against an access: email, a github: login, an opkey: fingerprint, or an agent: name.`
30  );
31}
32
33// The seat is identified by admin or can_merge, for supersede and for resume. An
34// agent's kind is descriptive and not authorizing (src/agents-schema.ts); can_merge is
35// the flag the seat holds and no driver does.
36export function callerIsSeat(agent: Agent): boolean {
37  return agent.admin || agent.scopes.flags.can_merge;
38}
39
40export interface JobResult {
41  ok: boolean;
42  action: string;
43  job?: JobRow;
44  jobs?: Array<JobListRow | JobRow>;
45  truncated?: boolean;
46  note?: string;
47  // post and edit: what the body lacks that a session acts best with (src/job-body-check.ts).
48  // A warning never refuses.
49  warnings?: string[];
50  refusal?: string;
51  // The outcome row this transition wrote, returned so the driver sees what the
52  // Worker checked rather than assuming its own numbers were taken. `notes` names
53  // every verification that could not run, which separates a count nobody checked
54  // from a count nobody could check.
55  outcome?: { row: JobOutcomeRow; notes: string[] };
56  // The latest resume's reason, for a job resumed at least once. See
57  // latestResumeNote in src/jobs-mirror.ts.
58  resume_note?: ResumeNote;
59  // Every resume note, newest first (the newest is resume_note), within a byte cap; and how
60  // many the answer leaves out. claim and list for one id (src/jobs-mirror.ts).
61  resume_notes?: ResumeNote[];
62  resume_notes_dropped?: number;
63  // block and complete: the other open pull requests that change the same files as the one
64  // this call named, or why that could not be checked (src/job-overlaps.ts). Absent when the
65  // call named no pull request of the namespace's repo.
66  overlaps?: OverlapReport;
67  // A job somebody marked started (src/jobs-begin.ts): who and when, so a second session
68  // reading it knows. begin, claim and list for one id return it. Never a refusal.
69  started?: StartedNotice;
70  // claim: the model Capsid recommends for this job, why, and how to follow it
71  // (src/model-routing.ts), and a note when the pull request it carries could not be read.
72  routing?: Record<string, unknown>;
73  routing_note?: string;
74  // list view "models": the learning table.
75  models?: Record<string, unknown>;
76  // edit, and list for one named id to a caller holding write: the versions earlier
77  // edits replaced, newest first (src/jobs-edit.ts, migrations/0033).
78  versions?: JobVersion[];
79  // fire: the session the routine started (src/routine-fire.ts). Either field is null
80  // when Anthropic's answer did not carry it in the expected shape.
81  session?: { id: string | null; url: string | null };
82  // routine: a namespace's routine as Capsid holds it. The token is never returned, only
83  // the secret's name and whether it is set.
84  routine?: { namespace: string; routine_id: string | null; token_secret: string; token_set: boolean };
85}
86
87export function refuse(action: string, refusal: string): JobResult {
88  return { ok: false, action, refusal };
89}
90
91export async function readJob(db: D1Database, id: string): Promise<JobRow | null> {
92  return db.prepare("SELECT * FROM jobs WHERE id = ?1").bind(id).first<JobRow>();
93}
94
95// The one path that fails a job that cannot be handed over: a corrupt requirement or
96// a bad signature at claim, a bad signature at resume. The UPDATE, the mirror and the
97// audit row are one guarded batch, so they commit only when the row is still as the
98// caller read it. Without the guard, a job another driver had claimed meanwhile would
99// get a mirror and an audit row saying it failed while its row said claimed. When the
100// guard aborts, the caller gets the current row to refuse with.
101export async function markJobFailed(
102  env: Env,
103  job: JobRow,
104  fromStatus: "queued" | "blocked",
105  summary: string,
106  auditAction: string,
107  actor: string,
108  auditParams: Record<string, unknown>,
109  now: Date
110): Promise<{ failed: true } | { failed: false; current: JobRow | null }> {
111  const failed = { ...job, status: "failed" as const, result_summary: summary, updated_at: now.toISOString() };
112  const committed = await guardedTransition(env, job, [
113    env.DB.prepare(
114      `UPDATE jobs SET status = 'failed', result_summary = ?2, summary_sig = NULL, updated_at = ?3
115       WHERE id = ?1 AND status = ?4 RETURNING id`
116    ).bind(job.id, summary, now.toISOString(), fromStatus),
117    ...(await mirrorStatements(env, failed, auditAction, actor)),
118    jobAudit(env.DB, actor, auditAction, failed, auditParams),
119  ]);
120  if (!committed) return { failed: false, current: await readJob(env.DB, job.id) };
121  return { failed: true };
122}
123
124// Every transition and its records are one batch. Committing the UPDATE alone and
125// writing the mirror and audit row in a second batch would let a throw in between
126// leave a moved row with no record of the move.
127//
128// `read` is the row the caller decided on. requireJobUnchanged, first in the batch,
129// aborts the whole batch unless the row still has that status, holder and updated_at,
130// and D1 runs a batch as one transaction, so the UPDATE and every record built from
131// `read` commit together or not at all. Returns false when the guard aborted, which
132// means nothing was written and the row changed since `read`.
133export async function guardedTransition(env: Env, read: JobRow, statements: D1PreparedStatement[]): Promise<boolean> {
134  try {
135    await env.DB.batch([requireJobUnchanged(env.DB, read.id, read.status, read.claimed_by, read.updated_at), ...statements]);
136    return true;
137  } catch (err) {
138    if (!isMissingRowAbort(err)) throw err;
139    return false;
140  }
141}
142
143/** The refusal for a job markJobFailed found had already moved. */
144export function movedBeforeFailing(action: string, id: string, current: JobRow | null, why: string): JobResult {
145  return refuse(
146    action,
147    `${id} ${why}, but it changed before it could be marked failed: it is now ${current?.status ?? "gone"}${current?.claimed_by ? `, held by ${current.claimed_by}` : ""}. Nothing was written.`
148  );
149}
150
151// The track-record bar, checked at claim (migrations/0011).
152//
153// Read only when a job sets one, which is rare. The record is computed from every
154// outcome row, so making every claim pay for that read would put a table scan in
155// front of the queue's hottest path.
156//
157// It goes through recordFor, the same function capsid_run_improve action status and the Portal call,
158// so the bar a claim is measured against is the number a human can read on the page.
159export async function recordShortfall(db: D1Database, actor: string, job: JobRow): Promise<string | null> {
160  if (!job.min_record) return null;
161  return missingForRecord(recordFor(actor, await loadRecordRows(db)), job.min_record);
162}
163
164// The corrections budget is a property of the work, not of the row.
165//
166// corrections_count lives on a row, and the unique open-title index covers `queued`,
167// `claimed` and `blocked` (migrations/0019) but not `failed`, so a failed job leaves
168// (namespace, title) free to be posted again. The new row starts at 0, and a per-row
169// cap would count how many times a row existed rather than how many times the work
170// had been sent back. So the count is per (namespace, title).
171//
172// Summed across every row for that work, whatever its status, and the current row is
173// one of them. Unindexed on purpose: jobs is a single-user queue of a few hundred rows
174// at most, and the only index on (namespace, title) is the partial one, which does not
175// cover finished rows.
176//
177// Fails closed. A read that throws, or a SUM that is not a finite number, returns
178// NaN, and atCorrectionCap treats a budget it cannot read as a budget already spent.
179export async function correctionsForWork(db: D1Database, namespace: string, title: string): Promise<number> {
180  try {
181    const row = await db
182      .prepare("SELECT COALESCE(SUM(corrections_count), 0) AS spent FROM jobs WHERE namespace = ?1 AND title = ?2")
183      .bind(namespace, title)
184      .first<{ spent: number }>();
185    const spent = row?.spent;
186    return typeof spent === "number" && Number.isFinite(spent) ? spent : Number.NaN;
187  } catch (err) {
188    logEvent("error", "CORRECTIONS_READ_FAILED", { message: `CORRECTIONS_READ_FAILED ${namespace}/${title}: ${err instanceof Error ? err.message : String(err)}` });
189    return Number.NaN;
190  }
191}