src/stale-jobs.ts

Functions

Imports

It imports src/env.ts, src/jobs-seat.ts, src/manual-steps.ts, src/ops-types/views.ts, src/time.ts. Imported by src/inbox.ts, src/merge-resume.ts, src/portal-stale.ts, src/tools/jobs.ts, src/whats-next.ts.

Source

1import type { Env } from "./env";
2import { MERGE_RESUME_ACTOR } from "./jobs-seat";
3import type { OpsStaleJob, StaleRule } from "./ops-types/views";
4import { readStepCache, stepsSeen, type StaleStepEntry } from "./manual-steps";
5import { HOUR_MS, DAY_MS } from "./time";
6
7// The stale view (capsid/research/design-stale-jobs.md, D5): `capsid_work_jobs` action list with
8// stale: true. Four rules, each a fact the Worker already holds, so reading the view
9// costs no GitHub reads:
10//   unchanged: a blocked or claimed job whose row has not changed in 3 days;
11//   resumed-not-completed: a job the merge-resume step resumed 24 hours or more ago that
12//     is still blocked, claimed or queued;
13//   prs-settled: a blocked job whose named pull requests are all merged or closed, read
14//     from the cache the merge-resume step writes (src/merge-resume.ts);
15//   steps-seen: a blocked job whose every "check:" line the watcher's last pass saw done,
16//     read from the cache that pass writes (src/manual-steps.ts).
17// Nothing here closes or moves a job; the view only says why a job looks stuck.
18
19export const STALE_UNCHANGED_MS = 3 * DAY_MS;
20export const STALE_RESUMED_MS = DAY_MS;
21const STALE_ROWS_MAX = 200;
22
23// One KV value (APP_KV) holding, per blocked job, the states of the pull requests it
24// names as the merge-resume step last read them.
25export const STALE_PRS_KEY = "stale:prs";
26
27export type PrState = "merged" | "closed" | "open" | "unread";
28
29export interface StalePrEntry {
30  merged: number;
31  closed: number;
32  open: number;
33  unread: number;
34  // ISO time of the read.
35  at: string;
36}
37
38export type StalePrCache = Record<string, StalePrEntry>;
39
40// The row and its rules are the Portal's contract too (GET /portal/api/stale), so they
41// live in src/ops-types/.
42export type { StaleRule };
43type StaleRow = OpsStaleJob;
44
45/** The counts one job's pull request reads add up to. */
46export function tallyPrStates(states: readonly PrState[], at: string): StalePrEntry {
47  const entry: StalePrEntry = { merged: 0, closed: 0, open: 0, unread: 0, at };
48  for (const state of states) entry[state] += 1;
49  return entry;
50}
51
52function isCount(value: unknown): value is number {
53  return typeof value === "number" && Number.isInteger(value) && value >= 0;
54}
55
56function isEntry(value: unknown): value is StalePrEntry {
57  if (typeof value !== "object" || value === null || Array.isArray(value)) return false;
58  const v = value as Record<string, unknown>;
59  return isCount(v.merged) && isCount(v.closed) && isCount(v.open) && isCount(v.unread) && typeof v.at === "string";
60}
61
62/** The cache as stored, keeping only entries of the right shape. null (never written) is
63 *  an empty cache; anything else that is not an object is reported as unreadable. */
64export function parseStaleCache(raw: unknown): { cache: StalePrCache; problem: string | null } {
65  if (raw === null || raw === undefined) return { cache: {}, problem: null };
66  if (typeof raw !== "object" || Array.isArray(raw)) return { cache: {}, problem: `${STALE_PRS_KEY} is not a JSON object` };
67  const cache: StalePrCache = {};
68  let dropped = 0;
69  for (const [id, entry] of Object.entries(raw as Record<string, unknown>)) {
70    if (isEntry(entry)) cache[id] = entry;
71    else dropped++;
72  }
73  return { cache, problem: dropped > 0 ? `${STALE_PRS_KEY} held ${dropped} entr${dropped === 1 ? "y" : "ies"} of the wrong shape, left out` : null };
74}
75
76/** The stored cache. A value that is not JSON is reported, not thrown, so the view
77 *  still serves the two rules that do not need it. */
78export async function readStaleCache(kv: KVNamespace): Promise<{ cache: StalePrCache; problem: string | null }> {
79  const text = await kv.get(STALE_PRS_KEY);
80  if (text === null) return { cache: {}, problem: null };
81  let raw: unknown;
82  try {
83    raw = JSON.parse(text);
84  } catch {
85    return { cache: {}, problem: `${STALE_PRS_KEY} is not JSON` };
86  }
87  return parseStaleCache(raw);
88}
89
90/** The next cache: this tick's reads replace each job's entry, and only jobs still
91 *  blocked keep one. */
92export function nextStaleCache(prior: StalePrCache, reads: StalePrCache, stillBlocked: ReadonlySet<string>): StalePrCache {
93  const next: StalePrCache = {};
94  for (const [id, entry] of Object.entries({ ...prior, ...reads })) {
95    if (stillBlocked.has(id)) next[id] = entry;
96  }
97  return next;
98}
99
100/** Rule C's test on one cache entry: nothing open, nothing unread, and at least one
101 *  pull request merged or closed. */
102export function prsSettled(entry: StalePrEntry | undefined): boolean {
103  return entry !== undefined && entry.open === 0 && entry.unread === 0 && entry.merged + entry.closed > 0;
104}
105
106function plural(n: number, word: string): string {
107  return `${n} ${word}${n === 1 ? "" : "s"}`;
108}
109
110/**
111 * Why one job is stale, or null. Pure, so the rules are tested without a database.
112 * The first rule that holds wins, the most actionable first: settled pull requests, then
113 * manual steps seen, then an auto-resume nobody finished, then a row nobody touched. A
114 * cache entry read before the job last changed says nothing about the job as it is now,
115 * so it is ignored.
116 */
117export function classifyStale(
118  job: { status: string; updated_at: string; auto_resumed_at: string | null },
119  entry: StalePrEntry | undefined,
120  now: Date,
121  steps?: StaleStepEntry
122): { rule: StaleRule; reason: string } | null {
123  const nowMs = now.getTime();
124  if (job.status === "blocked" && entry && entry.at >= job.updated_at && prsSettled(entry)) {
125    return {
126      rule: "prs-settled",
127      reason: `blocked, and every pull request it names is settled (${entry.merged} merged, ${entry.closed} closed without merging, read ${entry.at})`,
128    };
129  }
130  if (job.status === "blocked" && steps && steps.at >= job.updated_at && stepsSeen(steps)) {
131    return { rule: "steps-seen", reason: `blocked, and every manual step it names is seen done (${plural(steps.seen, "step")}, read ${steps.at})` };
132  }
133  if (job.auto_resumed_at && ["blocked", "claimed", "queued"].includes(job.status)) {
134    const since = Date.parse(job.auto_resumed_at);
135    if (Number.isFinite(since) && nowMs - since >= STALE_RESUMED_MS) {
136      const hours = Math.floor((nowMs - since) / HOUR_MS);
137      return {
138        rule: "resumed-not-completed",
139        reason: `auto-resumed by ${MERGE_RESUME_ACTOR} at ${job.auto_resumed_at} and still ${job.status} ${plural(hours, "hour")} later`,
140      };
141    }
142  }
143  if (job.status === "blocked" || job.status === "claimed") {
144    const since = Date.parse(job.updated_at);
145    if (Number.isFinite(since) && nowMs - since >= STALE_UNCHANGED_MS) {
146      const days = Math.floor((nowMs - since) / DAY_MS);
147      return { rule: "unchanged", reason: `${job.status} and unchanged since ${job.updated_at} (${plural(days, "day")})` };
148    }
149  }
150  return null;
151}
152
153export interface StaleJobsResult {
154  rows: StaleRow[];
155  truncated: boolean;
156  note?: string;
157}
158
159/**
160 * The stale view. The caller's namespace scope is checked before this runs, as for
161 * every capsid_work_jobs action list (src/tools/jobs.ts); a caller scoped to one namespace must name it
162 * (the registrar's namespaceRefusal), and an omitted namespace means every namespace.
163 */
164export async function staleJobs(env: Env, now: Date, opts: { namespace?: string } = {}): Promise<StaleJobsResult> {
165  const read = await readStaleCache(env.APP_KV);
166  const steps = await readStepCache(env.APP_KV);
167  const settled = [
168    ...Object.keys(read.cache).filter((id) => prsSettled(read.cache[id])),
169    ...Object.keys(steps.cache).filter((id) => stepsSeen(steps.cache[id])),
170  ];
171  const unchangedBefore = new Date(now.getTime() - STALE_UNCHANGED_MS).toISOString();
172  const resumedBefore = new Date(now.getTime() - STALE_RESUMED_MS).toISOString();
173  // The SQL narrows to the candidates; classifyStale decides, so the rules live in one
174  // place. ISO strings with milliseconds compare in time order. The namespace is bound as
175  // NULL for "every namespace" rather than spliced in, so the statement has no hole and
176  // test-integration/query-plans.test.ts plans it as written.
177  const { results } = await env.DB.prepare(
178    `SELECT j.id, j.namespace, j.title, j.status, j.updated_at,
179            (SELECT MAX(t.at) FROM job_touches t WHERE t.job_id = j.id AND t.actor = ?1 AND t.kind = 'resume') AS auto_resumed_at
180       FROM jobs j
181      WHERE j.status IN ('blocked', 'claimed', 'queued')
182        AND (?5 IS NULL OR j.namespace = ?5)
183        AND ((j.status IN ('blocked', 'claimed') AND j.updated_at <= ?2)
184          OR EXISTS (SELECT 1 FROM job_touches r WHERE r.job_id = j.id AND r.actor = ?1 AND r.kind = 'resume' AND r.at <= ?3)
185          OR (j.status = 'blocked' AND j.id IN (SELECT value FROM json_each(?4))))
186      ORDER BY j.updated_at ASC
187      LIMIT ?6`
188  )
189    .bind(MERGE_RESUME_ACTOR, unchangedBefore, resumedBefore, JSON.stringify(settled), opts.namespace ?? null, STALE_ROWS_MAX + 1)
190    .all<{ id: string; namespace: string; title: string; status: string; updated_at: string; auto_resumed_at: string | null }>();
191  const candidates = results ?? [];
192  const truncated = candidates.length > STALE_ROWS_MAX;
193  const rows: StaleRow[] = [];
194  for (const job of truncated ? candidates.slice(0, STALE_ROWS_MAX) : candidates) {
195    const why = classifyStale(job, read.cache[job.id], now, steps.cache[job.id]);
196    if (why) rows.push({ id: job.id, namespace: job.namespace, title: job.title, status: job.status, updated_at: job.updated_at, ...why });
197  }
198  const notes = [
199    ...(truncated ? [`more than ${STALE_ROWS_MAX} stale jobs; narrow by namespace.`] : []),
200    ...(read.problem ? [`the pull request cache could not be fully read (${read.problem}), so prs-settled may be missing rows.`] : []),
201    ...(steps.problem ? [`the manual steps cache could not be fully read (${steps.problem}), so steps-seen may be missing rows.`] : []),
202  ];
203  return { rows, truncated, ...(notes.length ? { note: notes.join(" ") } : {}) };
204}