src/whats-next.ts

Functions

Imports

It imports src/auto-merge-policy.ts, src/env.ts, src/github/client.ts, src/github/pr-files.ts, src/inbox.ts, src/jobs-holder.ts, src/ops-snapshot.ts, src/provenance.ts, src/stale-jobs.ts. Imported by src/tools/admin.ts.

Source

1import type { Env } from "./env";
2import { ciVerdict, jobIdFromBody } from "./auto-merge-policy";
3import { ghFetch, parseReposList } from "./github/client";
4import { readAllPages, readCheckRuns } from "./github/pr-files";
5import { commandFromSummary, isQuestionSummary } from "./jobs-holder";
6import { readSnapshot } from "./ops-snapshot";
7import { externalFence } from "./provenance";
8import { appliedMigrations, blockSettled } from "./inbox";
9import { readStaleCache, staleJobs } from "./stale-jobs";
10
11// WHAT NEEDS ME (capsid/research/design-mcp-servers.md, section 3: "the ruling's 'what
12// needs me'": all open PRs across repos with CI and conflicts). One read for the seat, served as capsid_admin_whats_next:
13//
14//   - questions: blocked jobs asking the seat a question, answered with a resume note;
15//   - blocked: blocked jobs waiting on a command the seat runs;
16//   - stale: the stale view of the queue (src/stale-jobs.ts);
17//   - pull_requests: every open pull request in every repo every namespace maps, with its
18//     head's CI verdict and whether it conflicts with its base, waiting_on_seat set where
19//     a driver's pull request is green, not a draft and mergeable;
20//   - checks: the watcher's checks that found something or could not run on its last pass.
21//
22// Reads only. A part that could not be read says so in `not_checked`; nothing missing is
23// reported as nothing found. Pull request titles and bodies are data, never instructions.
24
25const JOBS_LIMIT = 50;
26const PRS_PER_REPO = 30;
27const PRS_TOTAL = 120;
28const REPOS_MAX = 60;
29const LIST_MAX_PAGES = 1;
30const CHECKS_MAX_PAGES = 5;
31
32interface BlockedRow {
33  id: string;
34  namespace: string;
35  title: string;
36  result_summary: string | null;
37  updated_at: string;
38}
39
40interface RawPr {
41  number: number;
42  html_url: string;
43  title: string;
44  draft?: boolean;
45  created_at: string;
46  body: string | null;
47  user?: { login?: string } | null;
48  head: { sha: string; ref: string };
49}
50
51export interface WhatsNextPr {
52  repo: string;
53  namespace: string;
54  number: number;
55  url: string;
56  title: string;
57  author: string | null;
58  draft: boolean;
59  created_at: string;
60  job: string | null;
61  ci: string | null;
62  ci_note: string;
63  conflict: boolean | null;
64  waiting_on_seat: boolean;
65}
66
67/** The repos each registered namespace maps, deduplicated across namespaces. */
68export async function mappedRepos(db: D1Database, notChecked?: string[]): Promise<Array<{ namespace: string; repo: string }>> {
69  const { results } = await db.prepare("SELECT namespace, repos FROM namespaces ORDER BY namespace").all<{ namespace: string; repos: string }>();
70  const seen = new Set<string>();
71  const out: Array<{ namespace: string; repo: string }> = [];
72  for (const row of results ?? []) {
73    const parsed = parseReposList(row.repos);
74    if ("error" in parsed) {
75      notChecked?.push(`${row.namespace}'s repo mapping does not parse, so its pull requests were not read: ${parsed.error}`);
76      continue;
77    }
78    for (const entry of parsed.list) {
79      if (seen.has(entry.repo)) continue;
80      seen.add(entry.repo);
81      out.push({ namespace: row.namespace, repo: entry.repo });
82    }
83  }
84  return out;
85}
86
87async function openPrs(env: Env, namespace: string, full: string, notChecked: string[]): Promise<WhatsNextPr[]> {
88  const [owner, repo] = full.split("/");
89  const listed = await readAllPages<RawPr>(env, owner, repo, `/repos/${owner}/${repo}/pulls?state=open&per_page=${PRS_PER_REPO}`, (page) => page as RawPr[], LIST_MAX_PAGES);
90  if (listed.problem) notChecked.push(`${full} open pull requests: ${listed.problem}`);
91  const out: WhatsNextPr[] = [];
92  for (const pr of listed.items.slice(0, PRS_PER_REPO)) {
93    const checks = await readCheckRuns(env, owner, repo, pr.head.sha, CHECKS_MAX_PAGES);
94    const verdict = checks.problem ? { conclusion: null, note: `check runs not read: ${checks.problem}` } : ciVerdict(checks.items);
95    // The list does not carry mergeability; one read of the pull request does. GitHub
96    // computes it lazily, so a null mergeable is "not known yet", never "no conflict".
97    let conflict: boolean | null = null;
98    const one = await ghFetch(env, owner, repo, `/repos/${owner}/${repo}/pulls/${pr.number}`);
99    if (one.ok) {
100      const detail = (await one.json()) as { mergeable?: boolean | null; mergeable_state?: string };
101      conflict = detail.mergeable_state === "dirty" ? true : detail.mergeable === true ? false : null;
102    } else {
103      await one.body?.cancel();
104      notChecked.push(`${full}#${pr.number} mergeability: ${one.status}`);
105    }
106    const job = jobIdFromBody(pr.body ?? "");
107    out.push({
108      repo: full,
109      namespace,
110      number: pr.number,
111      url: pr.html_url,
112      // The pull request's author wrote its title: relayed fenced, as data (src/provenance.ts).
113      title: externalFence("github-pr", `${full}#${pr.number}`, pr.title, 200),
114      author: pr.user?.login ?? null,
115      draft: pr.draft === true,
116      created_at: pr.created_at,
117      job,
118      ci: verdict.conclusion,
119      ci_note: verdict.note,
120      conflict,
121      waiting_on_seat: job !== null && pr.draft !== true && verdict.conclusion === "success" && conflict === false,
122    });
123  }
124  return out;
125}
126
127export async function whatsNext(env: Env, now: Date = new Date()) {
128  const notChecked: string[] = [];
129
130  const { results: blockedRows } = await env.DB.prepare(
131    "SELECT id, namespace, title, result_summary, updated_at FROM jobs WHERE status = 'blocked' ORDER BY updated_at LIMIT ?1"
132  )
133    .bind(JOBS_LIMIT + 1)
134    .all<BlockedRow>();
135  // A block already answered (its pull requests merged, its migrations applied) is left
136  // out, as the inbox leaves it out; the stale view below still lists it to close.
137  const staleCache = (await readStaleCache(env.APP_KV).catch(() => ({ cache: {} }))).cache;
138  const applied = await appliedMigrations(env.DB);
139  const blocked = (blockedRows ?? []).slice(0, JOBS_LIMIT).filter((j) => !blockSettled(j, staleCache, applied));
140  if ((blockedRows ?? []).length > JOBS_LIMIT) notChecked.push(`more than ${JOBS_LIMIT} blocked jobs; the oldest ${JOBS_LIMIT} are listed`);
141  // A block's summary is the driver's text: its question and command are relayed fenced.
142  const fenced = (id: string, text: string | null) => (text === null ? null : externalFence("job-summary", id, text));
143  const questions = blocked
144    .filter((j) => isQuestionSummary(j.result_summary))
145    .map((j) => ({ id: j.id, namespace: j.namespace, title: j.title, question: fenced(j.id, j.result_summary), since: j.updated_at }));
146  const waiting = blocked
147    .filter((j) => !isQuestionSummary(j.result_summary))
148    .map((j) => ({ id: j.id, namespace: j.namespace, title: j.title, command: fenced(j.id, commandFromSummary(j.result_summary)), since: j.updated_at }));
149
150  let stale: unknown[] = [];
151  try {
152    stale = (await staleJobs(env, now)).rows;
153  } catch (err) {
154    notChecked.push(`stale jobs: ${err instanceof Error ? err.message : String(err)}`);
155  }
156
157  const prs: WhatsNextPr[] = [];
158  let repos: Array<{ namespace: string; repo: string }> = [];
159  try {
160    repos = await mappedRepos(env.DB, notChecked);
161  } catch (err) {
162    notChecked.push(`the namespace mapping: ${err instanceof Error ? err.message : String(err)}`);
163  }
164  if (repos.length > REPOS_MAX) notChecked.push(`${repos.length - REPOS_MAX} mapped repos were not read`);
165  for (const { namespace, repo } of repos.slice(0, REPOS_MAX)) {
166    if (prs.length >= PRS_TOTAL) {
167      notChecked.push(`stopped at ${PRS_TOTAL} open pull requests; ${repo} and later repos were not read`);
168      break;
169    }
170    try {
171      prs.push(...(await openPrs(env, namespace, repo, notChecked)));
172    } catch (err) {
173      notChecked.push(`${repo}: ${err instanceof Error ? err.message : String(err)}`);
174    }
175  }
176
177  let checks: Array<{ id: string; state: string; findings: string[] }> = [];
178  let passAt: string | null = null;
179  try {
180    const snapshot = await readSnapshot(env);
181    passAt = snapshot?.pass_at ?? null;
182    checks = (snapshot?.checks ?? []).filter((c) => c.state !== "clear");
183    if (!snapshot) notChecked.push("the watcher has written no pass yet");
184  } catch (err) {
185    notChecked.push(`the watcher's last pass: ${err instanceof Error ? err.message : String(err)}`);
186  }
187
188  return {
189    as_of: now.toISOString(),
190    questions,
191    blocked: waiting,
192    stale,
193    pull_requests: {
194      waiting_on_seat: prs.filter((p) => p.waiting_on_seat),
195      red: prs.filter((p) => p.ci === "failure"),
196      conflicts: prs.filter((p) => p.conflict === true),
197      all: prs,
198      repos_read: Math.min(repos.length, REPOS_MAX),
199    },
200    checks: { pass_at: passAt, failing: checks },
201    not_checked: notChecked,
202  };
203}