src/merge-safety.ts

Functions

Imports

It imports src/approval-executor.ts, src/auto-merge-policy.ts, src/env.ts, src/github/client.ts, src/log.ts, src/ops-types/feed.ts, src/store-guards.ts, src/time.ts, src/trends.ts, src/watcher.ts. Imported by src/auto-merge-tick.ts, src/controls-perform.ts, src/controls-plan.ts, src/ops-feed.ts, src/push.ts, src/status.ts, src/tick.ts.

Source

1import { AUTO_MERGED_APPROVAL_REASON } from "./approval-executor";
2import { loadMergePolicy } from "./auto-merge-policy";
3import type { Env } from "./env";
4import { ghFetch, parsePrUrl, resolveRepo } from "./github/client";
5import { logEvent } from "./log";
6import type { OpsMergeSafety, OpsMergeSafetyRow } from "./ops-types/feed";
7import { auditStatement } from "./store-guards";
8import { timeMs } from "./trends";
9import { passDue } from "./watcher";
10import { DAY_MS } from "./time";
11
12// THE AUTO-MERGE SAFETY METER AND SHUTOFF (docs/design/design-merge-pipeline.md, "The
13// safety meter"). Merge units (src/merge-units.ts) narrow one path class after a bad
14// unattended merge; this narrows the whole namespace. It only narrows: the tick stops
15// merging that namespace's pull requests unattended and they fall to class approve, and
16// Dustin's approvals still execute. Lifting it is the automerge_resume control.
17//
18// Three parts, none of them a D1 migration (audit_log rows and APP_KV only):
19//   1. Revert tracking. A merged pull request that reverts one the tick auto-merged is
20//      recorded as an `auto-merge-reverted` audit row. A rollback is the deploy gate's
21//      verdict; a revert is Dustin's (or a driver's) verdict, and nothing else records it.
22//   2. The meter: the last seven days per namespace, written to APP_KV so the status
23//      tool, the Portal feed and the phone alerts read it with no D1 or GitHub read.
24//   3. The shutoff: a namespace pauses when 2 or more of the pull requests the tick merged
25//      were rolled back or reverted in 7 days, or when that rate is more than 3 times the
26//      rate over the 30 days before.
27
28export const PAUSED_KEY = "merge-safety:paused";
29export const LIFTED_KEY = "merge-safety:lifted";
30export const REPORT_KEY = "merge-safety:report";
31export const REVERTS_STAMP_KEY = "merge-safety:reverts-last";
32
33const WEEK_MS = 7 * DAY_MS;
34// The previous 30 days are days 7 to 37 before now; the audit window also has to reach
35// the merge a rollback is joined to, so it is the merge units' 90.
36const BASELINE_MS = 30 * DAY_MS;
37const WINDOW_MS = 90 * DAY_MS;
38const PAUSE_BAD_COUNT = 2;
39const PAUSE_RATE_FACTOR = 3;
40const REVERTS_EVERY_MINUTES = 30;
41const CLOSED_PRS_READ = 50;
42const AUDIT_ROWS_LIMIT = 5000;
43const ACTOR = "capsid:merge-safety";
44
45export interface Paused {
46  since: string;
47  reason: string;
48}
49
50// ---------------------------------------------------------------------------
51// 1. Revert tracking
52
53/** The pull requests and commits a merged pull request says it reverts. Pure.
54 *  Recognised: GitHub's revert button ("Reverts owner/repo#N" in the body, with a title
55 *  `Revert "..."`), "reverts #N" or "revert #N" at the start of the title or of a body line
56 *  (same repo only, any letter case), and git's "This reverts commit <sha>" (7 to 40 hex). A mention of
57 *  another repo's pull request names none of this repo's. */
58export function revertTargets(title: string, body: string, repo: string): { prs: number[]; shas: string[] } {
59  const prs = new Set<number>();
60  const shas = new Set<string>();
61  const text = `${title ?? ""}\n${body ?? ""}`;
62  for (const m of text.matchAll(/\breverts?\s+([\w.-]+\/[\w.-]+)#(\d+)\b/gi)) {
63    if (m[1]!.toLowerCase() === repo.toLowerCase()) prs.add(Number(m[2]));
64  }
65  // A bare "revert #N" is a statement only at the start of the title or of a body line
66  // (after a list marker at most): "Don't revert #12 here" is prose and names nothing.
67  const lead = /^[ \t]*(?:[-*][ \t]+)?reverts?\s+#(\d+)\b/gim;
68  for (const m of `${title ?? ""}\n${body ?? ""}`.matchAll(lead)) prs.add(Number(m[1]));
69  for (const m of text.matchAll(/\bThis reverts commit\s+([0-9a-f]{7,40})\b/gi)) shas.add(m[1]!.toLowerCase());
70  return { prs: [...prs].sort((a, b) => a - b), shas: [...shas] };
71}
72
73interface ClosedPr {
74  number: number;
75  title?: string | null;
76  body?: string | null;
77  merged_at?: string | null;
78}
79
80interface AutoMergedRow {
81  repo: string;
82  pr: number;
83  head_sha: string | null;
84  path_class: string | null;
85}
86
87/** The auto-merged audit row for a pull request, or for a commit by its sha prefix
88 *  (merge_sha or head_sha). The same query shape the merge units use. */
89async function autoMergedBy(env: Env, namespace: string, repo: string, target: { pr: number } | { sha: string }): Promise<AutoMergedRow | null> {
90  const row =
91    "pr" in target
92      ? await env.DB.prepare(
93          `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`
94        )
95          .bind(namespace, repo, target.pr)
96          .first<{ params: string }>()
97      : // The sha is hex by the pattern that found it, so the prefix carries no wildcard.
98        await env.DB.prepare(
99          `SELECT params FROM audit_log WHERE action = 'auto-merged' AND namespace = ?1 AND json_extract(params, '$.repo') = ?2
100             AND (json_extract(params, '$.merge_sha') LIKE ?3 OR json_extract(params, '$.head_sha') LIKE ?3) ORDER BY id DESC LIMIT 1`
101        )
102          .bind(namespace, repo, `${target.sha}%`)
103          .first<{ params: string }>();
104  if (!row) return null;
105  const p = JSON.parse(row.params) as { repo: string; pr: number; head_sha?: string | null; path_class?: string | null };
106  return { repo: p.repo, pr: p.pr, head_sha: p.head_sha ?? null, path_class: p.path_class ?? null };
107}
108
109export interface RevertsReport {
110  ran: boolean;
111  note: string;
112  recorded: string[];
113}
114
115/** One namespace's revert pass: the recently merged pull requests that revert an
116 *  auto-merged one, each recorded once. Returns what it recorded. */
117export async function recordReverts(env: Env, namespace: string, now: Date): Promise<string[]> {
118  const { owner, repo } = await resolveRepo(env, namespace);
119  const full = `${owner}/${repo}`;
120  const resp = await ghFetch(env, owner, repo, `/repos/${owner}/${repo}/pulls?state=closed&sort=updated&direction=desc&per_page=${CLOSED_PRS_READ}`);
121  if (!resp.ok) throw new Error(`the closed pull requests of ${full} returned ${resp.status}`);
122  const recorded: string[] = [];
123  for (const closed of (await resp.json()) as ClosedPr[]) {
124    if (!closed.merged_at) continue;
125    // A revert older than the evidence window answers nothing the meter still reads.
126    if (now.getTime() - Date.parse(closed.merged_at) > WINDOW_MS) continue;
127    const targets = revertTargets(closed.title ?? "", closed.body ?? "", full);
128    const reverted = new Map<number, AutoMergedRow>();
129    for (const pr of targets.prs) {
130      const row = await autoMergedBy(env, namespace, full, { pr });
131      if (row) reverted.set(row.pr, row);
132    }
133    for (const sha of targets.shas) {
134      const row = await autoMergedBy(env, namespace, full, { sha });
135      if (row) reverted.set(row.pr, row);
136    }
137    for (const row of reverted.values()) {
138      // A pull request does not revert itself, and one revert is recorded once per
139      // reverted pull request however many ticks see it.
140      if (row.pr === closed.number) continue;
141      const seen = await env.DB.prepare(
142        `SELECT 1 AS found FROM audit_log WHERE action = 'auto-merge-reverted' AND namespace = ?1 AND json_extract(params, '$.repo') = ?2
143           AND json_extract(params, '$.pr') = ?3 AND json_extract(params, '$.reverted_by') = ?4 LIMIT 1`
144      )
145        .bind(namespace, full, row.pr, closed.number)
146        .first<{ found: number }>();
147      if (seen) continue;
148      await env.DB.batch([
149        auditStatement(env.DB, ACTOR, "auto-merge-reverted", namespace, null, {
150          repo: full,
151          pr: row.pr,
152          reverted_by: closed.number,
153          head_sha: row.head_sha,
154          path_class: row.path_class,
155          merged_at: closed.merged_at,
156          at: now.toISOString(),
157        }),
158      ]);
159      recorded.push(`${full}#${row.pr} by #${closed.number}`);
160    }
161  }
162  return recorded;
163}
164
165// A job's merged pull request that a later merge reverted, in any repo a recent outcome
166// names, auto-merged or not. The model routing learns from it (src/model-learning.ts): a
167// merge that had to be undone is not work that held. Recorded once per (job, pull
168// request, revert) as a `job-pr-reverted` audit row.
169const OUTCOME_REPOS_MAX = 30;
170const SHA_LOOKUPS_MAX = 20;
171
172/** The revert pass over the repos of the last 90 days' merged outcome pull requests.
173 *  Returns what it recorded; a GitHub failure for one repo is logged and the rest go on. */
174export async function recordOutcomeReverts(env: Env, now: Date): Promise<string[]> {
175  const since = new Date(now.getTime() - WINDOW_MS).toISOString().replace("T", " ").slice(0, 19);
176  const { results } = await env.DB.prepare(
177    `SELECT p.job_id AS job_id, p.pr_url AS pr_url, o.namespace AS namespace FROM job_outcome_prs p JOIN job_outcomes o ON o.job_id = p.job_id
178     WHERE p.merged = 1 AND p.recorded_at >= ?1 ORDER BY p.recorded_at DESC LIMIT ?2`
179  )
180    .bind(since, AUDIT_ROWS_LIMIT)
181    .all<{ job_id: string; pr_url: string; namespace: string }>();
182  const repos = new Map<string, { owner: string; repo: string; prs: Map<number, Array<{ job_id: string; namespace: string }>> }>();
183  for (const row of results ?? []) {
184    const ref = parsePrUrl(row.pr_url);
185    if (!ref) continue;
186    const key = `${ref.owner}/${ref.repo}`.toLowerCase();
187    const entry = repos.get(key) ?? (repos.set(key, { owner: ref.owner, repo: ref.repo, prs: new Map() }), repos.get(key)!);
188    entry.prs.set(ref.number, [...(entry.prs.get(ref.number) ?? []), { job_id: row.job_id, namespace: row.namespace }]);
189  }
190  const recorded: string[] = [];
191  for (const { owner, repo, prs } of [...repos.values()].slice(0, OUTCOME_REPOS_MAX)) {
192    const full = `${owner}/${repo}`;
193    try {
194      const resp = await ghFetch(env, owner, repo, `/repos/${owner}/${repo}/pulls?state=closed&sort=updated&direction=desc&per_page=${CLOSED_PRS_READ}`);
195      if (!resp.ok) throw new Error(`the closed pull requests of ${full} returned ${resp.status}`);
196      let shaLookups = 0;
197      for (const closed of (await resp.json()) as ClosedPr[]) {
198        if (!closed.merged_at || now.getTime() - Date.parse(closed.merged_at) > WINDOW_MS) continue;
199        const targets = revertTargets(closed.title ?? "", closed.body ?? "", full);
200        const numbers = new Set(targets.prs);
201        // git's "This reverts commit <sha>" names a commit: GitHub says which pull request
202        // brought it in.
203        for (const sha of targets.shas) {
204          if (shaLookups++ >= SHA_LOOKUPS_MAX) break;
205          const pulls = await ghFetch(env, owner, repo, `/repos/${owner}/${repo}/commits/${sha}/pulls`);
206          if (pulls.ok) for (const p of (await pulls.json()) as Array<{ number: number }>) numbers.add(p.number);
207        }
208        for (const pr of numbers) {
209          if (pr === closed.number) continue;
210          for (const job of prs.get(pr) ?? []) {
211            const seen = await env.DB.prepare(
212              `SELECT 1 AS found FROM audit_log WHERE action = 'job-pr-reverted' AND json_extract(params, '$.job_id') = ?1
213                 AND json_extract(params, '$.pr') = ?2 AND json_extract(params, '$.reverted_by') = ?3 LIMIT 1`
214            )
215              .bind(job.job_id, pr, closed.number)
216              .first<{ found: number }>();
217            if (seen) continue;
218            await env.DB.batch([
219              auditStatement(env.DB, ACTOR, "job-pr-reverted", job.namespace, null, {
220                job_id: job.job_id,
221                repo: full,
222                pr,
223                reverted_by: closed.number,
224                merged_at: closed.merged_at,
225                at: now.toISOString(),
226              }),
227            ]);
228            recorded.push(`${job.job_id} (${full}#${pr} by #${closed.number})`);
229          }
230        }
231      }
232    } catch (err) {
233      logEvent("error", "MERGE_SAFETY", { message: `MERGE_SAFETY could not read the reverts of ${full}: ${err instanceof Error ? err.message : String(err)}` });
234    }
235  }
236  return recorded;
237}
238
239/** The revert pass over every namespace the signed merge policy covers, and over the
240 *  repos of recent job outcomes, at most once in 30 minutes (a KV stamp, like the
241 *  watcher's). A policy that does not load, or a GitHub failure for one namespace, is
242 *  logged and the rest go on. */
243export async function revertsTick(env: Env, now: Date, policy?: Awaited<ReturnType<typeof loadMergePolicy>>): Promise<RevertsReport> {
244  const last = await env.APP_KV.get(REVERTS_STAMP_KEY).catch(() => null);
245  const due = passDue(last, REVERTS_EVERY_MINUTES, now);
246  if (!due.due) return { ran: false, note: due.reason, recorded: [] };
247  const outcomes = await recordOutcomeReverts(env, now);
248  const loaded = policy ?? (await loadMergePolicy(env));
249  if ("error" in loaded) {
250    await env.APP_KV.put(REVERTS_STAMP_KEY, now.toISOString());
251    return { ran: true, note: `${outcomes.length} revert(s) of job pull requests recorded; ${loaded.error}`, recorded: outcomes };
252  }
253  const recorded: string[] = [...outcomes];
254  for (const namespace of loaded.policy.namespaces) {
255    try {
256      recorded.push(...(await recordReverts(env, namespace, now)));
257    } catch (err) {
258      logEvent("error", "MERGE_SAFETY", { message: `MERGE_SAFETY could not read the reverts of ${namespace}: ${err instanceof Error ? err.message : String(err)}` });
259    }
260  }
261  await env.APP_KV.put(REVERTS_STAMP_KEY, now.toISOString());
262  return { ran: true, note: `${recorded.length} revert(s) recorded (job and auto-merged pull requests)${recorded.length ? `: ${recorded.join(", ")}` : ""}`, recorded };
263}
264
265// ---------------------------------------------------------------------------
266// 2. The weekly counts
267
268export interface AuditEvidence {
269  action: string;
270  namespace: string | null;
271  params: string | null;
272  at: string;
273}
274
275export interface OutcomeEvidence {
276  namespace: string;
277  repo: string;
278  pr: number | null;
279  outcome: string;
280  run_at: string;
281}
282
283export interface SafetyCounts {
284  namespace: string;
285  // The last seven days.
286  auto_merged: number;
287  approved: number;
288  deploys_verified: number;
289  rolled_back: number;
290  rollback_failed: number;
291  reverts: number;
292  // Rollbacks and reverts of pull requests the tick merged unattended, one per pull
293  // request however many deploys or reverts it drew, before the lifted time is applied.
294  bad_auto_7d: number;
295  // The times (ms) behind the judgement, over days 0 to 37: each bad event of an
296  // unattended pull request, and each unattended merge. judgeShutoff windows and lifts
297  // them, counting a pull request once per window at its latest event after the lift.
298  bad_at: Array<{ key: string; ms: number }>;
299  auto_at: number[];
300}
301
302function paramsOf(row: AuditEvidence): Record<string, unknown> {
303  try {
304    const parsed: unknown = row.params ? JSON.parse(row.params) : {};
305    return parsed && typeof parsed === "object" ? (parsed as Record<string, unknown>) : {};
306  } catch {
307    return {};
308  }
309}
310
311/** The counts per namespace from the evidence. Pure. `namespaces` are listed even with
312 *  no event; one seen in the evidence is added. A pull request counts as bad once, at its
313 *  first rollback or revert, so one incident is not two. */
314export function countsFrom(audit: readonly AuditEvidence[], outcomes: readonly OutcomeEvidence[], namespaces: readonly string[], now: Date): SafetyCounts[] {
315  const t = now.getTime();
316  const out = new Map<string, SafetyCounts>();
317  const get = (ns: string): SafetyCounts =>
318    out.get(ns) ?? (out.set(ns, { namespace: ns, auto_merged: 0, approved: 0, deploys_verified: 0, rolled_back: 0, rollback_failed: 0, reverts: 0, bad_auto_7d: 0, bad_at: [], auto_at: [] }), out.get(ns)!);
319  for (const ns of namespaces) get(ns);
320  const inWeek = (ms: number) => ms > t - WEEK_MS && ms <= t;
321  const inReach = (ms: number) => ms > t - WEEK_MS - BASELINE_MS && ms <= t;
322
323  // The unattended merges, by pull request, for the join below.
324  const autoPrs = new Set<string>();
325  for (const a of audit) {
326    if (a.action !== "auto-merged" || !a.namespace) continue;
327    const p = paramsOf(a);
328    autoPrs.add(`${a.namespace} ${String(p.repo)}#${String(p.pr)}`);
329    const ms = timeMs(a.at);
330    const c = get(a.namespace);
331    if (inWeek(ms)) c.auto_merged++;
332    if (inReach(ms)) c.auto_at.push(ms);
333  }
334
335  // Every bad event of each unattended pull request, kept so judgeShutoff can take the
336  // latest one after a resume (a fresh revert of a PR rolled back long ago counts).
337  const badEvents = new Map<string, { ns: string; ms: number[] }>();
338  const bad = (ns: string, repo: string, pr: number | null, ms: number) => {
339    if (pr === null || !autoPrs.has(`${ns} ${repo}#${pr}`)) return;
340    const key = `${ns} ${repo}#${pr}`;
341    const seen = badEvents.get(key) ?? { ns, ms: [] };
342    seen.ms.push(ms);
343    badEvents.set(key, seen);
344  };
345
346  for (const a of audit) {
347    if (!a.namespace) continue;
348    const c = get(a.namespace);
349    const p = paramsOf(a);
350    if (a.action === "approved-merged") {
351      // A policy merge of a pull request that held an approval also closes the approval
352      // with an approved-merged row; that row is the same merge (src/trends.ts).
353      if (p.reason === AUTO_MERGED_APPROVAL_REASON) continue;
354      if (inWeek(timeMs(a.at))) c.approved++;
355    } else if (a.action === "auto-merge-reverted") {
356      const ms = typeof p.merged_at === "string" && Number.isFinite(Date.parse(p.merged_at)) ? Date.parse(p.merged_at) : timeMs(a.at);
357      if (inWeek(ms)) c.reverts++;
358      bad(a.namespace, String(p.repo), typeof p.pr === "number" ? p.pr : null, ms);
359    }
360  }
361  for (const o of outcomes) {
362    const c = get(o.namespace);
363    const ms = timeMs(o.run_at);
364    if (o.outcome === "verified" && inWeek(ms)) c.deploys_verified++;
365    if (o.outcome === "rolled_back" || o.outcome === "rollback_failed") {
366      if (inWeek(ms)) c[o.outcome]++;
367      bad(o.namespace, o.repo, o.pr, ms);
368    }
369  }
370  for (const [key, { ns, ms }] of badEvents) {
371    const c = get(ns);
372    for (const m of ms) if (inReach(m)) c.bad_at.push({ key, ms: m });
373    // A pull request is one bad event this week however many it drew.
374    if (ms.some(inWeek)) c.bad_auto_7d++;
375  }
376  return [...out.values()];
377}
378
379/** The evidence from D1 and the counts from it. Reads only: two statements. */
380async function readSafetyCounts(env: Env, now: Date, namespaces: readonly string[]): Promise<SafetyCounts[]> {
381  const since = new Date(now.getTime() - WINDOW_MS);
382  const sqlite = since.toISOString().slice(0, 19).replace("T", " ");
383  const [audit, outcomes] = await Promise.all([
384    env.DB.prepare(
385      `SELECT action, namespace, params, at FROM audit_log WHERE action IN ('auto-merged', 'approved-merged', 'auto-merge-reverted') AND at >= ?1 ORDER BY id DESC LIMIT ?2`
386    )
387      .bind(sqlite, AUDIT_ROWS_LIMIT)
388      .all<AuditEvidence>(),
389    env.DB.prepare(`SELECT namespace, repo, pr, outcome, run_at FROM deploy_outcomes WHERE run_at >= ?1 AND outcome IN ('verified', 'rolled_back', 'rollback_failed') ORDER BY run_at LIMIT ?2`)
390      .bind(since.toISOString(), AUDIT_ROWS_LIMIT)
391      .all<OutcomeEvidence>(),
392  ]);
393  return countsFrom(audit.results ?? [], outcomes.results ?? [], namespaces, now);
394}
395
396// ---------------------------------------------------------------------------
397// 3. The shutoff
398
399/** The bad events and unattended merges of a namespace in the last seven days and in
400 *  the 30 days before, with the events at or before the lifted time left out. */
401function windowed(c: SafetyCounts, lifted: string | undefined, now: Date) {
402  const t = now.getTime();
403  const after = lifted && Number.isFinite(Date.parse(lifted)) ? Date.parse(lifted) : -Infinity;
404  const week = (ms: number) => ms > t - WEEK_MS;
405  // One event per pull request: its latest after the lifted time.
406  const byPr = new Map<string, number>();
407  for (const e of c.bad_at) if (e.ms > after) byPr.set(e.key, Math.max(byPr.get(e.key) ?? -Infinity, e.ms));
408  const latest = [...byPr.values()];
409  return {
410    bad7: latest.filter(week).length,
411    bad30: latest.filter((ms) => !week(ms)).length,
412    auto7: c.auto_at.filter(week).length,
413    auto30: c.auto_at.filter((ms) => !week(ms)).length,
414  };
415}
416
417/** The namespaces to pause now, each with its reason. Pure.
418 *  Pause when 2 or more of the tick's own merges were rolled back or reverted in 7 days,
419 *  or when the 7-day rate (bad over merges, merges floored at 1) is more than 3 times the
420 *  rate over days 7 to 37. The rate rule needs a nonzero baseline (at least one bad
421 *  event and at least one merge): with a clean baseline the rate is infinite for any
422 *  first event, and one bad event alone is the merge units' job, not a namespace's, so
423 *  only the count rule applies there. Events at or before the namespace's lifted time are
424 *  answered and count for nothing. A namespace already paused is not paused again. */
425export function judgeShutoff(counts: readonly SafetyCounts[], paused: Readonly<Record<string, Paused>>, lifted: Readonly<Record<string, string>>, now: Date): Record<string, string> {
426  const out: Record<string, string> = {};
427  for (const c of counts) {
428    if (paused[c.namespace]) continue;
429    const w = windowed(c, lifted[c.namespace], now);
430    if (w.bad7 >= PAUSE_BAD_COUNT) {
431      out[c.namespace] = `${w.bad7} of the pull requests auto-merge merged were rolled back or reverted in 7 days`;
432      continue;
433    }
434    if (w.bad30 >= 1 && w.auto30 >= 1) {
435      const rate7 = w.bad7 / Math.max(w.auto7, 1);
436      const rate30 = w.bad30 / w.auto30;
437      if (rate7 > PAUSE_RATE_FACTOR * rate30) {
438        out[c.namespace] = `${w.bad7} of ${w.auto7} auto-merged pull request(s) rolled back or reverted in 7 days, over ${PAUSE_RATE_FACTOR} times the ${w.bad30} of ${w.auto30} in the 30 days before`;
439      }
440    }
441  }
442  return out;
443}
444
445/** The paused namespaces, as the tick reads them. A value that does not read throws, and
446 *  the caller holds every namespace (fail closed), as for the suspended units. */
447export async function readPaused(kv: KVNamespace): Promise<Record<string, Paused>> {
448  const raw = await kv.get(PAUSED_KEY);
449  if (!raw) return {};
450  const parsed: unknown = JSON.parse(raw);
451  if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) throw new Error(`${PAUSED_KEY} is not a map of namespaces`);
452  return parsed as Record<string, Paused>;
453}
454
455async function readLifted(kv: KVNamespace): Promise<Record<string, string>> {
456  const raw = await kv.get(LIFTED_KEY);
457  return raw ? (JSON.parse(raw) as Record<string, string>) : {};
458}
459
460function reportFrom(counts: readonly SafetyCounts[], paused: Readonly<Record<string, Paused>>, lifted: Readonly<Record<string, string>>, now: Date): OpsMergeSafety {
461  const rows: OpsMergeSafetyRow[] = counts
462    .map((c) => ({
463      namespace: c.namespace,
464      auto_merged: c.auto_merged,
465      approved: c.approved,
466      deploys_verified: c.deploys_verified,
467      rolled_back: c.rolled_back,
468      rollback_failed: c.rollback_failed,
469      reverts: c.reverts,
470      bad_auto_7d: windowed(c, lifted[c.namespace], now).bad7,
471      paused: paused[c.namespace] ?? null,
472    }))
473    .sort((a, b) => a.namespace.localeCompare(b.namespace));
474  return { generated: now.toISOString(), namespaces: rows };
475}
476
477export interface SafetyTickReport {
478  ran: boolean;
479  note: string;
480  newly_paused: string[];
481}
482
483/** One pass: count the week, pause a namespace the counts condemn, store the report.
484 *  `namespaces` are those the signed policy covers (empty when it did not load). */
485export async function safetyTick(env: Env, now: Date, namespaces: readonly string[] = []): Promise<SafetyTickReport> {
486  const before = await readPaused(env.APP_KV);
487  const lifted = await readLifted(env.APP_KV);
488  const counts = await readSafetyCounts(env, now, namespaces);
489  const added = judgeShutoff(counts, before, lifted, now);
490  const since = now.toISOString();
491  const names = Object.keys(added);
492  let paused: Record<string, Paused> = { ...before };
493  if (names.length) {
494    // Read again just before writing and add only the new namespaces to that value, so a
495    // resume that landed while the counts were read is not written back over. KV has no
496    // compare-and-set: this narrows the window and does not close it.
497    paused = { ...(await readPaused(env.APP_KV)), ...Object.fromEntries(Object.entries(added).map(([ns, reason]) => [ns, { since, reason }])) };
498    // The pause first: if the audit row then fails, the namespace is held and the next
499    // tick finds it already paused and writes no row, which costs a row, not the hold.
500    await env.APP_KV.put(PAUSED_KEY, JSON.stringify(paused));
501    await env.DB.batch(
502      names.map((ns) =>
503        auditStatement(env.DB, ACTOR, "auto-merge-paused", ns, null, { reason: added[ns], since, counts: reportFrom(counts, paused, lifted, now).namespaces.find((r) => r.namespace === ns) })
504      )
505    );
506  }
507  const report = reportFrom(counts, paused, lifted, now);
508  const previous = await env.APP_KV.get(REPORT_KEY);
509  const stable = (r: OpsMergeSafety) => JSON.stringify({ ...r, generated: "" });
510  let changed = true;
511  try {
512    changed = !previous || stable(JSON.parse(previous) as OpsMergeSafety) !== stable(report);
513  } catch {
514    // An unreadable report is replaced.
515  }
516  if (changed) await env.APP_KV.put(REPORT_KEY, JSON.stringify(report));
517  if (!names.length && !changed) return { ran: false, note: "no change", newly_paused: [] };
518  return {
519    ran: true,
520    note: `${report.namespaces.length} namespace(s) counted${names.length ? `; paused: ${names.map((n) => `${n} (${added[n]})`).join(", ")}` : ""}`,
521    newly_paused: names,
522  };
523}
524
525export interface MergeSafetyTickReport {
526  ran: boolean;
527  note: string;
528  newly_paused: string[];
529}
530
531/** What the tick calls: the revert pass (half-hourly), then the meter and the shutoff. */
532export async function mergeSafetyTick(env: Env, now: Date): Promise<MergeSafetyTickReport> {
533  // The policy is loaded (and its signature checked) once for both passes.
534  const loaded = await loadMergePolicy(env);
535  const reverts = await revertsTick(env, now, loaded);
536  const safety = await safetyTick(env, now, "policy" in loaded ? loaded.policy.namespaces : []);
537  if (!reverts.ran && !safety.ran) return { ran: false, note: safety.note, newly_paused: [] };
538  return { ran: true, note: [reverts.recorded.length ? reverts.note : null, safety.ran ? safety.note : null].filter(Boolean).join("; "), newly_paused: safety.newly_paused };
539}
540
541/** The meter as the status tool and the Portal read it, or null before the first pass.
542 *  An unreadable report reads as none. */
543export async function readSafetyReport(kv: KVNamespace): Promise<OpsMergeSafety | null> {
544  const raw = await kv.get(REPORT_KEY).catch(() => null);
545  if (!raw) return null;
546  try {
547    const parsed = JSON.parse(raw) as OpsMergeSafety;
548    return Array.isArray(parsed.namespaces) ? parsed : null;
549  } catch {
550    return null;
551  }
552}
553
554// ---------------------------------------------------------------------------
555// 6. Resume
556
557/** What lifting a namespace's pause will do, or why it cannot. Reads only. */
558export async function planResume(kv: KVNamespace, rawNamespace: string | undefined): Promise<{ ok: true; namespace: string; paused: Paused } | { ok: false; refusal: string }> {
559  const namespace = (rawNamespace ?? "").trim();
560  if (!/^[a-z0-9-]+$/.test(namespace)) return { ok: false, refusal: `automerge_resume needs a namespace, for example 'capsid'; got '${rawNamespace ?? ""}'.` };
561  const paused = (await readPaused(kv))[namespace];
562  if (!paused) return { ok: false, refusal: `auto-merge for ${namespace} is not paused, so there is nothing to lift.` };
563  return { ok: true, namespace, paused };
564}
565
566/** Lift a namespace's pause: the tick may merge its pull requests unattended again, and
567 *  the rollbacks and reverts before now count against it no more. Audited. */
568export async function resumeNamespace(env: Env, actor: string, rawNamespace: string | undefined, reason: string, now: Date): Promise<{ ok: true; summary: string } | { ok: false; refusal: string }> {
569  const plan = await planResume(env.APP_KV, rawNamespace);
570  if (!plan.ok) return plan;
571  const lifted = { ...(await readLifted(env.APP_KV)), [plan.namespace]: now.toISOString() };
572  await env.APP_KV.put(LIFTED_KEY, JSON.stringify(lifted));
573  // Read again just before writing, so a pause of another namespace that landed since the
574  // plan is kept. KV has no compare-and-set: this narrows the window and does not close it.
575  const rest = await readPaused(env.APP_KV);
576  delete rest[plan.namespace];
577  await env.APP_KV.put(PAUSED_KEY, JSON.stringify(rest));
578  await env.DB.batch([auditStatement(env.DB, actor, "auto-merge-resumed", plan.namespace, null, { namespace: plan.namespace, reason, was: plan.paused })]);
579  return { ok: true, summary: `Lifted the pause on auto-merge for ${plan.namespace}. The tick may merge its pull requests unattended again; new bad merges pause it again.` };
580}