src/routine-fire.ts

Functions

Imports

It imports src/agents.ts, src/env.ts, src/jobs-mirror.ts, src/jobs-schema.ts, src/jobs-transition.ts, src/log.ts, src/redact.ts, src/signed-docs.ts, src/store-guards.ts, src/time.ts. Imported by src/tools/jobs.ts.

Source

1import type { Env } from "./env";
2import type { Agent } from "./agents";
3import { jobDocPath, outsideJobNamespace, startedNotice, type JobRow } from "./jobs-schema";
4import { jobAudit, mirrorStatements } from "./jobs-mirror";
5import { callerIsSeat, guardedTransition, readJob, refuse, type JobResult } from "./jobs-transition";
6import { verifySignedBody } from "./signed-docs";
7import { auditStatement } from "./store-guards";
8import { logEvent } from "./log";
9import { redactText } from "./redact";
10import { MINUTE_MS } from "./time";
11
12// CAPSID STARTS A SESSION ITSELF (job_a7d6f6009242, docs/routines.md). One Claude Code
13// routine serves every namespace, created by Dustin at claude.ai/code/routines on Opus,
14// with only the Capsid connection through the network secret. Its API trigger is a
15// per-routine URL and bearer token. The seat fires it for one queued job, and the
16// routine's prompt reads the job id from the routine-fire-payload block the fire text
17// arrives in. Nothing here claims the job: the session works it without a claim, and the
18// fire marks it started (the label begin sets, src/jobs-begin.ts) so the next session and
19// the next fire are told somebody is on it.
20//
21// Where each piece lives: the fire URL in APP_KV (`routine:<namespace>:url`), set by the
22// seat with capsid_work_jobs action "routine", because it is not a secret and a new
23// routine should not need a deploy; the token as a Worker secret, ROUTINE_<NS>_TOKEN,
24// set by Dustin with `npx wrangler secret put`, read by name and never returned.
25//
26// One routine for every namespace (job_77b45a741ddc): the seat sets the same fire URL on
27// each namespace, and Dustin sets one token. A namespace's own ROUTINE_<NS>_TOKEN wins;
28// otherwise the token falls back to the secret of any namespace whose stored fire URL
29// names the same routine id. A different routine id never borrows a token, and only the
30// secret's name is ever reported or audited, never its value.
31
32export const routineUrlKey = (namespace: string) => `routine:${namespace}:url`;
33
34/** The Worker secret holding a namespace's routine token: ROUTINE_<NS>_TOKEN, dashes as underscores. */
35export const routineTokenName = (namespace: string) => `ROUTINE_${namespace.toUpperCase().replace(/-/g, "_")}_TOKEN`;
36
37function routineToken(env: Env, namespace: string): string | null {
38  const value = (env as unknown as Record<string, unknown>)[routineTokenName(namespace)];
39  return typeof value === "string" && value.length > 0 ? value : null;
40}
41
42// The token a fire for this namespace sends, and the name of the secret it came from. The
43// namespace's own secret wins. Otherwise, when the namespace's fire URL names a routine,
44// the secret of another namespace whose stored fire URL names the same routine is used,
45// in namespace order so the answer is deterministic. A token is lent only from a
46// namespace the caller may act in, so a seat held to one namespace never fires with
47// another's credential. A different routine id never borrows.
48async function resolveRoutineToken(env: Env, agent: Agent, namespace: string, routineId: string | null): Promise<{ secret: string; value: string } | null> {
49  const own = routineToken(env, namespace);
50  if (own !== null) return { secret: routineTokenName(namespace), value: own };
51  if (routineId === null) return null;
52  const rows = (await env.DB.prepare("SELECT namespace FROM namespaces ORDER BY namespace").all<{ namespace: string }>()).results ?? [];
53  for (const { namespace: other } of rows) {
54    if (other === namespace || !ROUTINE_NAMESPACE.test(other) || outsideJobNamespace(agent, other)) continue;
55    const value = routineToken(env, other);
56    if (value === null) continue;
57    const url = await env.APP_KV.get(routineUrlKey(other));
58    if (url && routineIdOf(url) === routineId) return { secret: routineTokenName(other), value };
59  }
60  return null;
61}
62
63const namespaceRefusal = (action: string, namespace: string) =>
64  refuse(action, `${namespace} cannot have a routine: a routine's namespace is lowercase letters and digits joined by single dashes, so its token secret names it alone.`);
65
66// The one shape a fire URL takes (platform.claude.com/docs/en/api/claude-code/routines-fire,
67// read 2026-10-11). Anything else is refused, so the token is only ever sent to Anthropic.
68const FIRE_URL = /^https:\/\/api\.anthropic\.com\/v1\/claude_code\/routines\/(trig_[A-Za-z0-9]{8,64})\/fire$/;
69
70// The headers the fire endpoint takes. The beta header is still accepted and named in the
71// job, so it is sent; anthropic-version 2023-06-01 is the only value the endpoint accepts.
72const ANTHROPIC_VERSION = "2023-06-01";
73const ROUTINE_BETA = "experimental-cc-routine-2026-04-01";
74
75// A second fire for the same job inside this window is refused: the endpoint has no
76// idempotency key, so each fire is a new session on the subscription.
77export const REFIRE_MINUTES = 30;
78// A fire reserves itself before it posts, so two fires racing for one job cannot both
79// post. The reservation outlives the request's timeout and no longer.
80const RESERVE_MINUTES = 2;
81const FIRE_TIMEOUT_MS = 15_000;
82
83// Namespaces a routine may serve: lowercase letters and digits in dash-separated runs,
84// so ROUTINE_<NS>_TOKEN, which turns each dash into an underscore, names one namespace.
85const ROUTINE_NAMESPACE = /^[a-z0-9]+(-[a-z0-9]+)*$/;
86
87// What a fire's answer may name (Anthropic documents {"type":"routine_fire",
88// "claude_code_session_id":"session_01...","claude_code_session_url":"https://claude.ai/code/session_01..."}).
89// The id's charset is wider than the documented example, because the real answer failed
90// the strict one. Whatever is stored or returned matches one of these two shapes.
91const SESSION_ID = /^session_[A-Za-z0-9_-]{8,128}$/;
92const SESSION_URL = /^https:\/\/(?:[a-z0-9-]+\.)*claude\.ai(?:\/[A-Za-z0-9_.-]+)*\/session_[A-Za-z0-9_-]{8,128}\/?$/;
93const SESSION_SEGMENT = /^session_[A-Za-z0-9_-]{8,128}$/;
94const ANSWER_MAX_CHARS = 64 * 1024;
95const DIAG_FIELD_CHARS = 200;
96const DIAG_MAX_KEYS = 20;
97const DIAG_MAX_STRINGS = 10;
98
99/** A session URL as the answer names it, reduced to origin and path, or null when it is not a claude.ai session URL. */
100function readSessionUrl(raw: unknown): string | null {
101  if (typeof raw !== "string" || raw.length > 2048) return null;
102  let url: URL;
103  try {
104    url = new URL(raw);
105  } catch {
106    return null;
107  }
108  if (url.protocol !== "https:" || url.port !== "") return null;
109  if (url.hostname !== "claude.ai" && !url.hostname.endsWith(".claude.ai")) return null;
110  if (!url.pathname.split("/").some((segment) => SESSION_SEGMENT.test(segment))) return null;
111  const clean = `https://${url.hostname}${url.pathname}`;
112  return SESSION_URL.test(clean) ? clean : null;
113}
114
115/** A session id as the answer names it, or null. */
116function readSessionId(raw: unknown): string | null {
117  return typeof raw === "string" && SESSION_ID.test(raw) ? raw : null;
118}
119
120/**
121 * The session an answer names. The id and the URL each stand alone: a good id with no
122 * readable URL gets the URL built from it, and a good URL with no readable id gives its
123 * id back, so a session is recorded whenever either one reads.
124 */
125function readSession(answer: unknown): { id: string | null; url: string | null; mismatch: string | null } {
126  const body = answer !== null && typeof answer === "object" && !Array.isArray(answer) ? (answer as Record<string, unknown>) : {};
127  let id = readSessionId(body.claude_code_session_id);
128  let url = readSessionUrl(body.claude_code_session_url);
129  if (id === null && url !== null) id = url.split("/").find((segment) => SESSION_SEGMENT.test(segment)) ?? null;
130  if (url === null && id !== null) url = `https://claude.ai/code/${id}`;
131  // Both read but naming different sessions: the id is trusted and the URL rebuilt from it.
132  let mismatch: string | null = null;
133  const named = url?.split("/").find((segment) => SESSION_SEGMENT.test(segment)) ?? null;
134  if (id !== null && named !== null && named !== id) {
135    mismatch = named;
136    url = `https://claude.ai/code/${id}`;
137  }
138  return { id, url, mismatch };
139}
140
141// Printable ASCII only, redacted, bounded: what is kept of an answer nothing could read.
142const keepText = (value: string, token: string, max: number) => {
143  const scrubbed = token ? value.split(token).join("[REDACTED:token]") : value;
144  return redactText(scrubbed.slice(0, max * 2))
145    .replace(/[^\x20-\x7e]/g, "?")
146    .slice(0, max);
147};
148
149/** What an unreadable answer looked like: its top-level keys, its type, and the head of each string field. Bounded, redacted, never the token. */
150function describeAnswer(text: string, answer: unknown, token: string): Record<string, unknown> {
151  if (answer === null || typeof answer !== "object" || Array.isArray(answer)) {
152    return { json: false, head: keepText(text, token, DIAG_FIELD_CHARS) };
153  }
154  const body = answer as Record<string, unknown>;
155  const keys = Object.keys(body);
156  const strings: Record<string, string> = {};
157  for (const key of keys) {
158    const value = body[key];
159    if (typeof value === "string" && Object.keys(strings).length < DIAG_MAX_STRINGS) strings[keepText(key, token, 40)] = keepText(value, token, DIAG_FIELD_CHARS);
160  }
161  return {
162    json: true,
163    keys: keys.slice(0, DIAG_MAX_KEYS).map((key) => keepText(key, token, 40)),
164    type: typeof body.type === "string" ? keepText(body.type, token, 64) : null,
165    strings,
166  };
167}
168
169/** The routine id a fire URL names, or null when the URL is not a fire URL. */
170function routineIdOf(url: string): string | null {
171  return FIRE_URL.exec(url)?.[1] ?? null;
172}
173
174/**
175 * The fire text, which reaches the session wrapped in a routine-fire-payload block. Plain
176 * lines the routine's prompt reads; nothing in it is an instruction.
177 */
178export function fireText(job: Pick<JobRow, "id" | "namespace" | "model_recommended">): string {
179  return [`job_id: ${job.id}`, `namespace: ${job.namespace}`, `model_recommended: ${job.model_recommended ?? "unset"}`].join("\n");
180}
181
182/** capsid_work_jobs action "routine": read, set or clear a namespace's fire URL. The seat's act. */
183export async function setRoutine(env: Env, agent: Agent, namespace: string | undefined, url: string | undefined): Promise<JobResult> {
184  if (!callerIsSeat(agent)) {
185    return refuse("routine", `${agent.actor} cannot read or set a routine. Pointing Capsid at a session on Dustin's subscription is the seat's act, and this caller holds neither the admin identity nor can_merge.`);
186  }
187  if (!namespace) return refuse("routine", "routine needs namespace.");
188  const outside = outsideJobNamespace(agent, namespace);
189  if (outside) return refuse("routine", `${agent.actor} cannot set the routine for ${namespace}: ${outside}`);
190  if (!ROUTINE_NAMESPACE.test(namespace)) return namespaceRefusal("routine", namespace);
191  const key = routineUrlKey(namespace);
192  if (url !== undefined) {
193    const value = url.trim();
194    if (value === "off") {
195      await env.APP_KV.delete(key);
196    } else {
197      if (!routineIdOf(value)) {
198        return refuse("routine", `routine_url must be a fire URL, https://api.anthropic.com/v1/claude_code/routines/trig_.../fire, as the routine's API trigger shows it; or "off" to clear it. Nothing was changed.`);
199      }
200      await env.APP_KV.put(key, value);
201    }
202    await env.DB.batch([auditStatement(env.DB, agent.actor, "routine-set", namespace, null, { namespace, routine_id: value === "off" ? null : routineIdOf(value) })]);
203  }
204  const stored = await env.APP_KV.get(key);
205  const routineId = stored ? routineIdOf(stored) : null;
206  const token = await resolveRoutineToken(env, agent, namespace, routineId);
207  const tokenSecret = token?.secret ?? routineTokenName(namespace);
208  return {
209    ok: true,
210    action: "routine",
211    routine: { namespace, routine_id: routineId, token_secret: tokenSecret, token_set: token !== null },
212    note: token
213      ? undefined
214      : `No token is set for ${namespace}. Dustin sets it with: npx wrangler secret put ${tokenSecret}, or the seat sets the same routine's fire URL on a namespace whose token is set.`,
215  };
216}
217
218// One statement, so two fires for one job cannot both pass the check: a reservation is
219// written only while no fire of that job is recorded inside REFIRE_MINUTES and no
220// reservation inside RESERVE_MINUTES. Returns the reservation's id, or null when refused.
221async function reserveFire(db: D1Database, actor: string, job: JobRow, routineId: string, now: Date): Promise<number | null> {
222  const row = await db
223    .prepare(
224      `INSERT INTO audit_log (actor, action, namespace, path, params)
225       SELECT ?1, 'job-routine-firing', ?2, ?3, ?4
226       WHERE NOT EXISTS (
227         SELECT 1 FROM audit_log WHERE json_extract(params, '$.job_id') = ?5
228           AND ((action = 'job-routine-fired' AND at >= datetime(?6)) OR (action = 'job-routine-firing' AND at >= datetime(?7))))
229       RETURNING id`
230    )
231    .bind(
232      actor,
233      job.namespace,
234      jobDocPath(job.id),
235      JSON.stringify({ job_id: job.id, routine_id: routineId }),
236      job.id,
237      new Date(now.getTime() - REFIRE_MINUTES * MINUTE_MS).toISOString(),
238      new Date(now.getTime() - RESERVE_MINUTES * MINUTE_MS).toISOString()
239    )
240    .first<{ id: number }>();
241  return row?.id ?? null;
242}
243
244async function lastFire(db: D1Database, jobId: string): Promise<{ at: string; session_url: string | null } | null> {
245  const row = await db
246    .prepare(
247      `SELECT at, json_extract(params, '$.session_url') AS session_url FROM audit_log
248       WHERE action IN ('job-routine-fired', 'job-routine-firing') AND json_extract(params, '$.job_id') = ?1
249       ORDER BY id DESC LIMIT 1`
250    )
251    .bind(jobId)
252    .first<{ at: string; session_url: string | null }>();
253  return row ?? null;
254}
255
256/** capsid_work_jobs action "fire": start a session for one queued job by firing its namespace's routine. */
257export async function fireRoutine(env: Env, agent: Agent, now: Date, id: string): Promise<JobResult> {
258  if (!callerIsSeat(agent)) {
259    return refuse("fire", `${agent.actor} cannot fire a routine. Starting work on Dustin's subscription is the seat's act, and this caller holds neither the admin identity nor can_merge.`);
260  }
261  const job: JobRow | null = await readJob(env.DB, id);
262  if (!job) return refuse("fire", `no job ${id}.`);
263  const outside = outsideJobNamespace(agent, job.namespace);
264  if (outside) return refuse("fire", `${agent.actor} cannot fire for ${id}: ${outside}`);
265  if (!ROUTINE_NAMESPACE.test(job.namespace)) return namespaceRefusal("fire", job.namespace);
266  if (job.status !== "queued") {
267    return refuse("fire", `${id} is ${job.status}, not queued. A routine is fired for a queued job only.`);
268  }
269  const begun = startedNotice(job);
270  if (begun) return refuse("fire", `already_started: ${begun.note} No session was fired.`);
271  const verdict = await verifySignedBody(env.IMPROVE_SCORE_SECRET, job.body, "job body");
272  if (!verdict.ok) return refuse("fire", `${id} does not verify, so no session is started for it: ${verdict.reason}`);
273
274  const url = await env.APP_KV.get(routineUrlKey(job.namespace));
275  const routineId = url ? routineIdOf(url) : null;
276  if (!url || !routineId) {
277    return refuse("fire", `routine_missing: ${job.namespace} has no routine. Create it at claude.ai/code/routines (docs/routines.md), then the seat sets its fire URL with capsid_work_jobs action "routine".`);
278  }
279  const resolved = await resolveRoutineToken(env, agent, job.namespace, routineId);
280  if (resolved === null) {
281    const tokenSecret = routineTokenName(job.namespace);
282    return refuse("fire", `routine_token_missing: ${tokenSecret} is not set, and no namespace sharing routine ${routineId} has a token either. The seat sets the same fire URL on a namespace whose token is set (docs/routines.md), or Dustin generates the token on the routine's API trigger and sets it with: npx wrangler secret put ${tokenSecret}`);
283  }
284
285  const { secret: tokenSecret, value: token } = resolved;
286  if ((await reserveFire(env.DB, agent.actor, job, routineId, now)) === null) {
287    const earlier = await lastFire(env.DB, job.id);
288    return refuse("fire", `already_fired: ${id} was fired at ${earlier?.at ?? "a moment ago"}${earlier?.session_url ? ` (${earlier.session_url})` : ""}. Each fire is a new session, so a second one inside ${REFIRE_MINUTES} minutes is refused.`);
289  }
290
291  let res: Response;
292  try {
293    res = await fetch(url, {
294      method: "POST",
295      headers: {
296        Authorization: `Bearer ${token}`,
297        "anthropic-version": ANTHROPIC_VERSION,
298        "anthropic-beta": ROUTINE_BETA,
299        "Content-Type": "application/json",
300      },
301      body: JSON.stringify({ text: fireText(job) }),
302      signal: AbortSignal.timeout(FIRE_TIMEOUT_MS),
303    });
304  } catch (err) {
305    // No answer: the session may or may not exist, so the fire is recorded as one and a
306    // second fire waits out the window.
307    const message = err instanceof Error ? err.message : String(err);
308    await env.DB.batch([jobAudit(env.DB, agent.actor, "job-routine-fired", job, { routine_id: routineId, session_id: null, session_url: null, outcome: "unknown", error: message.slice(0, 200) })]);
309    return refuse("fire", `routine_unreachable: the fire request for ${routineId} got no answer (${message}). A session may have started; look at claude.ai/code before firing again after ${REFIRE_MINUTES} minutes.`);
310  }
311
312  if (res.status === 401) {
313    return refuse("fire", `routine_unauthorized: Anthropic refused ${tokenSecret} for ${routineId} (401). The token was revoked or regenerated; Dustin generates a new one on the routine's API trigger and sets it with npx wrangler secret put ${tokenSecret}.`);
314  }
315  if (res.status === 429) {
316    const retry = res.headers.get("Retry-After");
317    return refuse("fire", `routine_rate_limited: the hourly fire limit for ${routineId} or the account is reached (429; 30 per routine, 100 per account).${retry ? ` Retry after ${retry} seconds.` : ""}`);
318  }
319  if (res.status === 404) {
320    return refuse("fire", `routine_not_found: Anthropic has no routine ${routineId} (404). It was deleted; create it again and set its fire URL with action "routine".`);
321  }
322  if (res.status < 200 || res.status > 299) {
323    const detail = await errorType(res);
324    return refuse("fire", `routine_refused: Anthropic answered ${res.status}${detail ? ` (${detail})` : ""} for ${routineId}. A 400 is often a paused routine. No session was started.`);
325  }
326
327  // A 2xx means a session exists. Record it even when the answer is unreadable, so a
328  // second fire is still refused and the session is not lost.
329  const answerText = (await res.text().catch(() => "")).slice(0, ANSWER_MAX_CHARS);
330  let answer: unknown = null;
331  try {
332    answer = JSON.parse(answerText);
333  } catch {
334    answer = null;
335  }
336  const { id: sessionId, url: sessionUrl, mismatch } = readSession(answer);
337  // Nothing readable: keep what the answer looked like, so the next failure can be diagnosed.
338  const unreadable = sessionId === null && sessionUrl === null ? { answer: describeAnswer(answerText, answer, token) } : {};
339  let unrecorded = "";
340  try {
341    await env.DB.batch([jobAudit(env.DB, agent.actor, "job-routine-fired", job, { routine_id: routineId, session_id: sessionId, session_url: sessionUrl, token_secret: tokenSecret, ...(mismatch ? { session_url_mismatch: mismatch } : {}), ...unreadable })]);
342  } catch (err) {
343    // The session exists whatever D1 says, so the answer still returns it; the reservation
344    // row holds off a second fire for RESERVE_MINUTES.
345    const message = err instanceof Error ? err.message : String(err);
346    logEvent("error", "ROUTINE_FIRE_UNRECORDED", { message: `ROUTINE_FIRE_UNRECORDED: ${job.id} ${sessionId ?? "(no session id)"}: ${message}` });
347    unrecorded = ` The fire could not be recorded (${message}); keep this session URL.`;
348  }
349  const marked = await markStarted(env, agent, job, now, sessionUrl);
350  return {
351    ok: true,
352    action: "fire",
353    job: marked.job,
354    started: startedNotice(marked.job) ?? undefined,
355    session: { id: sessionId, url: sessionUrl },
356    note:
357      (sessionUrl
358        ? `${routineId} started ${sessionUrl} for ${job.id}. The job stays queued: the session works it without a claim and completes it naming its pull request.`
359        : `${routineId} answered ${res.status}, so a session started for ${job.id}, but its answer named no readable session id or URL. What the answer looked like is in the job-routine-fired audit row. Find the session at claude.ai/code.`) +
360      unrecorded +
361      marked.note,
362  };
363}
364
365// The started label, set as begin sets it and in one guarded batch with its mirror and
366// audit row. A job that moved or was begun since the read keeps what it has: the session
367// is already running, so that is said, never refused.
368async function markStarted(env: Env, agent: Agent, job: JobRow, now: Date, sessionUrl: string | null): Promise<{ job: JobRow; note: string }> {
369  const at = now.toISOString();
370  const started: JobRow = { ...job, started_at: at, started_by: agent.actor, updated_at: at };
371  const won = await guardedTransition(env, job, [
372    env.DB.prepare(
373      `UPDATE jobs SET started_at = ?2, started_by = ?3, updated_at = ?2 WHERE id = ?1 AND status = 'queued' AND started_at IS NULL RETURNING id`
374    ).bind(job.id, at, agent.actor),
375    ...(await mirrorStatements(env, started, "job-started", agent.actor)),
376    jobAudit(env.DB, agent.actor, "job-started", started, { started_at: at, session_url: sessionUrl }),
377  ]);
378  if (won) return { job: started, note: "" };
379  const moved = (await readJob(env.DB, job.id)) ?? job;
380  return { job: moved, note: ` The job was not marked started: it changed while the routine was fired (now ${moved.status}${moved.started_by ? `, started by ${moved.started_by}` : ""}).` };
381}
382
383async function errorType(res: Response): Promise<string | null> {
384  const body = (await res.json().catch(() => null)) as { error?: { type?: unknown } } | null;
385  const type = body?.error?.type;
386  return typeof type === "string" && /^[a-z_]{1,64}$/.test(type) ? type : null;
387}