feat(checks): durable verdict delivery — bucketed positions, quota reserve, reconcile sweep

apps/dispatcher/src/admission-summary.test.ts

@@ -2,25 +2,86 @@
 // formatting, exercised with fixed inputs like failure-summary.test.ts.
 
 import { describe, expect, it } from "vitest";
-import { queuedSummary, serialQueuedSummary } from "./admission-summary";
+import {
+  nextPositionReport,
+  POSITION_UPDATE_EVERY_MS,
+  type PositionReport,
+  positionBucket,
+  queuedSummary,
+  serialQueuedSummary,
+} from "./admission-summary";
+
+describe("positionBucket", () => {
+  it("places a waiter with nobody ahead in bucket 1", () => {
+    expect(positionBucket(0)).toBe("1");
+  });
+
+  it("buckets places 2-5, 6-20 and over 20 at their boundaries", () => {
+    expect([1, 4, 5, 19, 20, 100].map(positionBucket)).toEqual([
+      "2-5",
+      "2-5",
+      "6-20",
+      "6-20",
+      "over 20",
+      "over 20",
+    ]);
+  });
+});
+
+describe("nextPositionReport", () => {
+  const T0 = 1_000_000;
+
+  it("writes the first position it sees", () => {
+    expect(nextPositionReport(undefined, 7, T0)).toEqual({ bucket: "6-20", at: T0 });
+  });
+
+  it("writes nothing while the bucket holds, however long", () => {
+    const last = { bucket: "6-20" as const, at: T0 };
+    expect(nextPositionReport(last, 6, T0 + 10 * POSITION_UPDATE_EVERY_MS)).toBeUndefined();
+  });
+
+  it("holds a crossed bucket until a minute has passed since the last write", () => {
+    const last = { bucket: "6-20" as const, at: T0 };
+    expect(nextPositionReport(last, 3, T0 + POSITION_UPDATE_EVERY_MS - 1)).toBeUndefined();
+    expect(nextPositionReport(last, 3, T0 + POSITION_UPDATE_EVERY_MS)).toEqual({
+      bucket: "2-5",
+      at: T0 + POSITION_UPDATE_EVERY_MS,
+    });
+  });
+
+  it("bounds a draining queue of 30 polled every 20s to one write a minute at most", () => {
+    let last: PositionReport | undefined;
+    let writes = 0;
+    for (let i = 0; i <= 30; i++) {
+      const next = nextPositionReport(last, 30 - i, T0 + i * 20_000);
+      if (next !== undefined) {
+        writes += 1;
+        last = next;
+      }
+    }
+    // over 20 → 6-20 → 2-5 → 1: one write per bucket.
+    expect(writes).toBe(4);
+  });
+});
 
 describe("queuedSummary", () => {
   // 2026-06-05T12:34:00.000Z
   const TIMES_OUT_AT = Date.UTC(2026, 5, 5, 12, 34, 0);
 
-  it("renders position, pool usage, and the UTC give-up time", () => {
-    expect(queuedSummary(3, 16, 16, TIMES_OUT_AT)).toBe(
-      "Queued — waiting for a sandbox slot behind 3 runs (16/16 in use); times out 12:34 UTC",
+  it("renders the bucket, pool usage, and the UTC give-up time", () => {
+    expect(queuedSummary("2-5", 16, 16, TIMES_OUT_AT)).toBe(
+      "Queued — waiting for a sandbox slot, 2–5 in line (16/16 in use); times out 12:34 UTC",
     );
   });
 
-  it("singularises a one-run queue", () => {
-    expect(queuedSummary(1, 16, 16, TIMES_OUT_AT)).toContain("behind 1 run (");
+  it("names bucket 1 as next in line and the top bucket as over 20th", () => {
+    expect(queuedSummary("1", 16, 16, TIMES_OUT_AT)).toContain("next in line (");
+    expect(queuedSummary("over 20", 16, 16, TIMES_OUT_AT)).toContain("over 20th in line (");
   });
 
   it("zero-pads single-digit hours and minutes", () => {
     // 2026-06-05T03:05:00.000Z
-    expect(queuedSummary(0, 16, 16, Date.UTC(2026, 5, 5, 3, 5, 0))).toContain(
+    expect(queuedSummary("1", 16, 16, Date.UTC(2026, 5, 5, 3, 5, 0))).toContain(
       "times out 03:05 UTC",
     );
   });

apps/dispatcher/src/admission-summary.ts

@@ -7,27 +7,66 @@
 // and so cannot load under plain Node + Vitest), mirroring
 // failure-summary.ts.
 
+/**
+ * A waiter's place in line, coarsened. Place 1 is next in line (no live
+ * queued run ahead). The check-run shows the bucket, so a waiter writes only
+ * when its bucket changes.
+ */
+export type PositionBucket = "1" | "2-5" | "6-20" | "over 20";
+
+/** The bucket for a waiter with `ahead` live queued runs ahead of it. */
+export const positionBucket = (ahead: number): PositionBucket => {
+  const place = Math.max(0, ahead) + 1;
+  return place === 1 ? "1" : place <= 5 ? "2-5" : place <= 20 ? "6-20" : "over 20";
+};
+
+/** The shortest gap between two queue-position writes of one waiter. */
+export const POSITION_UPDATE_EVERY_MS = 60_000;
+
+/** The last queue position a waiter wrote to its check-run. */
+export type PositionReport = { readonly bucket: PositionBucket; readonly at: number };
+
+/**
+ * The position write a waiter makes at `now`, or `undefined` for none: a write
+ * happens when the bucket differs from the last one written and at least
+ * `POSITION_UPDATE_EVERY_MS` has passed since that write. A bucket crossed
+ * inside the minute is written on the first poll after it.
+ */
+export const nextPositionReport = (
+  last: PositionReport | undefined,
+  ahead: number,
+  now: number,
+): PositionReport | undefined => {
+  const bucket = positionBucket(ahead);
+  if (last !== undefined && (last.bucket === bucket || now - last.at < POSITION_UPDATE_EVERY_MS)) {
+    return undefined;
+  }
+  return { bucket, at: now };
+};
+
 /**
  * The "queued" line the admission gate posts to the in-progress check-run —
- * only when the position CHANGES, to bound GitHub API writes.
+ * only when the position bucket changes, to bound GitHub API writes.
  *
- * @param position   live queued runs ahead of this one.
+ * @param bucket     the waiter's place in line, coarsened (`positionBucket`).
  * @param poolBusy   admission slots in use in the run's pool.
  * @param cap        the pool's slot cap (`ADMISSION_CAP`).
  * @param timesOutAt ms-epoch the dispatch-age ceiling lapses (enqueue + max age).
  */
 export const queuedSummary = (
-  position: number,
+  bucket: PositionBucket,
   poolBusy: number,
   cap: number,
   timesOutAt: number,
 ): string => {
   const d = new Date(timesOutAt);
   const hh = String(d.getUTCHours()).padStart(2, "0");
   const mm = String(d.getUTCMinutes()).padStart(2, "0");
+  const place =
+    bucket === "1" ? "next in line" : bucket === "over 20" ? "over 20th in line" : `${bucket.replace("-", "–")} in line`;
   return (
-    `Queued — waiting for a sandbox slot behind ${position} ` +
-    `run${position === 1 ? "" : "s"} (${poolBusy}/${cap} in use); ` +
+    `Queued — waiting for a sandbox slot, ${place} ` +
+    `(${poolBusy}/${cap} in use); ` +
     `times out ${hh}:${mm} UTC`
   );
 };

apps/dispatcher/src/cron-parity.test.ts

@@ -16,6 +16,7 @@
 
 import { readFileSync } from "node:fs";
 import { describe, expect, it } from "vitest";
+import { DELIVERY_SWEEP_CRON } from "./delivery-sweep";
 import { runNames, schedulesByCron, scheduledRuns } from "./registry";
 
 /**
@@ -47,7 +48,15 @@ describe("cron parity", () => {
     // silently yields no crons passes every assertion below while checking
     // nothing.
     expect(armed.length).toBeGreaterThan(0);
-    expect(armed.filter((cron) => schedulesByCron(cron).length === 0)).toEqual([]);
+    // The reconcile sweep's cron is the one armed expression that binds no
+    // run: the scheduled handler routes it to the sweep.
+    expect(
+      armed.filter((cron) => cron !== DELIVERY_SWEEP_CRON && schedulesByCron(cron).length === 0),
+    ).toEqual([]);
+  });
+
+  it("arms the verdict reconcile sweep", () => {
+    expect(readArmedCrons()).toContain(DELIVERY_SWEEP_CRON);
   });
 
   it("every cron a scheduled run declares is armed unless deliberately left off", () => {

apps/dispatcher/src/delivery-sweep.test.ts

@@ -0,0 +1,182 @@
+// Tests for the verdict reconcile sweep — driven through in-memory ports.
+
+import type { CheckConcludeOptions, CheckDelivery } from "@fractalboxdev/flare-dispatch-core";
+import { describe, expect, it, vi } from "vitest";
+import {
+  conclusionFor,
+  DELIVERY_GRACE_MS,
+  DELIVERY_MAX_AGE_MS,
+  DELIVERY_SWEEP_LIMIT,
+  type DeliverySweepPorts,
+  sweepUndelivered,
+  type UndeliveredRow,
+} from "./delivery-sweep";
+
+const NOW = Date.UTC(2026, 9, 22, 12, 0, 0);
+
+const row = (over: Partial<UndeliveredRow> = {}): UndeliveredRow => ({
+  id: "exec-1",
+  run: "oxlint",
+  repo: "owner/name",
+  sha: "abc123",
+  status: "success",
+  attempt: 1,
+  inputJson: "{}",
+  checkRunId: "77",
+  installationId: 1,
+  completedAt: NOW - DELIVERY_GRACE_MS - 1,
+  ...over,
+});
+
+/** In-memory ports over `rows`; `answer` decides each conclude. */
+const makePorts = (
+  rows: UndeliveredRow[],
+  answer: (installationId: number, opts: CheckConcludeOptions) => CheckDelivery = (_, opts) => ({
+    _tag: "delivered",
+    checkRunId: opts.checkRunId === "noop" ? "9001" : opts.checkRunId,
+  }),
+) => {
+  const concluded: { installationId: number; opts: CheckConcludeOptions }[] = [];
+  const marked: { id: string; checkRunId: string; at: number }[] = [];
+  const refusedRows: { id: string; at: number }[] = [];
+  const listUndelivered = vi.fn<DeliverySweepPorts["listUndelivered"]>(
+    async ({ completedBefore, completedAfter, limit }) =>
+      rows
+        .filter((r) => r.completedAt < completedBefore && r.completedAt >= completedAfter)
+        .sort((a, b) => a.completedAt - b.completedAt)
+        .slice(0, limit),
+  );
+  const ports: DeliverySweepPorts = {
+    listUndelivered,
+    conclude: async (installationId, opts) => {
+      concluded.push({ installationId, opts });
+      return answer(installationId, opts);
+    },
+    markDelivered: async (id, checkRunId, at) => {
+      marked.push({ id, checkRunId, at });
+    },
+    markRefused: async (id, at) => {
+      refusedRows.push({ id, at });
+    },
+    now: () => NOW,
+  };
+  return { ports, concluded, marked, refusedRows, listUndelivered };
+};
+
+describe("sweepUndelivered", () => {
+  it("asks for rows past the grace period and inside the max age, oldest first, bounded", async () => {
+    const { ports, listUndelivered } = makePorts([]);
+    await sweepUndelivered(ports);
+    expect(listUndelivered).toHaveBeenCalledWith({
+      completedBefore: NOW - DELIVERY_GRACE_MS,
+      completedAfter: NOW - DELIVERY_MAX_AGE_MS,
+      limit: DELIVERY_SWEEP_LIMIT,
+    });
+  });
+
+  it("re-sends oldest first and stamps each delivered row", async () => {
+    const { ports, concluded, marked } = makePorts([
+      row({ id: "newer", completedAt: NOW - DELIVERY_GRACE_MS - 1 }),
+      row({ id: "older", completedAt: NOW - DELIVERY_GRACE_MS - 60_000 }),
+    ]);
+
+    const result = await sweepUndelivered(ports);
+
+    expect(concluded.map((c) => c.opts.output?.summary.includes("older") ?? false)).toEqual([
+      true,
+      false,
+    ]);
+    expect(marked.map((m) => m.id)).toEqual(["older", "newer"]);
+    expect(result).toEqual({ delivered: 2, undelivered: 0, refused: 0, deferred: 0 });
+  });
+
+  it("records the check-run created at conclusion for a row whose opener degraded", async () => {
+    const { ports, concluded, marked } = makePorts([row({ checkRunId: "noop" })]);
+    await sweepUndelivered(ports);
+    expect(concluded[0]!.opts.checkRunId).toBe("noop");
+    expect(marked).toEqual([{ id: "exec-1", checkRunId: "9001", at: NOW }]);
+  });
+
+  it("stops an installation at its first refusal and keeps sending for the others", async () => {
+    const { ports, concluded, marked } = makePorts(
+      [
+        row({ id: "a1", installationId: 1, completedAt: NOW - DELIVERY_GRACE_MS - 3 }),
+        row({ id: "b1", installationId: 2, completedAt: NOW - DELIVERY_GRACE_MS - 2 }),
+        row({ id: "a2", installationId: 1, completedAt: NOW - DELIVERY_GRACE_MS - 1 }),
+      ],
+      (installationId, opts) =>
+        installationId === 1
+          ? { _tag: "undelivered", reason: "installation quota exhausted" }
+          : { _tag: "delivered", checkRunId: opts.checkRunId },
+    );
+    vi.spyOn(console, "warn").mockImplementation(() => {});
+
+    const result = await sweepUndelivered(ports);
+
+    expect(concluded.map((c) => c.installationId)).toEqual([1, 2]);
+    expect(marked.map((m) => m.id)).toEqual(["b1"]);
+    expect(result).toEqual({ delivered: 1, undelivered: 1, refused: 0, deferred: 1 });
+  });
+
+  it("takes a refused row out of the queue without holding up its installation", async () => {
+    const { ports, concluded, marked, refusedRows } = makePorts(
+      [
+        row({ id: "gone", installationId: 1, sha: "dead", completedAt: NOW - DELIVERY_GRACE_MS - 2 }),
+        row({ id: "next", installationId: 1, completedAt: NOW - DELIVERY_GRACE_MS - 1 }),
+      ],
+      (_, opts) =>
+        opts.sha === "dead"
+          ? { _tag: "refused", reason: "HTTP 404" }
+          : { _tag: "delivered", checkRunId: opts.checkRunId },
+    );
+    vi.spyOn(console, "warn").mockImplementation(() => {});
+    const result = await sweepUndelivered(ports);
+
+    expect(concluded.map((c) => c.opts.sha)).toEqual(["dead", "abc123"]);
+    expect(refusedRows).toEqual([{ id: "gone", at: NOW }]);
+    expect(marked.map((m) => m.id)).toEqual(["next"]);
+    expect(result).toEqual({ delivered: 1, undelivered: 0, refused: 1, deferred: 0 });
+  });
+
+  it("counts a rejected conclude as undelivered and a failed stamp as delivered", async () => {
+    const { ports } = makePorts([row({ id: "x", installationId: 1 }), row({ id: "y", installationId: 2 })]);
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
+    const result = await sweepUndelivered({
+      ...ports,
+      conclude: async (installationId, opts) => {
+        if (installationId === 1) throw new Error("boom");
+        return { _tag: "delivered", checkRunId: opts.checkRunId };
+      },
+      markDelivered: async () => {
+        throw new Error("d1 down");
+      },
+    });
+    expect(result).toEqual({ delivered: 1, undelivered: 1, refused: 0, deferred: 0 });
+    expect(warn).toHaveBeenCalled();
+  });
+});
+
+describe("conclusionFor", () => {
+  it("names, titles and concludes the check-run as the workflow does", () => {
+    const opts = conclusionFor(
+      row({ run: "check", inputJson: JSON.stringify({ checkLabel: "lint" }), attempt: 2, status: "skipped" }),
+    );
+    expect(opts).toMatchObject({
+      repo: "owner/name",
+      sha: "abc123",
+      name: "flare-dispatch/check:lint",
+      checkRunId: "77",
+      conclusion: "neutral",
+      output: { title: "flare-dispatch/check:lint (attempt 2)" },
+    });
+    expect(opts.output?.summary).toContain("⊘ check — skipped.");
+  });
+
+  it("names a row with unreadable inputs without a label", () => {
+    expect(conclusionFor(row({ inputJson: "{not json" })).name).toBe("flare-dispatch/oxlint");
+  });
+
+  it("concludes a failed execution as failure", () => {
+    expect(conclusionFor(row({ status: "failure" })).conclusion).toBe("failure");
+  });
+});