Skip to the content

Capsid menu

src/jobs.ts

116 lines and 5 functions in DrDustinEdwards/capsid, at commit daa875b on 2026-10-24. This file on GitHub.

Functions

The functions of jobs.ts
FunctionLinesComplexityCoveredCRAPMutation
jobsSummary62 to 1208100%WatchCRAP 8No data: mutation testing has not run on this repo yet
<anonymous:71>71 to 711100%OkCRAP 1No data: mutation testing has not run on this repo yet
<anonymous:90>90 to 902100%OkCRAP 2No data: mutation testing has not run on this repo yet
<anonymous:104>104 to 112322%WatchCRAP 7.2No data: mutation testing has not run on this repo yet
<anonymous:113>113 to 118133%OkCRAP 1.3No data: mutation testing has not run on this repo yet

Imports

It imports src/job-signing.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-transition.ts. Imported by src/skills-refresh.ts, src/status.ts, src/tools/jobs.ts, src/watcher.ts.

Source

import { isQuestionSummary } from "./jobs-holder";
import { checkJobText, type SignatureCheck } from "./job-signing";

// The work queue. The seat posts a job from a chat; a driver session on a machine
// claims it, does it, and reports back.
//
// Every transition is `UPDATE ... WHERE status = <expected> RETURNING id`, never
// meta.changes. D1's meta.changes is inflated by the FTS5 triggers on documents, and
// these batches carry a document write, so the count would be of the triggers as
// much as the row. No row back means somebody else got there first. The holder
// transitions (complete, fail, block) put the UPDATE in the same batch as
// its records, behind requireJobUnchanged (src/store-guards.ts), which aborts the
// whole batch when the row is not in the state the caller read.
//
// The row is the source of truth for status. The mirror document at
// <namespace>/jobs/<id>.md is rewritten in the same batch as every transition, so a
// reader who found the job through capsid_start_session or capsid_find_docs sees the state the table holds.
// It is a projection, and its body says so.
//
// The queue is split by who acts. jobs-claim.ts posts, lists and claims;
// jobs-holder.ts holds the transitions the claiming driver makes; jobs-seat.ts holds
// the ones made on a job the caller does not hold (admin fail, supersede, resume);
// jobs-edit.ts holds the seat's edit of a queued or blocked job, and jobs-park.ts its
// parking of a queued one; jobs-begin.ts holds the started label any writer may set;
// jobs-mirror.ts writes the mirror document and audit row; jobs-transition.ts holds
// what they share. This module is the one other modules import from, and it keeps the capsid_run_improve
// action status summary.

export { claimJob, listJobs, postJob } from "./jobs-claim";
export { blockJob, commandFromSummary, completeJob, failJob, RESUME_MARKER } from "./jobs-holder";
export { adminFailJob, completeAsCaller, failAsCaller, recordAsCaller, releaseJob, resumeJob, supersedeJob } from "./jobs-seat";
export { editJob, jobVersions } from "./jobs-edit";
export { parkJob, unparkJob } from "./jobs-park";
export { beginJob } from "./jobs-begin";
export type { JobResult } from "./jobs-transition";

// The capsid_run_improve action status block. Counted per namespace, plus what a human has to look at:
// the blocked jobs, with the command each is waiting on. Blocked is the only status
// whose rows come back rather than a count, because a count of blocked jobs tells
// nobody what to run, and the Portal shows exactly these.
//
// done_today rather than done: a lifetime total only goes up and stops being
// information. What the seat wants to know is whether the queue moved today.
export interface JobsSummary {
  queued: number;
  claimed: number;
  blocked: number;
  done_today: number;
  // How often each job has hit a gate and how often a human sent it back. A job on
  // its third gate reads differently from one stuck at the same gate since it was
  // posted, and the count is what tells them apart.
  // command_signature says whether waiting_on is what the holder's block wrote
  // (src/job-signing.ts); a "mismatch" summary is withheld.
  // question is true for a block that asks something rather than waits on a command run
  // (capsid_work_jobs action block with question: true); the seat answers it with a resume note.
  blocked_jobs: Array<{ id: string; title: string; waiting_on: string | null; question: boolean; command_signature: SignatureCheck | null; blocked_times: number; resumed: number }>;
  // The claimed jobs and who holds each, so the seat can see a claim whose holder is
  // gone and release it.
  claimed_jobs: Array<{ id: string; title: string; held_by: string | null; claimed_at: string | null }>;
}

export async function jobsSummary(db: D1Database, namespace: string, now: Date, secret?: string): Promise<JobsSummary> {
  const day = now.toISOString().slice(0, 10);
  const counts = await db
    .prepare(
      `SELECT status, COUNT(*) AS n FROM jobs
       WHERE namespace = ?1 AND status IN ('queued', 'claimed', 'blocked') GROUP BY status`
    )
    .bind(namespace)
    .all<{ status: string; n: number }>();
  const byStatus = new Map((counts.results ?? []).map((r) => [r.status, r.n]));
  // substr, not a range: updated_at is an ISO string here and datetime('now') from the
  // table default, and the two agree only on the first ten characters.
  const doneToday = await db
    .prepare(
      `SELECT COUNT(*) AS n FROM jobs
       WHERE namespace = ?1 AND status = 'done' AND substr(updated_at, 1, 10) = ?2`
    )
    .bind(namespace, day)
    .first<{ n: number }>();
  const blocked = await db
    .prepare(
      `SELECT id, title, result_summary, summary_sig, blocked_count, resumed_count FROM jobs
       WHERE namespace = ?1 AND status = 'blocked' ORDER BY updated_at DESC LIMIT 20`
    )
    .bind(namespace)
    .all<{ id: string; title: string; result_summary: string | null; summary_sig: string | null; blocked_count: number; resumed_count: number }>();
  const blockedRows = blocked.results ?? [];
  const checks = await Promise.all(
    blockedRows.map((r) => (r.result_summary === null ? Promise.resolve(null) : checkJobText(secret, "summary", r.id, { summary: r.result_summary }, r.summary_sig)))
  );
  const claimed = await db
    .prepare(
      `SELECT id, title, claimed_by, claimed_at FROM jobs
       WHERE namespace = ?1 AND status = 'claimed' ORDER BY updated_at DESC LIMIT 20`
    )
    .bind(namespace)
    .all<{ id: string; title: string; claimed_by: string | null; claimed_at: string | null }>();
  return {
    queued: byStatus.get("queued") ?? 0,
    claimed: byStatus.get("claimed") ?? 0,
    blocked: byStatus.get("blocked") ?? 0,
    done_today: doneToday?.n ?? 0,
    blocked_jobs: blockedRows.map((r, i) => ({
      id: r.id,
      title: r.title,
      waiting_on: checks[i] === "mismatch" ? "WITHHELD: this summary's signature does not match, so it was changed after the block wrote it. Do not run its command." : r.result_summary,
      question: checks[i] !== "mismatch" && isQuestionSummary(r.result_summary),
      command_signature: checks[i],
      blocked_times: r.blocked_count,
      resumed: r.resumed_count,
    })),
    claimed_jobs: (claimed.results ?? []).map((r) => ({
      id: r.id,
      title: r.title,
      held_by: r.claimed_by,
      claimed_at: r.claimed_at,
    })),
  };
}