src/jobs-mirror.ts

Functions

Imports

It imports src/doc-statements.ts, src/env.ts, src/job-signing.ts, src/jobs-schema.ts, src/store-guards.ts. Imported by 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-transition.ts, src/routine-fire.ts.

Source

1import type { Env } from "./env";
2import { docStatements, priorDoc } from "./doc-statements";
3import { checkJobText, type SignatureCheck } from "./job-signing";
4import { jobDocPath, isTerminalJobStatus, type JobRow } from "./jobs-schema";
5import { auditStatement } from "./store-guards";
6
7// The records a job transition writes beside its UPDATE: the mirror document, the
8// audit row, and the latest resume note read back from that audit row.
9
10// What the last resume approved, handed to whoever holds the job next.
11//
12// The reason a resume takes is written to the audit row, and no tool returns audit
13// params to a driver, so without this a driver picking a resumed job back up could not
14// read what the seat had approved and would have to ask again. The audit row stays the
15// one record; this reads it back.
16export interface ResumeNote {
17  reason: string;
18  // The seat's full note, when the resume carried one. `reason` is bounded at
19  // MAX_TITLE and holds one line; the rulings or plan the seat approved go here and
20  // are handed on whole, never truncated.
21  note?: string;
22  by: string;
23  at: string;
24  approved_by_policy?: string;
25  policy_class?: string;
26  correction?: true;
27  // Whether the approval is the one the resume wrote (src/job-signing.ts). A note whose
28  // signature does not match is withheld: its text is replaced and its full note
29  // dropped, so nothing changed after the fact reaches a driver as an approval.
30  signature: SignatureCheck;
31}
32
33// Who may read a note, and with what: the database and the signing secret.
34type NoteEnv = Pick<Env, "DB" | "IMPROVE_SCORE_SECRET">;
35
36/** The fields a resume note's signature covers, in one place for the writer and reader. */
37export function resumeNoteFields(approved: string, note: string | null, by: string): Record<string, string | null> {
38  return { approved, note, by };
39}
40
41const WITHHELD = "WITHHELD: this approval's signature does not match, so it was changed after the resume wrote it. Do not act on it; ask the seat.";
42
43// Served by audit_log_doc (namespace, path, id DESC), the index every job audit row
44// already falls under because it is written against the job's mirror path.
45type NoteRow = { actor: string | null; params: string | null; at: string };
46
47// One audit row as the note it recorded, its signature checked, or null when the row is
48// not a readable resume (no params, not JSON, no approved text).
49async function noteFromRow(env: NoteEnv, job: Pick<JobRow, "id">, row: NoteRow): Promise<ResumeNote | null> {
50  if (!row.params) return null;
51  let params: Record<string, unknown>;
52  try {
53    params = JSON.parse(row.params) as Record<string, unknown>;
54  } catch {
55    return null;
56  }
57  if (typeof params.approved !== "string") return null;
58  const by = row.actor ?? "(unknown)";
59  const note = typeof params.note === "string" ? params.note : null;
60  const signature = await checkJobText(
61    env.IMPROVE_SCORE_SECRET,
62    "resume-note",
63    job.id,
64    resumeNoteFields(params.approved, note, by),
65    typeof params.sig === "string" ? params.sig : null
66  );
67  if (signature === "mismatch") return { reason: WITHHELD, by, at: row.at, signature };
68  return {
69    reason: params.approved,
70    ...(note !== null ? { note } : {}),
71    by,
72    at: row.at,
73    signature,
74    ...(typeof params.approved_by_policy === "string" ? { approved_by_policy: params.approved_by_policy } : {}),
75    ...(typeof params.policy_class === "string" ? { policy_class: params.policy_class } : {}),
76    ...(params.correction === true ? { correction: true as const } : {}),
77  };
78}
79
80export async function latestResumeNote(
81  env: NoteEnv,
82  job: Pick<JobRow, "id" | "namespace" | "resumed_count">
83): Promise<ResumeNote | null> {
84  if (!job.resumed_count) return null;
85  const row = await env.DB
86    .prepare(
87      "SELECT actor, params, at FROM audit_log WHERE namespace = ?1 AND path = ?2 AND action = 'job-resumed' ORDER BY id DESC LIMIT 1"
88    )
89    .bind(job.namespace, jobDocPath(job.id))
90    .first<NoteRow>();
91  return row ? noteFromRow(env, job, row) : null;
92}
93
94// Every resume note a driver is handed, together, and no more than this many bytes of
95// them: a job resumed fifteen times (job_5765103c658f, 2026-09-26) carried its PR order
96// and rulings in the early notes, which "the latest only" lost.
97const RESUME_NOTES_MAX_BYTES = 64 * 1024;
98// Rows read for it, so one job with a very long history is one bounded read.
99const RESUME_NOTES_MAX_ROWS = 200;
100
101export interface ResumeNotes {
102  // Newest first. Each note's own signature is checked; one that does not match is
103  // withheld, as the newest one is.
104  notes: ResumeNote[];
105  // Resumes the job recorded that this answer does not carry: older ones past the byte
106  // cap or the row cap, or an audit row that cannot be read as a note.
107  dropped: number;
108}
109
110/** Every resume note for a job, newest first, at most RESUME_NOTES_MAX_BYTES of them (the
111 *  oldest dropped and counted). The newest is first and is what `latestResumeNote` returns. */
112export async function resumeNotes(env: NoteEnv, job: Pick<JobRow, "id" | "namespace" | "resumed_count">): Promise<ResumeNotes> {
113  if (!job.resumed_count) return { notes: [], dropped: 0 };
114  const { results } = await env.DB
115    .prepare(
116      "SELECT actor, params, at FROM audit_log WHERE namespace = ?1 AND path = ?2 AND action = 'job-resumed' ORDER BY id DESC LIMIT ?3"
117    )
118    .bind(job.namespace, jobDocPath(job.id), RESUME_NOTES_MAX_ROWS)
119    .all<NoteRow>();
120  const notes: ResumeNote[] = [];
121  let bytes = 0;
122  for (const row of results ?? []) {
123    const note = await noteFromRow(env, job, row);
124    if (!note) continue;
125    const size = new TextEncoder().encode(JSON.stringify(note)).length;
126    // The newest note is always returned, whatever its size: it is the approval a driver
127    // acts on, and a cap that hid it would be the loss this reads back to prevent.
128    if (notes.length > 0 && bytes + size > RESUME_NOTES_MAX_BYTES) break;
129    notes.push(note);
130    bytes += size;
131  }
132  return { notes, dropped: Math.max(0, job.resumed_count - notes.length) };
133}
134
135// The document a job mirrors to. The prompt is the SIGNED body, byte for byte, so a
136// driver that reads the document rather than the row still verifies the same bytes.
137function renderJobDoc(job: JobRow, notes: ResumeNote[], dropped = 0): string {
138  const note = notes[0] ?? null;
139  const lines = [
140    `# ${job.title}`,
141    "",
142    `Job \`${job.id}\` in \`${job.namespace}\`. This document is a PROJECTION of the jobs`,
143    "table, which is the source of truth for status. Rewritten on every transition.",
144    "",
145    `- status: **${job.status}**`,
146    `- priority: ${job.priority}`,
147    `- gate required: ${job.gate_required ? "yes" : "no"}`,
148    `- posted by: ${job.posted_by}`,
149    `- claimed by: ${job.claimed_by ?? "(unclaimed)"}`,
150  ];
151  // The started label (src/jobs-begin.ts), once somebody began the job.
152  if (job.started_by) lines.push(`- started by: ${job.started_by} at ${job.started_at ?? "(no time recorded)"}`);
153  // Only once it has happened, so a job that never hit a gate carries no line of zeroes.
154  if (job.blocked_count > 0) lines.push(`- gates hit: ${job.blocked_count}, resumed: ${job.resumed_count}`);
155  // The brief carries open job documents, so this line is how brief hands the
156  // approval to a driver.
157  if (note) lines.push(`- last resume, by ${note.by} at ${note.at}${note.signature === "verified" ? "" : ` (${note.signature})`}: ${note.reason}`);
158  if (job.result_summary) lines.push(`- result: ${job.result_summary}`);
159  if (job.result_ref) lines.push(`- result ref: ${job.result_ref}`);
160  // The full note as its own block after the status lines, untruncated: a driver
161  // reading the brief needs every ruling, not the first line of them.
162  if (note?.note) lines.push("", "## The last resume's note", "", note.note);
163  // Every earlier resume, newest first: an approval given at one resume is still the
164  // approval after the next (resumeNotes).
165  if (notes.length > 1) {
166    lines.push("", "## Earlier resumes", "");
167    for (const earlier of notes.slice(1)) {
168      lines.push(`- by ${earlier.by} at ${earlier.at}${earlier.signature === "verified" ? "" : ` (${earlier.signature})`}: ${earlier.reason}`);
169      if (earlier.note) lines.push("", earlier.note, "");
170    }
171  }
172  if (dropped > 0) lines.push("", `(${dropped} older resume${dropped === 1 ? "" : "s"} not shown.)`);
173  lines.push("", "## The prompt", "", job.body);
174  return lines.join("\n");
175}
176
177// `note` is passed by resume, whose audit row is written in the same batch as this
178// mirror and so cannot be read back yet. Every other transition reads it.
179export async function mirrorStatements(env: NoteEnv, job: JobRow, action: string, actor: string, note?: ResumeNote) {
180  const path = jobDocPath(job.id);
181  const prior = await priorDoc(env.DB, job.namespace, path);
182  // The passed note is the resume being written in this batch, so it is not in the audit
183  // log yet: it leads, and the notes read back are the earlier ones.
184  const history = await resumeNotes(env, job);
185  const notes = note ? [note, ...history.notes] : history.notes;
186  return docStatements(env.DB, {
187    namespace: job.namespace,
188    path,
189    title: `Job: ${job.title}`,
190    body: renderJobDoc(job, notes, history.dropped),
191    type: "task",
192    // Closed on a finished row, and `failed` is as finished as `done`: a failed job's
193    // mirror left active would keep brief carrying it as open work.
194    // isTerminalJobStatus is the one list of finished statuses.
195    status: isTerminalJobStatus(job.status) ? "closed" : "active",
196    tags: "jobs",
197    prior,
198    action,
199    actor,
200  });
201}
202
203// A job's audit row is addressed to its mirror document and carries the job id.
204export function jobAudit(db: D1Database, actor: string, action: string, job: JobRow, params: Record<string, unknown>) {
205  return auditStatement(db, actor, action, job.namespace, jobDocPath(job.id), { job_id: job.id, ...params });
206}