src/tools/jobs.ts

Functions

Imports

It imports src/agents-schema.ts, src/job-body-check.ts, src/job-checklist.ts, src/job-claims.ts, src/job-fact-check.ts, src/job-outcomes.ts, src/jobs-schema.ts, src/jobs.ts, src/limits.ts, src/model-learning.ts, src/model-routing.ts, src/overnight-plan.ts, src/routine-fire.ts, src/scope.ts, src/stale-jobs.ts, src/tools/docs.ts. Imported by src/server.ts.

Source

1import { McpServer } from "@modelcontextprotocol/server";
2import { JOBS_TOOL } from "../scope";
3import { z } from "zod";
4import { bounded, MAX_BODY, MAX_RESUME_NOTE, MAX_TITLE, nsName, resultRef } from "../limits";
5import { digestSinceFrom, readOvernightDigest, readOvernightPlan } from "../overnight-plan";
6import { readModelLearning } from "../model-learning";
7import { JOB_KINDS } from "../model-routing";
8import { CORRECTION_CAP, JOB_ACTIONS, JOB_STATUSES, isJobStatus } from "../jobs-schema";
9import { SCOPE_FLAGS } from "../agents-schema";
10import { beginJob, blockJob, claimJob, completeAsCaller, recordAsCaller, editJob, failAsCaller, jobVersions, listJobs, parkJob, postJob, releaseJob, resumeJob, supersedeJob, unparkJob, type JobResult } from "../jobs";
11import { fireRoutine, setRoutine } from "../routine-fire";
12import { parseEvidence } from "../job-outcomes";
13import { claimSchema } from "../job-claims";
14import { staleJobs } from "../stale-jobs";
15import { jobTextRefusal, missingPartsWarning } from "../job-body-check";
16import { checklistWarning } from "../job-checklist";
17import { factCheckBody } from "../job-fact-check";
18import { fail, ok, type ToolCtx } from "./docs";
19
20const MAX_JOB_ID = 64;
21
22// How many pull requests one job may name as evidence. Each one costs a GitHub read
23// inside `complete`, so the bound stops one call fanning out without limit.
24const MAX_EVIDENCE_PRS = 10;
25
26// The caller-facing body checks (src/job-body-check.ts): post refuses an unfilled placeholder
27// and a claim read as false (src/job-fact-check.ts) here, edit inside editJob once the caller
28// is known to be the seat, and both warn when a body lacks the five headed parts or states a
29// claim that could not be checked.
30function withWarnings(result: JobResult, body: string | undefined, facts: string[] = []): JobResult {
31  if (!result.ok) return result;
32  const missing = body === undefined ? null : missingPartsWarning(body);
33  const long = body === undefined ? null : checklistWarning(body);
34  const warnings = [...(result.warnings ?? []), ...facts, ...(missing ? [missing] : []), ...(long ? [long] : [])];
35  return warnings.length > 0 ? { ...result, warnings } : result;
36}
37
38// A refusal is an error to the client (isError true), so a client that keys on
39// isError does not read it as success. The body is the same JSON either way.
40function reply(result: JobResult) {
41  return result.ok ? ok(result) : { ...ok(result), isError: true };
42}
43
44export function defineJobOperations(server: McpServer, ctx: ToolCtx): void {
45  const { env, agent } = ctx;
46
47  // The work queue's one tool, a ruled exception to the tool surface rule (CLAUDE.md).
48  // Its actions share one tool because they are one subsystem
49  // with one row shape, and a caller that has the tool has the whole lifecycle.
50  ctx.ops.define(
51    "jobs",
52    {
53      description:
54        `The work queue. Each job mirrors to <namespace>/jobs/<id>.md. list needs the read grant; every other action needs the write grant. "Seat" below means the admin or a can_merge caller. Full rules: docs/work-queue.md.
55post: queue a job from namespace, title and body. The body is signed; a driver refuses one that does not verify. Refused while another open job has the same (namespace, title), or when the title or body holds an unfilled placeholder (<ID>, {{...}}, TODO:, TBD) outside a code block; warns when the body lacks GOAL, DO, OUT OF SCOPE, DONE WHEN or DECIDED/RULES parts. A body stating as done that a PR merged, a version was tagged or published to npm is checked and refused when false.
56list: filter by namespace, status and id, without bodies; one named id returns its body and edit versions to a writer. stale: true returns blocked or claimed jobs that look stuck (unchanged, resumed-not-completed, prs-settled, steps-seen). view: plan, digest or models.
57claim: take the top queued job in a namespace, or a named id. The claim never expires. Returns the recommended model. Refused without the job's required_flags or min_record, or while the namespace's breaker is open.
58begin: label a queued job started (who, when). Verifies the body's signature; never refuses otherwise.
59complete: needs result_summary; writes one outcome row and verifies each named pull request against GitHub. Refused until CI has passed on the head of each one it names: running waits, red is refused, and no passing check on a code change is not green. Also closes a queued job nobody holds when it names the job's pull request (the pull request is the claim), and lets the seat close a blocked job another credential holds.
60record: the seat's close of a queued job somebody else did, with the same review gate and evidence checks.
61block: needs reason and the command the human must run, or question: true with the question as the reason, answered by a resume note.
62Complete and block that name a pull request also report open pull requests sharing a file (overlaps); a failed read says not checked and never blocks.
63fail: needs reason. The seat may fail a job somebody else holds.
64complete, fail and block take an optional claim (what the agent says it did), recorded before anything is verified.
65resume: returns a blocked job to claimed for its blocker, or to the caller with take; a job blocked by a shared identity (access:, github:, opkey:) goes back to queued. Needs reason and re-verifies the signature. A plain resume needs the seat; the claimant may resume on approved_by_policy for a branch push or a pull request. correction spends one correction; corrections are capped at ${CORRECTION_CAP} per piece of work, past which only the admin may resume.
66supersede: end a job replaced before work began; needs reason. Any writer on a queued job; the holder or the seat on a claimed job with no work recorded.
67release: the seat returns a job another credential holds to the queue; needs reason.
68edit: the seat changes title, body, priority or gate_required of a queued or blocked job; the old version is kept.
69park / unpark: the seat sets a queued job aside with a one-line reason, or returns it to queued. A parked job is never claimed or planned and holds no title slot.
70routine: read a namespace's routine, or with routine_url set or clear ("off") its fire URL. Seat only.
71fire: the seat starts a cloud session for one queued, verified job through its namespace's routine (token in ROUTINE_<NS>_TOKEN) and marks it started. Refusals name their cause (already started or fired, no routine or token, Anthropic's answer).
72block and complete act only on a job this caller holds, except as stated above.`,
73      inputSchema: {
74        action: z.enum(JOB_ACTIONS).describe("post | list | claim | complete | record | fail | block | resume | supersede | release | begin | fire | routine | edit | park | unpark."),
75        namespace: nsName.optional().describe('For post, the namespace the work belongs to. For list and claim, the namespace to filter or pick from. For routine, the namespace whose routine is read or set.'),
76        title: bounded(MAX_TITLE).optional().describe('For post and edit: the title. One open job per (namespace, title).'),
77        body: bounded(MAX_BODY).optional().describe('For post and edit: the full prompt the driver executes. Signed on the way in.'),
78        priority: z.number().int().optional().describe('For post and edit: higher runs first. Defaults to 0 at post.'),
79        gate_required: z.boolean().optional().describe('For post and edit: the work needs a human confirmation (a push, a deploy, a secret).'),
80        review_required: z
81          .boolean()
82          .optional()
83          .describe(
84            "For post: complete and block on a job carrying a pull request are refused until a PR comment starts 'REVIEW:' and ends APPROVE, CHANGES or BLOCK; the newest counts. APPROVE must quote the current head sha (at least 7 hex characters). CHANGES sends the job back and spends a correction; BLOCK blocks it with the objection as the reason."
85          ),
86        required_flags: z
87          .array(z.enum(SCOPE_FLAGS))
88          .optional()
89          .describe(
90            `For post: the flags the claiming driver must hold (${SCOPE_FLAGS.join(", ")}). The claim refuses an agent without them and leaves the job queued.`
91          ),
92        min_record: z
93          .object({ prs_merged: z.number().int().nonnegative() })
94          .optional()
95          .describe(
96            "For post: the minimum merged pull requests on the claiming agent's record. The claim refuses an agent below it and leaves the job queued."
97          ),
98        kind: z
99          .enum(JOB_KINDS)
100          .optional()
101          .describe(
102            `For post: the job's kind (${JOB_KINDS.join(", ")}). Optional: when absent Capsid reads it from the poster and the title. The kind and the job's risk decide the model Capsid recommends; nobody names a model.`
103          ),
104        status: bounded(32).optional().describe(`For list: one of ${JOB_STATUSES.join(" | ")}.`),
105        view: z
106          .enum(["plan", "digest", "models"])
107          .optional()
108          .describe(
109            'For list: "plan" returns the overnight plan, per repo the queued jobs in priority order that fit about 8 hours, gated ones included and marked gated (the session does the ordinary work and lists each risky step in its pull request), parked ones never, with what was skipped and why (a hand-started session can follow it); "digest" returns the morning digest, the pull requests ready, what blocked and why, and each job\'s usage. "models" returns what the model routing has learned: per job kind and model, the share merged with green CI, corrections and cost, with the rule changes the evidence supports and each trial arm\'s verdict against its kind\'s rule (never applied). All three are read-only. A caller scoped to one namespace names it; the plan is built over every namespace and narrowed to the one named.'
110          ),
111        stale: z
112          .boolean()
113          .optional()
114          .describe(
115            "For list: true returns the stale view, the blocked and claimed jobs that look stuck, each with rule (unchanged | resumed-not-completed | prs-settled | steps-seen) and a one-line reason. Read-only and costs no GitHub reads. Refused with view, status or id, and on any other action."
116          ),
117        since: bounded(40).optional().describe('For list with view "digest": the ISO time to report from. Defaults to the last 24 hours.'),
118        id: bounded(MAX_JOB_ID).optional().describe('The job id: optional for claim, required for complete, fail, block, resume, supersede, release, begin, fire, edit, park and unpark. For list, narrows to that job, with its body for a caller holding write.'),
119        result_summary: bounded(MAX_TITLE).optional().describe('For complete: what happened, in one sentence.'),
120        result_ref: resultRef.optional().describe('For complete: where the work landed, a document key or a PR URL.'),
121        reason: bounded(MAX_TITLE).optional().describe('For fail and block: why. For resume: what was approved, in one line; recorded in the audit row and returned as resume_note. For supersede: why the job was replaced or withdrawn. For park: one line saying when the job comes back. For release: why the holder is not coming back.'),
122        note: bounded(MAX_RESUME_NOTE)
123          .optional()
124          .describe(
125            "For resume: the full approval when it is longer than a line. Recorded in the audit row and returned as resume_note.note and in the mirrored document. reason is still required."
126          ),
127        replaced_by: bounded(MAX_JOB_ID)
128          .optional()
129          .describe("For supersede: the id of the replacing job, which must exist, be in the same namespace and not be this job. Omit it for a withdrawal."),
130        command: bounded(MAX_TITLE).optional().describe('For block: the exact command the human must run.'),
131        routine_url: bounded(256)
132          .optional()
133          .describe('For routine: the namespace\'s fire URL, https://api.anthropic.com/v1/claude_code/routines/trig_.../fire, or "off" to clear it. Omit it to read the routine.'),
134        question: z
135          .boolean()
136          .optional()
137          .describe(
138            "For block: true when the session needs an answer, not a command run. reason is the question; the Worker fixes the command as 'Answer in a resume note' (a command sent with it is refused). The job is marked a question in capsid_run_improve action status and the Portal's feed, and the seat answers with resume and a note, which every later claim and list for the job returns."
139          ),
140        approved_by_policy: bounded(32)
141          .optional()
142          .describe(
143            "For resume: the signed gate policy version (capsid/policy/gates.md) the approval is made under. The blocked command must match a policy class (an additive migration, a branch push, opening a pull request) or the resume is refused. The admin or a can_merge caller may approve any class; a driver only its own job, and only a branch push or a pull request."
144          ),
145        take: z
146          .boolean()
147          .optional()
148          .describe("For resume: the caller takes the claim instead of the driver that blocked the job. Runs every check a claim runs."),
149        correction: z
150          .boolean()
151          .optional()
152          .describe("For resume: sends the work back to be corrected and spends one correction. A plain resume or an admin resume spends none."),
153        evidence: z
154          .union([
155            z.object({
156              prs: z.array(resultRef).max(MAX_EVIDENCE_PRS).optional().describe("Pull request URLs this job produced."),
157              commits: z.number().int().nonnegative().optional(),
158              files_changed: z.number().int().nonnegative().optional(),
159              tests_added: z.number().int().nonnegative().optional(),
160            }),
161            // A JSON string is accepted too. An MCP client caches the tool schema at
162            // connect time, so a session that connected before this parameter existed
163            // refuses the object locally, and some clients flatten an object to a
164            // string. Both are a caller that knows what it did and cannot say so in the
165            // shape asked for; refusing them would leave the outcome row storing nulls.
166            bounded(MAX_BODY),
167          ])
168          .optional()
169          .describe(
170            "For complete: what the work produced, as an object or a JSON string. An omitted field is stored as NULL, not 0. For each named pull request the stored merge state, commit count and file count are read from GitHub; tests_added is stored as given."
171          ),
172        // The same object-or-string union as evidence, for the same reason. The object
173        // is strict (src/job-claims.ts), and the string form is parsed against the same
174        // schema, so an unknown key is refused either way rather than dropped.
175        claim: z
176          .union([claimSchema, bounded(MAX_BODY)])
177          .optional()
178          .describe(
179            "For complete, fail and block: what the agent says it did, as an object or a JSON string. prs_opened and prs_merged (pull request URLs), tests {run, passed, failed, result: pass|fail|partial|not_run}, deploy_state (none|pending|deployed|verified|failed), files_touched (paths), usage {cost_usd, active_seconds, tokens {input, output, cache_read, cache_creation}} for this session only (self-reported), versions {model_id, client_name, client_version, permission_mode} (self-reported). Recorded as sent, apart from what the Worker verifies. An omitted field is NULL, never 0; an unknown key or a string that is not JSON is refused."
180          ),
181      },
182    },
183    async (args) => {
184      const now = new Date();
185      try {
186        if (args.stale !== undefined && args.action !== "list") {
187          return fail(`stale is for action list only; ${args.action} does not take it.`);
188        }
189        if (args.action === "list") {
190          // The read action needs the read grant. The registrar names no grant for
191          // an "action" tool, so it is checked here, and a read the caller may not
192          // make is refused for its own reason rather than by a neighbouring check.
193          const listRefusal = ctx.scope({ tool: JOBS_TOOL, action: "list", grant: "read", namespace: args.namespace });
194          if (listRefusal) return fail(listRefusal);
195          // The stale view (src/stale-jobs.ts). Its own view of the queue, so it does not
196          // combine with another view, a status filter or one id. Narrowed to the namespace
197          // the caller named; a caller scoped to one namespace must name it (the registrar).
198          if (args.stale === true) {
199            if (args.view !== undefined || args.status !== undefined || args.id !== undefined) {
200              return fail("stale: true is its own view of the queue; send it without view, status or id (namespace may narrow it).");
201            }
202            const stale = await staleJobs(env, now, { namespace: args.namespace });
203            return ok({
204              ok: true,
205              action: "list",
206              stale: stale.rows,
207              ...(stale.truncated ? { truncated: true } : {}),
208              ...(stale.note ? { note: stale.note } : {}),
209            });
210          }
211          if (args.view === "plan") return ok(await readOvernightPlan(env, { namespace: args.namespace }, now));
212          // Outcomes across every namespace, as a rate per kind and model: a caller scoped to
213          // one namespace has no business reading the whole account's record.
214          if (args.view === "models") {
215            const wide = ctx.scope({ tool: JOBS_TOOL, action: "list", grant: "read", namespace: "*" });
216            if (wide) return fail(`the models view reads every namespace's outcomes, which this caller may not: ${wide}`);
217            return ok({ ok: true, action: "list", models: await readModelLearning(env.DB, now) });
218          }
219          if (args.view === "digest") {
220            const since = digestSinceFrom(args.since, now);
221            if (!since.ok) return fail(since.refusal);
222            return ok(await readOvernightDigest(env, { namespace: args.namespace, since: since.since }, now));
223          }
224          if (args.status !== undefined && !isJobStatus(args.status)) {
225            return fail(`'${args.status}' is not a job status. One of: ${JOB_STATUSES.join(", ")}.`);
226          }
227          // The body only for ONE named job and a caller that could claim it anyway.
228          // Asked through ctx.scope, the one shape a handler may use for a grant that
229          // depends on the action.
230          const withBody =
231            Boolean(args.id) && ctx.scope({ tool: JOBS_TOOL, action: "list", grant: "write", namespace: args.namespace }) === null;
232          const listed = await listJobs(env, { namespace: args.namespace, status: args.status, id: args.id }, { withBody });
233          // The versions earlier edits replaced carry old bodies, so they go to the same
234          // caller the body does.
235          if (withBody && args.id && listed.jobs?.length === 1) listed.versions = await jobVersions(env.DB, args.id);
236          return ok(listed);
237        }
238        // Every other action changes the queue, so it needs the write grant, checked
239        // here with the action so the tools axis can narrow it (a watcher scoped to
240        // `capsid_work_jobs.post` may post and may not claim).
241        const refusal = ctx.scope({ tool: JOBS_TOOL, action: args.action, grant: "write", namespace: args.namespace });
242        if (refusal) return fail(refusal);
243        switch (args.action) {
244          case "post": {
245            if (!args.namespace || !args.title || !args.body) {
246              return fail("post needs namespace, title and body.");
247            }
248            const unfilled = jobTextRefusal(args.title, args.body);
249            if (unfilled) return reply({ ok: false, action: "post", refusal: unfilled });
250            const facts = await factCheckBody(env, args.namespace, args.body);
251            if (facts.refusal) return reply({ ok: false, action: "post", refusal: facts.refusal });
252            return reply(
253              withWarnings(await postJob(env, agent, now, {
254                namespace: args.namespace,
255                title: args.title,
256                body: args.body,
257                priority: args.priority,
258                gate_required: args.gate_required,
259                required_scopes: args.required_flags ? { flags: args.required_flags } : undefined,
260                min_record: args.min_record,
261                review_required: args.review_required,
262                kind: args.kind,
263              }), args.body, facts.warnings)
264            );
265          }
266          case "claim":
267            return reply(await claimJob(env, agent, now, { namespace: args.namespace, id: args.id }));
268          case "complete": {
269            if (!args.id) return fail("complete needs the job id.");
270            // A string that does not parse is refused rather than ignored, so a row
271            // never silently says nothing happened.
272            const parsed = parseEvidence(args.evidence);
273            if ("error" in parsed) return fail(parsed.error);
274            const parsedEvidence = parsed.evidence;
275            return reply(
276              await completeAsCaller(env, agent, now, args.id, {
277                result_summary: args.result_summary ?? "",
278                result_ref: args.result_ref,
279                evidence: parsedEvidence,
280                claim: args.claim,
281                // As received, so job_claims.raw keeps the evidence the driver sent and
282                // not parseEvidence's reading of it.
283                raw: { evidence: args.evidence, claim: args.claim, result_summary: args.result_summary, result_ref: args.result_ref },
284              })
285            );
286          }
287          case "record": {
288            if (!args.id) return fail("record needs the job id.");
289            const parsed = parseEvidence(args.evidence);
290            if ("error" in parsed) return fail(parsed.error);
291            return reply(
292              await recordAsCaller(env, agent, now, args.id, {
293                result_summary: args.result_summary ?? "",
294                result_ref: args.result_ref,
295                evidence: parsed.evidence,
296                claim: args.claim,
297                raw: { evidence: args.evidence, claim: args.claim, result_summary: args.result_summary, result_ref: args.result_ref },
298              })
299            );
300          }
301          case "fail": {
302            if (!args.id) return fail("fail needs the job id.");
303            return reply(
304              await failAsCaller(env, agent, now, args.id, args.reason ?? "", {
305                claim: args.claim,
306                raw: { claim: args.claim, reason: args.reason },
307              })
308            );
309          }
310          case "fire": {
311            if (!args.id) return fail("fire needs the job id.");
312            return reply(await fireRoutine(env, agent, now, args.id));
313          }
314          case "routine": {
315            return reply(await setRoutine(env, agent, args.namespace, args.routine_url));
316          }
317          case "begin": {
318            if (!args.id) return fail("begin needs the job id.");
319            return reply(await beginJob(env, agent, now, args.id));
320          }
321          case "release": {
322            if (!args.id) return fail("release needs the job id.");
323            return reply(await releaseJob(env, agent, now, args.id, args.reason ?? ""));
324          }
325          case "block": {
326            if (!args.id) return fail("block needs the job id.");
327            return reply(
328              await blockJob(env, agent, now, args.id, {
329                reason: args.reason ?? "",
330                command: args.command,
331                question: args.question,
332                claim: args.claim,
333                raw: { claim: args.claim, reason: args.reason, command: args.command },
334              })
335            );
336          }
337          case "resume": {
338            if (!args.id) return fail("resume needs the job id.");
339            return reply(await resumeJob(env, agent, now, args.id, args.reason ?? "", {
340                approvedByPolicy: args.approved_by_policy,
341                take: args.take,
342                correction: args.correction,
343                note: args.note,
344              }));
345          }
346          case "edit": {
347            if (!args.id) return fail("edit needs the job id.");
348            return reply(
349              withWarnings(await editJob(env, agent, now, args.id, {
350                title: args.title,
351                body: args.body,
352                priority: args.priority,
353                gate_required: args.gate_required,
354              }), args.body)
355            );
356          }
357          case "park": {
358            if (!args.id) return fail("park needs the job id.");
359            return reply(await parkJob(env, agent, now, args.id, args.reason));
360          }
361          case "unpark": {
362            if (!args.id) return fail("unpark needs the job id.");
363            return reply(await unparkJob(env, agent, now, args.id));
364          }
365          case "supersede": {
366            if (!args.id) return fail("supersede needs the job id.");
367            return reply(await supersedeJob(env, agent, now, args.id, { reason: args.reason ?? "", replaced_by: args.replaced_by }));
368          }
369        }
370        return fail(`unknown jobs action '${String(args.action as unknown as string)}'.`);
371      } catch (err) {
372        return fail(err instanceof Error ? err.message : String(err));
373      }
374    }
375  );
376}