src/merge-units.ts

Functions

Imports

It imports src/env.ts, src/log.ts, src/ops-types/feed.ts, src/store-guards.ts, src/time.ts. Imported by src/auto-merge-tick.ts, src/controls-perform.ts, src/controls-plan.ts, src/inbox.ts, src/tick.ts.

Source

1import type { Env } from "./env";
2import type { InboxItem } from "./ops-types/feed";
3import { logEvent } from "./log";
4import { auditStatement } from "./store-guards";
5import { DAY_MS } from "./time";
6
7// EARNED PROMOTION AND DEMOTION (docs/design/design-merge-pipeline.md section 7, PR 9;
8// Dustin's DECIDE 9 and 10 of 2026-10-09). A unit is (namespace, path class). REPORT-ONLY:
9// this proposes and never widens anything (conventions 11.3). The one thing it changes on
10// its own only narrows: a rollback of a pull request the tick merged unattended suspends
11// that unit at once, so the tick stops merging its pull requests unattended and they fall
12// to class approve (src/auto-merge-tick.ts reads the suspensions).
13//
14// Evidence per unit, over 90 days:
15//   - approved merges, from pr_approvals (namespace and path_class, migrations/0043);
16//   - approvals voided for anything but a new push (an undo, red CI);
17//   - rollbacks, from deploy_outcomes, joined to the pull request's unit through the
18//     tick's auto-merged audit row or its approval;
19//   - reverts of a pull request the tick merged, from the auto-merge-reverted audit rows
20//     src/merge-safety.ts records. A revert counts as a rollback of an unattended merge:
21//     it suspends the unit and blocks its promotion.
22// The CI-green share joins in with the signed unit list (PR 10).
23//
24// Promotion proposal: at least 20 approved merges over at least 14 days, no rollback or revert, and
25// at most one approval voided for anything but a new push. Demotion: a rollback of an
26// unattended merge suspends its unit and proposes removing the unit from the signed list.
27// Both reach Dustin as inbox items of kind "proposal". Acting on a proposal is a signed
28// policy change (the sign_policy_change control, src/policy-sign.ts), and a suspension is
29// lifted with the unit_resume control (liftSuspension below).
30
31export const SUSPENDED_KEY = "merge-units:suspended";
32export const REPORT_KEY = "merge-units:report";
33// When Dustin lifted each unit's suspension (the unit_resume control): a rollback from
34// before then is answered and suspends the unit no more.
35export const LIFTED_KEY = "merge-units:lifted";
36const WINDOW_MS = 90 * DAY_MS;
37const PROMOTE_MERGES = 20;
38const PROMOTE_SPAN_MS = 14 * DAY_MS;
39const PROMOTE_MAX_BAD_VOIDS = 1;
40// A void that only says the pull request changed after it was approved is not evidence
41// against the unit: the approval was simply of an older head.
42const NOT_AGAINST = /^(new push|head moved|replaced)\b/;
43
44export const unitKey = (namespace: string, pathClass: string) => `${namespace}/${pathClass}`;
45
46export interface ApprovalEvidence {
47  namespace: string | null;
48  path_class: string | null;
49  status: string;
50  close_reason: string | null;
51  closed_at: string | null;
52}
53
54export interface RollbackEvidence {
55  // A deploy the gate rolled back, or a merged pull request that reverts an unattended
56  // merge (src/merge-safety.ts). For a revert, run_at is when the revert merged, run_id
57  // is 0 and head_sha is the reverted merge's (zeros when the row has none), and `reverted_by` is the revert's number.
58  kind: "rollback" | "revert";
59  reverted_by?: number | null;
60  namespace: string;
61  repo: string;
62  pr: number | null;
63  head_sha: string;
64  run_id: number;
65  run_at: string;
66  // How the pull request reached the default branch, and its unit, when known.
67  merged_by: "tick" | "approval" | "unknown";
68  path_class: string | null;
69}
70
71export interface Suspension {
72  unit: string;
73  since: string;
74  reason: string;
75}
76
77interface UnitProposal {
78  kind: "promote" | "demote";
79  unit: string;
80  title: string;
81  evidence: { approved_merges: number; span_days: number; rollbacks: number; bad_voids: number };
82}
83
84export interface UnitsReport {
85  generated: string;
86  proposals: UnitProposal[];
87  suspended: Suspension[];
88  // Rollbacks whose pull request's unit could not be found, so nothing was suspended.
89  unplaced: string[];
90}
91
92const happened = (r: RollbackEvidence) => (r.kind === "revert" ? `was reverted by ${r.repo}#${r.reverted_by}` : "was rolled back");
93
94/** The proposals and the suspensions, from the evidence. Pure. */
95export function judgeUnits(
96  approvals: readonly ApprovalEvidence[],
97  rollbacks: readonly RollbackEvidence[],
98  suspended: readonly Suspension[],
99  now: Date,
100  lifted: Readonly<Record<string, string>> = {}
101): UnitsReport {
102  const units = new Map<string, { merged: number[]; bad: number; rollbacks: number }>();
103  const unit = (k: string) => units.get(k) ?? (units.set(k, { merged: [], bad: 0, rollbacks: 0 }), units.get(k)!);
104  for (const a of approvals) {
105    if (!a.namespace || !a.path_class) continue;
106    const u = unit(unitKey(a.namespace, a.path_class));
107    if (a.status === "merged" && a.closed_at) u.merged.push(Date.parse(a.closed_at));
108    if (a.status === "voided" && !NOT_AGAINST.test(a.close_reason ?? "")) u.bad++;
109  }
110  const nowSuspended = new Map(suspended.map((s) => [s.unit, s]));
111  const unplaced: string[] = [];
112  const demote: UnitProposal[] = [];
113  for (const r of rollbacks) {
114    if (!r.path_class) {
115      unplaced.push(`${r.repo} ${r.kind === "revert" ? `revert #${r.reverted_by}` : `run ${r.run_id}`} at ${r.head_sha.slice(0, 7)}: ${r.pr ? `#${r.pr}'s unit is not recorded` : "no pull request is recorded for it"}`);
116      continue;
117    }
118    const k = unitKey(r.namespace, r.path_class);
119    unit(k).rollbacks++;
120    if (r.merged_by !== "tick") continue;
121    // Dustin lifted the suspension after this rollback: it is answered.
122    if (lifted[k] && Date.parse(r.run_at) <= Date.parse(lifted[k]!)) continue;
123    if (!nowSuspended.has(k)) nowSuspended.set(k, { unit: k, since: now.toISOString(), reason: `${r.repo}#${r.pr} merged unattended and ${happened(r)}${r.kind === "revert" ? "" : ` (run ${r.run_id})`}` });
124    if (!demote.some((d) => d.unit === k)) {
125      demote.push({ kind: "demote", unit: k, title: `Remove ${k.replace("/", " ")} from auto: ${r.repo}#${r.pr} merged unattended and ${happened(r)}`, evidence: { approved_merges: 0, span_days: 0, rollbacks: 0, bad_voids: 0 } });
126    }
127  }
128  const promote: UnitProposal[] = [];
129  for (const [k, u] of units) {
130    const d = demote.find((p) => p.unit === k);
131    const span = u.merged.length ? Math.max(...u.merged) - Math.min(...u.merged) : 0;
132    const evidence = { approved_merges: u.merged.length, span_days: Math.floor(span / DAY_MS), rollbacks: u.rollbacks, bad_voids: u.bad };
133    if (d) d.evidence = evidence;
134    if (nowSuspended.has(k)) continue;
135    if (u.merged.length >= PROMOTE_MERGES && span >= PROMOTE_SPAN_MS && u.rollbacks === 0 && u.bad <= PROMOTE_MAX_BAD_VOIDS) {
136      promote.push({ kind: "promote", unit: k, title: `Promote ${k.replace("/", " ")} to auto: ${u.merged.length} approved over ${evidence.span_days} days, 0 rollbacks`, evidence });
137    }
138  }
139  return { generated: now.toISOString(), proposals: [...demote, ...promote], suspended: [...nowSuspended.values()], unplaced };
140}
141
142/** The suspended units, as the tick reads them. An unreadable list reads as none
143 *  suspended is not safe, so it throws and the caller decides. */
144export async function readSuspended(kv: KVNamespace): Promise<Suspension[]> {
145  const raw = await kv.get(SUSPENDED_KEY);
146  return raw ? (JSON.parse(raw) as Suspension[]) : [];
147}
148
149async function readLifted(kv: KVNamespace): Promise<Record<string, string>> {
150  const raw = await kv.get(LIFTED_KEY);
151  return raw ? (JSON.parse(raw) as Record<string, string>) : {};
152}
153
154/** What lifting a unit's suspension will do, or why it cannot. Reads only. */
155export async function planLift(kv: KVNamespace, rawUnit: string | undefined): Promise<{ ok: true; unit: string; suspension: Suspension } | { ok: false; refusal: string }> {
156  const m = /^\s*([a-z0-9-]+)\s*\/\s*([a-z-]+)\s*$/.exec(rawUnit ?? "");
157  if (!m) return { ok: false, refusal: `unit_resume needs unit as '<namespace>/<path class>', for example 'capsid/docs'; got '${rawUnit ?? ""}'.` };
158  const unit = unitKey(m[1]!, m[2]!);
159  const suspension = (await readSuspended(kv)).find((s) => s.unit === unit);
160  if (!suspension) return { ok: false, refusal: `${unit} is not suspended, so there is nothing to lift.` };
161  return { ok: true, unit, suspension };
162}
163
164/** Lift a unit's suspension: the tick may merge its pull requests unattended again, and
165 *  the rollbacks before now suspend it no more. Audited. */
166export async function liftSuspension(env: Env, actor: string, rawUnit: string | undefined, reason: string, now: Date): Promise<{ ok: true; summary: string } | { ok: false; refusal: string }> {
167  const plan = await planLift(env.APP_KV, rawUnit);
168  if (!plan.ok) return plan;
169  const rest = (await readSuspended(env.APP_KV)).filter((s) => s.unit !== plan.unit);
170  const lifted = { ...(await readLifted(env.APP_KV)), [plan.unit]: now.toISOString() };
171  await env.APP_KV.put(LIFTED_KEY, JSON.stringify(lifted));
172  await env.APP_KV.put(SUSPENDED_KEY, JSON.stringify(rest));
173  await env.DB.batch([auditStatement(env.DB, actor, "merge-unit-resumed", plan.unit.split("/")[0]!, null, { unit: plan.unit, reason, was: plan.suspension })]);
174  return { ok: true, summary: `Lifted the suspension of ${plan.unit}. The tick may merge its pull requests unattended again; a new rollback suspends it again.` };
175}
176
177/** The evidence from D1. Reads only. */
178export async function readUnitEvidence(env: Env, now: Date): Promise<{ approvals: ApprovalEvidence[]; rollbacks: RollbackEvidence[] }> {
179  const since = new Date(now.getTime() - WINDOW_MS).toISOString();
180  const approvals = await env.DB.prepare(
181    `SELECT namespace, path_class, status, close_reason, closed_at FROM pr_approvals WHERE approved_at >= ?1 AND namespace IS NOT NULL ORDER BY id LIMIT 2000`
182  )
183    .bind(since)
184    .all<ApprovalEvidence>();
185  const rolled = await env.DB.prepare(
186    `SELECT namespace, repo, pr, head_sha, run_id, run_at FROM deploy_outcomes WHERE outcome IN ('rolled_back', 'rollback_failed') AND run_at >= ?1 ORDER BY run_at LIMIT 50`
187  )
188    .bind(since)
189    .all<{ namespace: string; repo: string; pr: number | null; head_sha: string; run_id: number; run_at: string }>();
190  const rollbacks: RollbackEvidence[] = [];
191  for (const r of rolled.results ?? []) {
192    let merged_by: RollbackEvidence["merged_by"] = "unknown";
193    let path_class: string | null = null;
194    if (r.pr !== null) {
195      // The tick's own record of an unattended merge carries the unit since PR 9.
196      const auto = await env.DB.prepare(
197        `SELECT params FROM audit_log WHERE action = 'auto-merged' AND namespace = ?1 AND json_extract(params, '$.repo') = ?2 AND json_extract(params, '$.pr') = ?3 ORDER BY id DESC LIMIT 1`
198      )
199        .bind(r.namespace, r.repo, r.pr)
200        .first<{ params: string }>();
201      if (auto) {
202        merged_by = "tick";
203        path_class = ((JSON.parse(auto.params) as { path_class?: unknown }).path_class as string | undefined) ?? null;
204      } else {
205        const approved = await env.DB.prepare(`SELECT path_class FROM pr_approvals WHERE repo = ?1 AND pr = ?2 AND status = 'merged' ORDER BY id DESC LIMIT 1`)
206          .bind(r.repo, r.pr)
207          .first<{ path_class: string | null }>();
208        if (approved) {
209          merged_by = "approval";
210          path_class = approved.path_class;
211        }
212      }
213    }
214    rollbacks.push({ ...r, kind: "rollback", merged_by, path_class });
215  }
216  // The reverts of pull requests the tick merged (src/merge-safety.ts): the audit row
217  // carries the unit, so a revert places itself.
218  const reverted = await env.DB.prepare(
219    `SELECT namespace, params FROM audit_log WHERE action = 'auto-merge-reverted' AND at >= ?1 ORDER BY id DESC LIMIT 50`
220  )
221    .bind(since.slice(0, 19).replace("T", " "))
222    .all<{ namespace: string; params: string }>();
223  for (const row of reverted.results ?? []) {
224    let p: { repo: string; pr: number; reverted_by: number; head_sha?: string | null; path_class?: string | null; merged_at: string };
225    try {
226      p = JSON.parse(row.params);
227    } catch (err) {
228      // One malformed row must not fail the step that suspends units.
229      logEvent("error", "MERGE_UNITS", { message: `MERGE_UNITS skipped an unreadable auto-merge-reverted row: ${err instanceof Error ? err.message : String(err)}` });
230      continue;
231    }
232    rollbacks.push({
233      kind: "revert",
234      reverted_by: p.reverted_by,
235      namespace: row.namespace,
236      repo: p.repo,
237      pr: p.pr,
238      head_sha: p.head_sha ?? "0".repeat(40),
239      run_id: 0,
240      run_at: p.merged_at,
241      merged_by: "tick",
242      path_class: p.path_class ?? null,
243    });
244  }
245  return { approvals: approvals.results ?? [], rollbacks };
246}
247
248export interface UnitsTickReport {
249  ran: boolean;
250  note: string;
251  newly_suspended: string[];
252}
253
254/** One pass: judge the units, store the report, and suspend a unit a rollback hit. */
255export async function unitsTick(env: Env, now: Date, read: typeof readUnitEvidence = readUnitEvidence): Promise<UnitsTickReport> {
256  const before = await readSuspended(env.APP_KV);
257  const lifted = await readLifted(env.APP_KV);
258  const evidence = await read(env, now);
259  const report = judgeUnits(evidence.approvals, evidence.rollbacks, before, now, lifted);
260  const added = report.suspended.filter((s) => !before.some((b) => b.unit === s.unit));
261  if (added.length) {
262    await env.APP_KV.put(SUSPENDED_KEY, JSON.stringify(report.suspended));
263    await env.DB.batch(added.map((s) => auditStatement(env.DB, "capsid:merge-units", "merge-unit-suspended", s.unit.split("/")[0]!, null, s)));
264  }
265  const previous = await env.APP_KV.get(REPORT_KEY);
266  const stable = (r: UnitsReport) => JSON.stringify({ ...r, generated: "" });
267  let changed = true;
268  try {
269    changed = !previous || stable(JSON.parse(previous) as UnitsReport) !== stable(report);
270  } catch {
271    // An unreadable report is replaced.
272  }
273  if (changed) await env.APP_KV.put(REPORT_KEY, JSON.stringify(report));
274  if (!added.length && !changed) return { ran: false, note: "no change", newly_suspended: [] };
275  return {
276    ran: true,
277    note: `${report.proposals.length} proposal(s), ${report.suspended.length} unit(s) suspended${added.length ? `, newly: ${added.map((s) => s.unit).join(", ")}` : ""}${report.unplaced.length ? `; ${report.unplaced.length} rollback(s) with no known unit` : ""}`,
278    newly_suspended: added.map((s) => s.unit),
279  };
280}
281
282/** The proposals and suspensions as inbox items, per namespace. */
283export async function unitInboxItems(kv: KVNamespace): Promise<Array<{ namespace: string; item: InboxItem }>> {
284  const raw = await kv.get(REPORT_KEY).catch(() => null);
285  if (!raw) return [];
286  let report: UnitsReport;
287  try {
288    report = JSON.parse(raw) as UnitsReport;
289  } catch {
290    return [];
291  }
292  const out: Array<{ namespace: string; item: InboxItem }> = [];
293  for (const s of report.suspended ?? []) {
294    out.push({ namespace: s.unit.split("/")[0]!, item: { title: `${s.unit.replace("/", " ")} is suspended from auto: ${s.reason}`, kind: "proposal", link: null, since: s.since } });
295  }
296  for (const p of report.proposals ?? []) out.push({ namespace: p.unit.split("/")[0]!, item: { title: p.title, kind: "proposal", link: null, since: report.generated } });
297  return out;
298}