src/job-outcomes.ts

Functions

Imports

It imports src/env.ts, src/github.ts, src/github/client.ts, src/jobs-schema.ts, src/time.ts. Imported by src/complete-ci.ts, src/job-claims.ts, src/jobs-holder.ts, src/jobs-seat.ts, src/jobs-transition.ts, src/outcome-prs.ts, src/tools/jobs.ts.

Source

1import type { Env } from "./env";
2import { ciStatus } from "./github";
3import { ghFetch, parsePrUrl, resolveRepo } from "./github/client";
4import type { JobRow } from "./jobs-schema";
5import { MINUTE_MS } from "./time";
6
7// One outcome row per finished job, written once, carrying counts.
8//
9// The Worker never stores a driver's count for something it could have checked. This
10// Worker holds a GitHub App token that reaches every repo in the portfolio, so for
11// anything that ended in a pull request the authority is GitHub and the driver's
12// number is an opinion. Where a check ran, the stored number is GitHub's and the field
13// is marked verified; where it could not run, the driver's number is stored and marked
14// unverified. A reader can always tell which.
15//
16// Null is not zero: a field nobody reported is null, a field counted and found empty
17// is 0. Collapsing the two would make "no driver reports test counts"
18// indistinguishable from "no driver adds tests".
19
20export const OUTCOME_RESULT_KINDS = ["pr", "doc", "none"] as const;
21export type OutcomeResultKind = (typeof OUTCOME_RESULT_KINDS)[number];
22
23// The fields a driver may report. An omitted field stays null in the row.
24export interface JobEvidence {
25  // Pull request URLs: the only field that unlocks verification.
26  prs?: string[];
27  commits?: number;
28  files_changed?: number;
29  tests_added?: number;
30}
31
32// Which fields the Worker checked itself, stored as JSON on the row, written out in full.
33interface VerifiedFields {
34  prs_opened: boolean;
35  prs_merged: boolean;
36  commits: boolean;
37  files_changed: boolean;
38  ci_green: boolean;
39}
40
41export interface JobOutcomeRow {
42  job_id: string;
43  agent: string;
44  namespace: string;
45  prs_opened: number | null;
46  prs_merged: number | null;
47  commits: number | null;
48  files_changed: number | null;
49  tests_added: number | null;
50  ci_green: number | null;
51  blocked_count: number;
52  resumed_count: number;
53  duration_minutes: number | null;
54  result_kind: OutcomeResultKind;
55  verified: string;
56  recorded_at: string;
57  // What the job's Claude Code sessions reported by OpenTelemetry (migrations/0025),
58  // summed from session_usage when the row is written. NULL when no telemetry for the
59  // job reached Capsid, never 0.
60  cost_usd: number | null;
61  tokens_input: number | null;
62  tokens_output: number | null;
63  tokens_cache_read: number | null;
64  tokens_cache_creation: number | null;
65  active_seconds: number | null;
66  // The job's kind, the model Capsid recommended and the model the session reported it ran
67  // (migrations/0032, src/model-routing.ts). NULL where not routed or not reported.
68  job_kind?: string | null;
69  model_chosen?: string | null;
70  model_actual?: string | null;
71}
72
73// A job's telemetry totals, as the outcome row stores them.
74export type JobUsage = Pick<
75  JobOutcomeRow,
76  "cost_usd" | "tokens_input" | "tokens_output" | "tokens_cache_read" | "tokens_cache_creation" | "active_seconds"
77>;
78
79const NO_USAGE: JobUsage = {
80  cost_usd: null,
81  tokens_input: null,
82  tokens_output: null,
83  tokens_cache_read: null,
84  tokens_cache_creation: null,
85  active_seconds: null,
86};
87
88const TOKEN_COLUMN: Record<string, "tokens_input" | "tokens_output" | "tokens_cache_read" | "tokens_cache_creation"> = {
89  input: "tokens_input",
90  output: "tokens_output",
91  cacheRead: "tokens_cache_read",
92  cacheCreation: "tokens_cache_creation",
93};
94
95/**
96 * The per-job totals from session_usage rows (metric, kind, total), which is what
97 * readJobUsage selects. Per metric family: a family with no row at all stays NULL
98 * (nothing was reported); once a family has a row, a type it lacks is 0, because
99 * Claude Code exports no point for a counter that did not move.
100 */
101export function usageFromTotals(rows: ReadonlyArray<{ metric: string; kind: string; total: number | null }>): JobUsage {
102  const usage: JobUsage = { ...NO_USAGE };
103  const families = new Set(rows.map((r) => r.metric));
104  if (families.has("claude_code.cost.usage")) usage.cost_usd = 0;
105  if (families.has("claude_code.token.usage")) {
106    for (const column of Object.values(TOKEN_COLUMN)) usage[column] = 0;
107  }
108  if (families.has("claude_code.active_time.total")) usage.active_seconds = 0;
109  for (const row of rows) {
110    const total = Number(row.total ?? 0);
111    if (!Number.isFinite(total)) continue;
112    if (row.metric === "claude_code.cost.usage") usage.cost_usd = (usage.cost_usd ?? 0) + total;
113    else if (row.metric === "claude_code.active_time.total") usage.active_seconds = (usage.active_seconds ?? 0) + total;
114    else if (row.metric === "claude_code.token.usage" && Object.hasOwn(TOKEN_COLUMN, row.kind)) {
115      const column = TOKEN_COLUMN[row.kind];
116      usage[column] = (usage[column] ?? 0) + total;
117    }
118  }
119  // The token columns are INTEGER.
120  for (const column of Object.values(TOKEN_COLUMN)) {
121    const value = usage[column];
122    if (value !== null) usage[column] = Math.round(value);
123  }
124  return usage;
125}
126
127/** The job's telemetry totals from session_usage (src/ops-otlp.ts writes it). */
128export async function readJobUsage(db: D1Database, jobId: string): Promise<JobUsage> {
129  const { results } = await db
130    .prepare(
131      `SELECT metric, kind, SUM(value) AS total FROM session_usage
132       WHERE job_id = ?1 AND metric IN ('claude_code.cost.usage', 'claude_code.token.usage', 'claude_code.active_time.total')
133       GROUP BY metric, kind`
134    )
135    .bind(jobId)
136    .all<{ metric: string; kind: string; total: number | null }>();
137  return usageFromTotals(results ?? []);
138}
139
140const NOTHING_VERIFIED: VerifiedFields = {
141  prs_opened: false,
142  prs_merged: false,
143  commits: false,
144  files_changed: false,
145  ci_green: false,
146};
147
148// Derived from result_ref, not declared: `resultRef` in src/limits.ts admits only a
149// document path or an https URL.
150export function resultKindOf(resultRef: string | null | undefined): OutcomeResultKind {
151  if (!resultRef) return "none";
152  return /^https:\/\//i.test(resultRef) ? "pr" : "doc";
153}
154
155// Whole minutes from the first claim, including time blocked at a gate; resume does
156// not reset claimed_at (migrations/0011's comment describes an earlier rule). A job the
157// seat released keeps its first claim time. null with no claim timestamp,
158// and null rather than negative if the clocks disagree.
159export function durationMinutes(claimedAt: string | null, now: Date): number | null {
160  if (!claimedAt) return null;
161  const started = Date.parse(claimedAt);
162  if (!Number.isFinite(started)) return null;
163  const elapsed = now.getTime() - started;
164  if (elapsed < 0) return null;
165  return Math.round(elapsed / MINUTE_MS);
166}
167
168interface PrFacts {
169  merged: boolean;
170  commits: number;
171  changed_files: number;
172  head_sha: string;
173  // The repo this pull request is on, resolved through the namespace mapping, so the
174  // CI lookup asks that repo and not the namespace primary.
175  repo: string;
176  // Which pull request answered at this number: GitHub's global node id and when it was
177  // opened. A recreated or transferred repo reuses numbers, so a URL alone does not say
178  // which pull request an outcome named (src/outcome-prs.ts). Null when GitHub omitted it.
179  node_id: string | null;
180  created_at: string | null;
181}
182
183export interface EvidenceVerdict {
184  prs_opened: number | null;
185  prs_merged: number | null;
186  commits: number | null;
187  files_changed: number | null;
188  tests_added: number | null;
189  ci_green: number | null;
190  verified: VerifiedFields;
191  // Every verification that could not run, and why.
192  notes: string[];
193  // Each named pull request's merge state as GitHub reported it, keyed by URL. A URL
194  // the Worker could not read is absent, never false. Written to job_outcome_prs at
195  // complete time, so the row says which pull requests were verified.
196  pr_states?: Record<string, boolean>;
197  // Each pull request read, by URL: which one it was, pinned on its join row so a later
198  // read of the same URL can tell whether the number still names it.
199  pr_identity?: Record<string, { node_id: string | null; created_at: string | null }>;
200  // What GitHub itself said, kept apart from the fields above, which mix it with the
201  // driver's numbers. job_evaluations (src/job-claims.ts) sets these beside the claim,
202  // so a check that disagreed with the driver is recorded as a disagreement rather
203  // than as the one number job_outcomes keeps. Null wherever the Worker did not read
204  // it; merged has an entry only for a pull request that was read.
205  github?: GitHubFacts;
206  // The driver's numbers as sent, before any of them was replaced by GitHub's.
207  reported?: { prs: string[] | null; commits: number | null; files_changed: number | null; tests_added: number | null };
208}
209
210export interface GitHubFacts {
211  merged: Record<string, boolean>;
212  // Sums over every named pull request, null unless every one was read: a sum over
213  // some of them is not the job's total.
214  commits: number | null;
215  files_changed: number | null;
216  // CI on the last named pull request's head, 1 or 0, null when not answered.
217  ci_green: number | null;
218}
219
220export async function prFacts(env: Env, namespace: string, url: string): Promise<PrFacts | string> {
221  const parsed = parsePrUrl(url);
222  if (!parsed) return `${url} is not a GitHub pull request URL, so nothing could be verified about it`;
223  const { owner, repo, number } = parsed;
224  // Resolved through the namespace mapping, the authorization boundary, so the
225  // outcome recorder cannot read a repo the namespace does not map.
226  let resolved;
227  try {
228    resolved = await resolveRepo(env, namespace, `${owner}/${repo}`);
229  } catch (err) {
230    return `${url}: ${err instanceof Error ? err.message : String(err)}`;
231  }
232  // Caught too: verifyEvidence promises its callers a note, never a throw.
233  let pr: { merged?: boolean; commits?: number; changed_files?: number; head?: { sha?: string }; node_id?: string; created_at?: string };
234  try {
235    const resp = await ghFetch(env, resolved.owner, resolved.repo, `/repos/${resolved.owner}/${resolved.repo}/pulls/${number}`);
236    if (!resp.ok) return `${url}: GitHub answered ${resp.status}, so its state could not be read`;
237    pr = (await resp.json()) as typeof pr;
238  } catch (err) {
239    return `${url}: GitHub could not be read (${err instanceof Error ? err.message : String(err)})`;
240  }
241  return {
242    merged: pr.merged === true,
243    commits: typeof pr.commits === "number" ? pr.commits : 0,
244    changed_files: typeof pr.changed_files === "number" ? pr.changed_files : 0,
245    head_sha: pr.head?.sha ?? "",
246    repo: resolved.full,
247    node_id: typeof pr.node_id === "string" && pr.node_id ? pr.node_id : null,
248    created_at: typeof pr.created_at === "string" && pr.created_at ? pr.created_at : null,
249  };
250}
251
252// CI is green only when every run has finished and none failed. No runs, or a run
253// still going, is null and unverified: not answered yet is not failed. `skipped` and
254// `neutral` do not fail, since a paths filter that excluded the change did not judge it.
255const NOT_A_FAILURE = new Set(["success", "skipped", "neutral"]);
256
257async function ciGreenForSha(env: Env, namespace: string, sha: string, repo: string): Promise<{ green: boolean | null; note?: string }> {
258  if (!sha) return { green: null, note: "the pull request carried no head sha, so CI could not be looked up" };
259  let status;
260  try {
261    // The repo the pull request is on, not the namespace primary.
262    status = await ciStatus(env, namespace, repo, { ref: sha, limit: 20 });
263  } catch (err) {
264    return { green: null, note: `CI could not be read for ${sha.slice(0, 7)}: ${err instanceof Error ? err.message : String(err)}` };
265  }
266  const runs = status.runs ?? [];
267  if (runs.length === 0) return { green: null, note: `no workflow runs for ${sha.slice(0, 7)}, so CI has nothing to say about it` };
268  if (runs.some((r) => r.status !== "completed")) {
269    return { green: null, note: `CI for ${sha.slice(0, 7)} has not finished, so it is neither green nor red yet` };
270  }
271  return { green: runs.every((r) => NOT_A_FAILURE.has(r.conclusion ?? "")) };
272}
273
274// Evidence arrives as an object or as a JSON string, because some clients flatten an
275// object argument to a string, and a client with a stale cached schema may refuse
276// the object. A string that does not parse is refused, not ignored, so evidence is
277// never silently discarded.
278export type EvidenceInput = JobEvidence | string | undefined;
279
280export function parseEvidence(input: EvidenceInput): { evidence: JobEvidence | undefined } | { error: string } {
281  if (input === undefined) return { evidence: undefined };
282  if (typeof input !== "string") return { evidence: input };
283  const text = input.trim();
284  if (text.length === 0) return { evidence: undefined };
285  let parsed: unknown;
286  try {
287    parsed = JSON.parse(text);
288  } catch {
289    return { error: `evidence was sent as a string that is not JSON: ${text.slice(0, 120)}` };
290  }
291  if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
292    return { error: "evidence parsed to something that is not an object, so there are no fields to read." };
293  }
294  const row = parsed as Record<string, unknown>;
295  const num = (key: string): number | undefined => {
296    const value = row[key];
297    if (value === undefined || value === null) return undefined;
298    const n = Number(value);
299    // Dropped rather than coerced: a stored 0 would read as counted and found none.
300    return Number.isInteger(n) && n >= 0 ? n : undefined;
301  };
302  const prs = Array.isArray(row.prs) ? row.prs.filter((p): p is string => typeof p === "string") : undefined;
303  return {
304    evidence: {
305      ...(prs && prs.length > 0 ? { prs } : {}),
306      ...(num("commits") !== undefined ? { commits: num("commits") } : {}),
307      ...(num("files_changed") !== undefined ? { files_changed: num("files_changed") } : {}),
308      ...(num("tests_added") !== undefined ? { tests_added: num("tests_added") } : {}),
309    },
310  };
311}
312
313// The pull requests a finished job names: evidence.prs, plus the job's result_ref when it
314// is a pull request URL. A session that sends only result_ref (the seat closing a job it
315// took, or a driver that left evidence out) named its pull request all the same, and
316// GitHub, not the sender, is the authority on it (job_d5262df1dc32). Deduplicated, in
317// order, so the pull request named twice is read and counted once.
318export function namedPrUrls(evidence: JobEvidence | undefined, resultRef: string | null | undefined): string[] {
319  const urls: string[] = [];
320  for (const url of [...(evidence?.prs ?? []), ...(parsePrUrl(resultRef) ? [resultRef!.trim()] : [])]) {
321    if (!urls.includes(url)) urls.push(url);
322  }
323  return urls;
324}
325
326// Best effort, and it never fails the job: with GitHub unreachable the outcome records
327// the driver's own numbers, marked unverified, with a note saying why.
328//
329// resultRef is the job's result_ref. `reported` stays what the evidence said, so the
330// record of what the driver sent is not changed by what the Worker also read.
331export async function verifyEvidence(
332  env: Env,
333  namespace: string,
334  evidence: JobEvidence | undefined,
335  resultRef?: string | null
336): Promise<EvidenceVerdict> {
337  const reported = {
338    commits: evidence?.commits ?? null,
339    files_changed: evidence?.files_changed ?? null,
340    tests_added: evidence?.tests_added ?? null,
341  };
342  const urls = namedPrUrls(evidence, resultRef);
343  const github: GitHubFacts = { merged: {}, commits: null, files_changed: null, ci_green: null };
344  const verdict: EvidenceVerdict = {
345    prs_opened: urls.length > 0 ? urls.length : null,
346    prs_merged: null,
347    ...reported,
348    ci_green: null,
349    verified: { ...NOTHING_VERIFIED },
350    notes: [],
351    github,
352    reported: { prs: evidence?.prs && evidence.prs.length > 0 ? [...evidence.prs] : null, ...reported },
353  };
354  if (urls.length === 0) return verdict;
355
356  const facts: PrFacts[] = [];
357  const read = new Map<string, PrFacts>();
358  for (const url of urls) {
359    const one = await prFacts(env, namespace, url);
360    if (typeof one === "string") verdict.notes.push(one);
361    else {
362      facts.push(one);
363      read.set(url, one);
364    }
365  }
366  verdict.pr_states = Object.fromEntries([...read].map(([url, f]) => [url, f.merged]));
367  verdict.pr_identity = Object.fromEntries([...read].map(([url, f]) => [url, { node_id: f.node_id, created_at: f.created_at }]));
368  github.merged = { ...verdict.pr_states };
369
370  // A partly read list keeps what was read. prs_merged counts the pull requests that
371  // were read and is marked verified, as the re-verification sweep already does.
372  // prs_opened stays the number named and UNVERIFIED, so no reader takes the merged
373  // count for a total: the agent record's merge rate reads only rows whose prs_opened
374  // is verified, and the run goes unjudged. Commits and files stay the
375  // driver's, since a sum over some of the pull requests is not the job's total.
376  if (facts.length !== urls.length) {
377    if (facts.length > 0) {
378      verdict.prs_merged = facts.filter((f) => f.merged).length;
379      verdict.verified.prs_merged = true;
380    }
381    verdict.notes.push(
382      `${facts.length} of ${urls.length} named pull requests could be read. prs_merged counts the ${facts.length} read and is verified; prs_opened, commits and files_changed stay the driver's and unverified, and the run is not judged.`
383    );
384    // CI on the last named pull request, when that one was read.
385    const lastUrl = urls[urls.length - 1];
386    const lastRead = read.get(lastUrl);
387    if (!lastRead) {
388      verdict.notes.push(`the last named pull request (${lastUrl}) could not be read, so CI was not looked up`);
389      return verdict;
390    }
391    const partialCi = await ciGreenForSha(env, namespace, lastRead.head_sha, lastRead.repo);
392    if (partialCi.note) verdict.notes.push(partialCi.note);
393    if (partialCi.green !== null) {
394      verdict.ci_green = partialCi.green ? 1 : 0;
395      verdict.verified.ci_green = true;
396      github.ci_green = verdict.ci_green;
397    }
398    return verdict;
399  }
400
401  verdict.prs_opened = facts.length;
402  verdict.prs_merged = facts.filter((f) => f.merged).length;
403  verdict.commits = facts.reduce((total, f) => total + f.commits, 0);
404  verdict.files_changed = facts.reduce((total, f) => total + f.changed_files, 0);
405  verdict.verified.prs_opened = true;
406  verdict.verified.prs_merged = true;
407  verdict.verified.commits = true;
408  verdict.verified.files_changed = true;
409  github.commits = verdict.commits;
410  github.files_changed = verdict.files_changed;
411
412  // CI on the last pull request's head, the one a driver opens at the end of its work.
413  const last = facts[facts.length - 1];
414  const ci = await ciGreenForSha(env, namespace, last.head_sha, last.repo);
415  if (ci.note) verdict.notes.push(ci.note);
416  if (ci.green !== null) {
417    verdict.ci_green = ci.green ? 1 : 0;
418    verdict.verified.ci_green = true;
419    github.ci_green = verdict.ci_green;
420  }
421  return verdict;
422}
423
424// usage: the job's telemetry totals (readJobUsage). Omitted is no telemetry: NULL.
425// actualModel: the model_id the session reported in its claim, or null when it reported none.
426export function outcomeFrom(job: JobRow, verdict: EvidenceVerdict, now: Date, usage: JobUsage = NO_USAGE, actualModel: string | null = null): JobOutcomeRow {
427  return {
428    job_id: job.id,
429    // Copied, not joined, because a release clears claimed_by.
430    // "unattributed" only keeps the NOT NULL column satisfied.
431    agent: job.claimed_by ?? "unattributed",
432    namespace: job.namespace,
433    prs_opened: verdict.prs_opened,
434    prs_merged: verdict.prs_merged,
435    commits: verdict.commits,
436    files_changed: verdict.files_changed,
437    tests_added: verdict.tests_added,
438    ci_green: verdict.ci_green,
439    blocked_count: job.blocked_count,
440    resumed_count: job.resumed_count,
441    duration_minutes: durationMinutes(job.claimed_at, now),
442    result_kind: resultKindOf(job.result_ref),
443    verified: JSON.stringify(verdict.verified),
444    recorded_at: now.toISOString(),
445    cost_usd: usage.cost_usd,
446    tokens_input: usage.tokens_input,
447    tokens_output: usage.tokens_output,
448    tokens_cache_read: usage.tokens_cache_read,
449    tokens_cache_creation: usage.tokens_cache_creation,
450    active_seconds: usage.active_seconds,
451    job_kind: job.kind ?? null,
452    model_chosen: job.model_recommended ?? null,
453    model_actual: actualModel,
454  };
455}
456
457// ON CONFLICT DO NOTHING: the first record of a job stands, and a second terminal
458// transition neither rewrites it nor aborts its own batch.
459export function outcomeStatement(db: D1Database, row: JobOutcomeRow) {
460  return db
461    .prepare(
462      `INSERT INTO job_outcomes (job_id, agent, namespace, prs_opened, prs_merged, commits, files_changed,
463         tests_added, ci_green, blocked_count, resumed_count, duration_minutes, result_kind, verified,
464         recorded_at, cost_usd, tokens_input, tokens_output,
465         tokens_cache_read, tokens_cache_creation, active_seconds, job_kind, model_chosen, model_actual)
466       VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24)
467       ON CONFLICT(job_id) DO NOTHING`
468    )
469    .bind(
470      row.job_id,
471      row.agent,
472      row.namespace,
473      row.prs_opened,
474      row.prs_merged,
475      row.commits,
476      row.files_changed,
477      row.tests_added,
478      row.ci_green,
479      row.blocked_count,
480      row.resumed_count,
481      row.duration_minutes,
482      row.result_kind,
483      row.verified,
484      row.recorded_at,
485      row.cost_usd,
486      row.tokens_input,
487      row.tokens_output,
488      row.tokens_cache_read,
489      row.tokens_cache_creation,
490      row.active_seconds,
491      row.job_kind ?? null,
492      row.model_chosen ?? null,
493      row.model_actual ?? null
494    );
495}