src/jobs-claim.ts

Functions

Imports

It imports src/agents.ts, src/env.ts, src/job-breaker.ts, src/jobs-mirror.ts, src/jobs-schema.ts, src/jobs-transition.ts, src/model-routing-pr.ts, src/model-routing.ts, src/provenance.ts, src/signed-docs.ts. Imported by src/jobs-edit.ts, src/jobs-transition.ts, src/jobs.ts.

Source

1import type { Env } from "./env";
2import type { Agent } from "./agents";
3import {
4  JOBS_ROWS_MAX,
5  OPEN_JOB_STATUSES,
6  corruptRequirement,
7  missingForJob,
8  mintJobId,
9  parseRequiredScopes,
10  serializeMinRecord,
11  swallowedParamTag,
12  swallowedTagRefusal,
13  serializeRequiredScopes,
14  startedNotice,
15  type JobRow,
16  type JobStatus,
17  type MinRecord,
18  type RequiredScopes,
19} from "./jobs-schema";
20import { signTaskBody, verifySignedBody } from "./signed-docs";
21import { JOB_KINDS, isJobKind, routeJob, type Routing } from "./model-routing";
22import { pullRequestRisk } from "./model-routing-pr";
23import { jobAudit, mirrorStatements, resumeNotes } from "./jobs-mirror";
24import { breakerRefusal, breakerState } from "./job-breaker";
25import { externalFenceProblem, jobOrigin } from "./provenance";
26import {
27  actorShapeRefusal,
28  guardedTransition,
29  markJobFailed,
30  movedBeforeFailing,
31  readJob,
32  recordShortfall,
33  refuse,
34  type JobResult,
35} from "./jobs-transition";
36
37// Getting work into and out of the queue: post, list and claim.
38
39// The body is signed at post, with the same key and envelope as the signed policy
40// documents (src/signed-docs.ts). The driver refuses a job whose body does not
41// verify, so a row edited by a raw D1 splice, or a document mirrored back over by
42// hand, cannot steer a session that holds local shell and repo credentials.
43//
44// Unconfigured is a refusal, not a skip. With no IMPROVE_SCORE_SECRET there is no key
45// to sign with, and a queue of unsignable jobs would be refused by the driver one at
46// a time instead of here.
47
48/** The open statuses in prose, derived rather than retyped, so a status added to
49 *  OPEN_JOB_STATUSES cannot leave the refusal claiming a shorter list than the index
50 *  enforces. */
51function openMeans(): string {
52  const names = [...OPEN_JOB_STATUSES];
53  const last = names.pop();
54  return names.length ? `${names.join(", ")} or ${last}` : String(last);
55}
56
57/** What a duplicate post collided with. The unique index is the rule and D1's error
58 *  names only the constraint, so the row is read back to say which job holds the
59 *  title and what state it is in. A blocked holder is the case that matters, because
60 *  it is waiting on a human rather than on a driver.
61 *
62 *  The lookup only shapes the message, never the rule: if the holder finished between
63 *  the constraint firing and this read, the refusal still stands. */
64async function duplicateRefusal(env: Env, namespace: string, title: string): Promise<string> {
65  const placeholders = OPEN_JOB_STATUSES.map((_, i) => `?${i + 3}`).join(", ");
66  const holder = await env.DB.prepare(
67    `SELECT id, status FROM jobs WHERE namespace = ?1 AND title = ?2 AND status IN (${placeholders}) LIMIT 1`
68  )
69    .bind(namespace, title, ...OPEN_JOB_STATUSES)
70    .first<{ id: string; status: string }>()
71    .catch(() => null);
72  const named = holder ? `: ${holder.id} is ${holder.status}` : "";
73  const waiting = holder?.status === "blocked" ? " That one is blocked, which means somebody is already waiting on it." : "";
74  return `${namespace} already has an open job titled '${title}'${named}. Finish or fail that one first, or post this under a different title. Open means ${openMeans()}.${waiting}`;
75}
76
77/** The body checks and the signature, shared by post and edit so an edited body is
78 *  refused for what post refuses and signed exactly as post signs it. Returns the
79 *  stored (signed) text, or the problem to refuse with. */
80export async function signedJobBody(env: Env, body: string): Promise<{ ok: true; signed: string } | { ok: false; problem: string }> {
81  if (!env.IMPROVE_SCORE_SECRET) {
82    return {
83      ok: false,
84      problem:
85        "job signing is not configured on this Worker (IMPROVE_SCORE_SECRET is unset), so this job could not be signed and no driver would execute it. Refusing rather than queueing work nothing can verify.",
86    };
87  }
88  if (!body.trim()) return { ok: false, problem: "a job needs a body. The body is the prompt the driver executes." };
89  // A body is the prompt a driver executes, so a malformed post is worse here than
90  // anywhere else: the swallowed text is signed along with everything else and the
91  // driver runs whatever survived.
92  const swallowed = swallowedParamTag(body);
93  if (swallowed) return { ok: false, problem: swallowedTagRefusal("body", swallowed) };
94  // An external fence is how relayed text is told from instruction (src/provenance.ts), so
95  // a body whose fence is unlabelled or never closes is refused rather than signed.
96  const fenceProblem = externalFenceProblem(body);
97  if (fenceProblem) return { ok: false, problem: `the body has a malformed external fence: ${fenceProblem}. Nothing was written.` };
98  return { ok: true, signed: await signTaskBody(env.IMPROVE_SCORE_SECRET, body) };
99}
100
101export async function postJob(
102  env: Env,
103  agent: Agent,
104  now: Date,
105  args: {
106    namespace: string;
107    title: string;
108    body: string;
109    priority?: number;
110    gate_required?: boolean;
111    required_scopes?: Partial<RequiredScopes>;
112    min_record?: MinRecord;
113    review_required?: boolean;
114    // Optional: when absent Capsid reads it from the title and the poster (src/model-routing.ts).
115    kind?: string;
116  }
117): Promise<JobResult> {
118  const actor = agent.actor;
119  const title = args.title.trim();
120  if (!title) return refuse("post", "a job needs a title: it is how the queue refuses a duplicate while one is still open.");
121  const signedBody = await signedJobBody(env, args.body);
122  if (!signedBody.ok) return refuse("post", signedBody.problem);
123  // A registered namespace, as write requires for a document. Otherwise a caller
124  // scoped to * could post into a namespace that does not exist, and the mirror
125  // document would land where write refuses the same path.
126  const registered = await env.DB.prepare("SELECT namespace FROM namespaces WHERE namespace = ?1").bind(args.namespace).first();
127  if (!registered) {
128    return refuse(
129      "post",
130      `unknown namespace '${args.namespace}'. Nothing was written. A job and its mirror document live in a registered namespace; check the spelling against capsid_find_docs action namespaces, or have the admin create it with capsid_admin_manage_namespaces action register.`
131    );
132  }
133
134  if (args.kind !== undefined && !isJobKind(args.kind)) {
135    return refuse("post", `'${args.kind}' is not a job kind. One of: ${JOB_KINDS.join(", ")}, or leave it out and Capsid reads it from the title.`);
136  }
137  // The model is assigned here, by the Worker, from the kind and the risk the job states
138  // about itself. Nobody types it.
139  const id = mintJobId();
140  const routing = routeJob({
141    id,
142    title,
143    body: args.body,
144    namespace: args.namespace,
145    posted_by: actor,
146    kind: args.kind,
147    required_flags: args.required_scopes?.flags,
148    gate_required: args.gate_required,
149  });
150
151  const job: JobRow = {
152    id,
153    namespace: args.namespace,
154    title,
155    body: signedBody.signed,
156    priority: args.priority ?? 0,
157    status: "queued",
158    posted_by: actor,
159    claimed_by: null,
160    claimed_at: null,
161    result_ref: null,
162    result_summary: null,
163    gate_required: args.gate_required ? 1 : 0,
164    // NULL when nothing was asked for, rather than an empty requirement object. The
165    // column means "this job needs something unusual", and empty JSON would read as a
166    // requirement nobody can see.
167    required_scopes: args.required_scopes?.flags?.length ? serializeRequiredScopes(args.required_scopes) : null,
168    // A bar of zero is no bar, so it is stored as NULL rather than as a requirement
169    // every agent trivially meets.
170    min_record: args.min_record?.prs_merged ? serializeMinRecord(args.min_record) : null,
171    blocked_count: 0,
172    resumed_count: 0,
173    corrections_count: 0,
174    review_required: args.review_required ? 1 : 0,
175    kind: routing.kind,
176    model_recommended: routing.model,
177    effort_recommended: routing.effort,
178    routing_reason: routing.reason,
179    created_at: now.toISOString(),
180    updated_at: now.toISOString(),
181  };
182
183  const statements = [
184    env.DB.prepare(
185      `INSERT INTO jobs (id, namespace, title, body, priority, status, posted_by, gate_required, required_scopes, min_record, review_required, created_at, updated_at,
186         kind, model_recommended, effort_recommended, routing_reason)
187       VALUES (?1, ?2, ?3, ?4, ?5, 'queued', ?6, ?7, ?8, ?9, ?10, ?11, ?11, ?12, ?13, ?14, ?15)`
188    ).bind(
189      job.id,
190      job.namespace,
191      job.title,
192      job.body,
193      job.priority,
194      job.posted_by,
195      job.gate_required,
196      job.required_scopes,
197      job.min_record,
198      job.review_required,
199      job.created_at,
200      routing.kind,
201      routing.model,
202      routing.effort,
203      routing.reason
204    ),
205    ...(await mirrorStatements(env, job, "job-posted", actor)),
206    jobAudit(env.DB, actor, "job-posted", job, {
207      origin: jobOrigin(actor),
208      title: job.title,
209      priority: job.priority,
210      gate_required: job.gate_required,
211      ...(job.required_scopes ? { required_scopes: job.required_scopes } : {}),
212      ...(job.min_record ? { min_record: job.min_record } : {}),
213      ...(job.review_required ? { review_required: true } : {}),
214      model_recommended: routing.model,
215      effort_recommended: routing.effort,
216      kind: routing.kind,
217    }),
218  ];
219  try {
220    await env.DB.batch(statements);
221  } catch (err) {
222    const message = err instanceof Error ? err.message : String(err);
223    // The partial unique index over (namespace, title) for open statuses, reported as
224    // what it means rather than as the constraint's own text.
225    if (/UNIQUE/i.test(message)) {
226      return refuse("post", await duplicateRefusal(env, args.namespace, title));
227    }
228    throw err;
229  }
230  return { ok: true, action: "post", job };
231}
232
233// List never carries a body unless one job was asked for by a writer. A body is the
234// signed prompt a driver executes with shell and repo credentials, and a list that
235// selected every column would let a read grant on a namespace read every prompt queued
236// in it. The columns are named so the body never leaves D1 for a list. claim returns
237// it, and so does list when `withBody` is set, which the tool sets only for a single
238// named id and a caller holding write.
239const JOB_LIST_COLUMNS = [
240  "id",
241  "namespace",
242  "title",
243  "status",
244  "priority",
245  "posted_by",
246  "claimed_by",
247  "gate_required",
248  "review_required",
249  "blocked_count",
250  "resumed_count",
251  "created_at",
252  "updated_at",
253  "result_summary",
254  "kind",
255  "model_recommended",
256  "effort_recommended",
257  "started_at",
258  "started_by",
259] as const;
260
261export type JobListRow = Pick<JobRow, (typeof JOB_LIST_COLUMNS)[number]>;
262
263export async function listJobs(
264  env: Env,
265  args: { namespace?: string; status?: JobStatus; id?: string },
266  opts: { withBody?: boolean } = {}
267): Promise<JobResult> {
268  const where: string[] = [];
269  const binds: unknown[] = [];
270  const withBody = opts.withBody === true && Boolean(args.id);
271  if (args.id) {
272    binds.push(args.id);
273    where.push(`id = ?${binds.length}`);
274  }
275  if (args.namespace) {
276    binds.push(args.namespace);
277    where.push(`namespace = ?${binds.length}`);
278  }
279  if (args.status) {
280    binds.push(args.status);
281    where.push(`status = ?${binds.length}`);
282  }
283  binds.push(JOBS_ROWS_MAX + 1);
284  const clause = where.length ? `WHERE ${where.join(" AND ")}` : "";
285  const { results } = await env.DB.prepare(
286    `SELECT ${[...JOB_LIST_COLUMNS, ...(withBody ? ["body"] : [])].join(", ")} FROM jobs ${clause} ORDER BY priority DESC, created_at ASC LIMIT ?${binds.length}`
287  )
288    .bind(...binds)
289    .all<JobListRow>();
290  const rows = results ?? [];
291  // One extra row asked for, so "exactly the page" is distinguishable from "there are
292  // more". Same shape as every other bounded read here.
293  const truncated = rows.length > JOBS_ROWS_MAX;
294  // One named job carries its latest resume note. Not every row of a wide list, which
295  // would cost one read per resumed job.
296  const listNotes = args.id && rows.length === 1 ? await resumeNotes(env, rows[0]) : null;
297  const listNote = listNotes?.notes[0] ?? null;
298  // One named job somebody began says who and when (src/jobs-begin.ts).
299  const listStarted = args.id && rows.length === 1 ? startedNotice(rows[0]) : null;
300  return {
301    ok: true,
302    action: "list",
303    ...(listStarted ? { started: listStarted } : {}),
304    ...(listNote ? { resume_note: listNote } : {}),
305    ...(listNotes && listNotes.notes.length > 0 ? { resume_notes: listNotes.notes } : {}),
306    ...(listNotes && listNotes.dropped > 0 ? { resume_notes_dropped: listNotes.dropped } : {}),
307    jobs: truncated ? rows.slice(0, JOBS_ROWS_MAX) : rows,
308    ...(truncated ? { truncated: true, note: `more than ${JOBS_ROWS_MAX} jobs match; narrow by namespace or status.` } : {}),
309  };
310}
311
312/** What a claim hands the driver about the model (capsid/conventions.md, "Model routing"): the
313 *  recommendation, why, and how to follow it. The driver never picks a model itself. */
314function deliveryOf(routing: Routing) {
315  return {
316    model: routing.model,
317    effort: routing.effort,
318    kind: routing.kind,
319    risk: routing.risk,
320    reason: routing.reason,
321    trial: routing.trial,
322    rules_version: routing.rules_version,
323    deliver:
324      `A scheduled or cloud launch starts on --model ${routing.model}. An interactive driver keeps its own session for coordination and does the job's work in a subagent on model ${routing.model} (the Agent tool's model parameter), at effort ${routing.effort}.`,
325  };
326}
327
328// A claim names the caller as the job's holder, with no lease: it never expires, and a
329// caller may hold any number of jobs, on any repo. Every holder action names the job it
330// acts on. A holder that is gone is released by the seat (releaseJob).
331export async function claimJob(
332  env: Env,
333  agent: Agent,
334  now: Date,
335  args: { namespace?: string; id?: string }
336): Promise<JobResult> {
337  const actor = agent.actor;
338  const badActor = actorShapeRefusal("claim", actor);
339  if (badActor) return badActor;
340  // Either a named job or the highest-priority queued one in a namespace. The SELECT
341  // only picks a candidate; the guarded UPDATE below is what claims it, so two drivers
342  // reading the same candidate still resolve to one winner.
343  let candidate: JobRow | null = null;
344  if (args.id) {
345    candidate = await readJob(env.DB, args.id);
346    if (!candidate) return refuse("claim", `no job ${args.id}.`);
347    if (candidate.status !== "queued") {
348      return refuse("claim", `${args.id} is ${candidate.status}, not queued${candidate.claimed_by ? ` (held by ${candidate.claimed_by})` : ""}.`);
349    }
350  } else {
351    if (!args.namespace) return refuse("claim", "claim needs a namespace to pick from, or an id to claim.");
352    candidate = await env.DB.prepare(
353      "SELECT * FROM jobs WHERE namespace = ?1 AND status = 'queued' ORDER BY priority DESC, created_at ASC LIMIT 1"
354    )
355      .bind(args.namespace)
356      .first<JobRow>();
357    if (!candidate) return refuse("claim", `no queued jobs in ${args.namespace}.`);
358  }
359
360  // A namespace whose drivers keep failing stops handing out work until a person looks
361  // (src/job-breaker.ts). Checked before anything else touches the candidate, so an
362  // open breaker leaves the queue exactly as it was.
363  const breaker = await breakerState(env, candidate.namespace, now);
364  if (breaker.open) return refuse("claim", `${actor} cannot claim ${candidate.id}: ${breakerRefusal(breaker)}`);
365
366  // What the job needs of the driver is checked before the job is taken. A claim
367  // that took the job and then refused would park it on a driver that cannot do it.
368  // The check runs through checkScope, so a job
369  // requirement and an agent scope are compared by the same function that decides
370  // every tool call.
371  //
372  // A requirement that cannot be read fails the job, as a bad signature does below:
373  // left queued, the same row would be refused to every driver in turn, and a claim
374  // with no id takes the top queued job, so it would stop the namespace's queue.
375  const corrupt = corruptRequirement(candidate);
376  if (corrupt) {
377    const reason = `${candidate.id} has a corrupt requirement: ${corrupt}. A requirement that cannot be read is not the same as none, so it was not handed out.`;
378    const marked = await markJobFailed(env, candidate, "queued", reason, "job-requirement-corrupt", actor, { reason: corrupt }, now);
379    if (!marked.failed) return movedBeforeFailing("claim", candidate.id, marked.current, "has a corrupt requirement");
380    return refuse("claim", `${reason} It has been marked failed.`);
381  }
382
383  const missing = missingForJob(agent, candidate.namespace, candidate.required_scopes);
384  if (missing) {
385    return refuse(
386      "claim",
387      `${actor} cannot claim ${candidate.id} ('${candidate.title}'): ${missing} The job stays queued for a driver that can do it.`
388    );
389  }
390
391  // What it needs of the driver's history, on the same terms and in the same place:
392  // before the job is taken, so it stays queued for a driver that meets the bar.
393  const shortfall = await recordShortfall(env.DB, actor, candidate);
394  if (shortfall) {
395    return refuse("claim", `${actor} cannot claim ${candidate.id} ('${candidate.title}'): ${shortfall} The job stays queued for a driver that can do it.`);
396  }
397
398  // The signature is checked before the job is handed over, not by the driver after
399  // it has one. A job body is executable input that arrives as a database row, and the
400  // driver is a session holding local shell and repo credentials.
401  //
402  // A body that does not verify was edited after `post` signed it. That job is failed
403  // here rather than left queued: leaving it would hand the same broken row to every
404  // driver in turn, and the queue would never drain.
405  //
406  // Unlike a run document (verifyTaskDoc) there is no actor check. Only the loop writes
407  // a run document, so its audit actor is the loop; a job is posted by a human seat,
408  // and this Worker's signature is what proves it went through `post`.
409  const verdict = await verifySignedBody(env.IMPROVE_SCORE_SECRET, candidate.body, "job body");
410  if (!verdict.ok) {
411    const marked = await markJobFailed(env, candidate, "queued", verdict.reason, "job-signature-refused", actor, { reason: verdict.reason }, now);
412    if (!marked.failed) return movedBeforeFailing("claim", candidate.id, marked.current, "failed its signature check");
413    return refuse("claim", `${candidate.id} failed its signature check and has been marked failed: ${verdict.reason}`);
414  }
415
416  // The approval a resume handed on is checked the same way, before the job is taken: a note
417  // changed after the resume wrote it is an instruction nobody gave, so the job is
418  // failed rather than handed to a driver with it (src/job-signing.ts).
419  const claimNotes = await resumeNotes(env, candidate);
420  const claimNote = claimNotes.notes[0] ?? null;
421  if (claimNote?.signature === "mismatch") {
422    const reason = `its last resume note's signature does not match: the approval was changed after ${claimNote.by} wrote it at ${claimNote.at}.`;
423    const marked = await markJobFailed(env, candidate, "queued", reason, "job-resume-note-refused", actor, { reason }, now);
424    if (!marked.failed) return movedBeforeFailing("claim", candidate.id, marked.current, "carries a resume note that failed its signature check");
425    return refuse("claim", `${candidate.id} was not handed out and has been marked failed: ${reason}`);
426  }
427
428  // The model for this job, assigned at post and settled again here: a job posted before
429  // routing existed is routed now, and a job that already carries a pull request is read
430  // on that PR's files, which can only raise the risk (src/model-routing.ts riskOf).
431  const required = parseRequiredScopes(candidate.required_scopes);
432  const prRisk = await pullRequestRisk(env, candidate.namespace, candidate.result_ref);
433  const routing = routeJob({
434    id: candidate.id,
435    title: candidate.title,
436    body: candidate.body,
437    namespace: candidate.namespace,
438    posted_by: candidate.posted_by,
439    kind: candidate.kind,
440    required_flags: required.ok ? required.value.flags : [],
441    gate_required: candidate.gate_required === 1,
442    pr_risk: prRisk.risk,
443  });
444
445  const holder = actor;
446  const claimed: JobRow = {
447    ...candidate,
448    kind: routing.kind,
449    model_recommended: routing.model,
450    effort_recommended: routing.effort,
451    routing_reason: routing.reason,
452    status: "claimed",
453    claimed_by: holder,
454    // A job the seat released keeps its first claim time, so its duration covers its
455    // whole working life.
456    claimed_at: candidate.claimed_at ?? now.toISOString(),
457    updated_at: now.toISOString(),
458  };
459  // One batch with its records, behind the guard. Two drivers reading the same
460  // candidate still resolve to one winner, because the first commit moves updated_at
461  // and the second batch's guard aborts before its UPDATE runs.
462  const won = await guardedTransition(env, candidate, [
463    env.DB.prepare(
464      `UPDATE jobs SET status = 'claimed', claimed_by = ?2, claimed_at = COALESCE(claimed_at, ?3), updated_at = ?3,
465         kind = ?4, model_recommended = ?5, effort_recommended = ?6, routing_reason = ?7
466       WHERE id = ?1 AND status = 'queued' RETURNING id`
467    ).bind(candidate.id, holder, now.toISOString(), routing.kind, routing.model, routing.effort, routing.reason),
468    ...(await mirrorStatements(env, claimed, "job-claimed", actor)),
469    jobAudit(env.DB, actor, "job-claimed", claimed, {}),
470  ]);
471  if (!won) {
472    return refuse("claim", `${candidate.id} was claimed by someone else between reading it and taking it. Ask again.`);
473  }
474  const startedOn = startedNotice(candidate);
475  // A job that went back to the queue after a resume reaches its next driver here, so
476  // the approval has to come with it: the note read and checked above.
477  return {
478    ok: true,
479    action: "claim",
480    job: claimed,
481    // A label refuses nothing; the claimant is told who began it.
482    ...(startedOn ? { started: startedOn } : {}),
483    ...(claimNote ? { resume_note: claimNote } : {}),
484    ...(claimNotes.notes.length > 0 ? { resume_notes: claimNotes.notes } : {}),
485    ...(claimNotes.dropped > 0 ? { resume_notes_dropped: claimNotes.dropped } : {}),
486    routing: deliveryOf(routing),
487    ...(prRisk.note ? { routing_note: prRisk.note } : {}),
488  };
489}