feat(dispatch): opt-in dispatch, counted supersessions and a two-tier contextful gate

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

@@ -67,7 +67,7 @@ describe("leaveQueueForEndedParent", () => {
         { executionId: "part-a", parentExecutionId: "gate", parentStatus: "failure" },
         {
           stepDo: (name, body) => Effect.promise(async () => { events.push(`step:${name}`); return body(); }),
-          store: { cancel: (ids) => Effect.sync(() => { events.push(`cancel:${ids.join(",")}`); }) },
+          store: { cancel: (ids) => Effect.sync(() => { events.push(`cancel:${ids.join(",")}`); return { admittedSlots: 0 }; }) },
           concludeCancelled: (summary) => Effect.sync(() => { events.push(`conclude:${summary}`); }),
         },
       ),

apps/dispatcher/src/dashboard.test.ts

@@ -27,6 +27,7 @@ const baseRow = (over: Partial<DashboardRow> = {}): DashboardRow => ({
   logsUrl: null,
   demosUrl: null,
   selfHealPrUrl: null,
+  supersededBy: null,
   ...over,
 });
 
@@ -114,6 +115,24 @@ describe("renderDashboard", () => {
     expect(html).not.toContain("badge selfheal");
   });
 
+  it("labels a cancelled execution with the execution that superseded it", () => {
+    const html = renderDashboard(data([baseRow({ status: "cancelled", supersededBy: "gate-newer" })]));
+    expect(html).toContain('superseded by <span class="sha">gate-newer</span>');
+  });
+
+  it("labels a cancelled execution a pull-request close superseded", () => {
+    const html = renderDashboard(
+      data([baseRow({ status: "cancelled", supersededBy: "pr-closed:1000:contextful-gate:o/r#7" })]),
+    );
+    expect(html).toContain("pull request closed");
+    expect(html).not.toContain("superseded by");
+  });
+
+  it("labels no execution that is not cancelled", () => {
+    const html = renderDashboard(data([baseRow({ status: "running", supersededBy: "gate-newer" })]));
+    expect(html).not.toContain("superseded by");
+  });
+
   it("escapes HTML in execution fields", () => {
     const html = renderDashboard(data([baseRow({ repo: "<script>" })]));
     expect(html).not.toContain("<script>");
@@ -448,6 +467,22 @@ describe("GET /v1/dashboard.json — SPA feed", () => {
     expect(body.rows[0]?.selfHealPrUrl).toBeNull();
   });
 
+  it("carries the superseding execution of a cancelled row, null otherwise", async () => {
+    const { env } = fixture([
+      execRow({ id: "old", status: "cancelled", superseded_by: "new", started_at: 1000 }),
+      execRow({ id: "new", status: "running", started_at: 2000 }),
+    ]);
+    const res = await handleRequest(
+      new Request("https://dispatcher.example.com/v1/dashboard.json"),
+      env,
+    );
+    const body = (await res.json()) as { rows: { id: string; supersededBy: string | null }[] };
+    expect(body.rows.map((row) => [row.id, row.supersededBy])).toEqual([
+      ["new", null],
+      ["old", "new"],
+    ]);
+  });
+
   it("sets selfHealPrUrl for a self-heal-pr run that opened a fix PR (prStaged)", async () => {
     const { env } = fixture([
       execRow({

apps/dispatcher/src/dashboard.ts

@@ -11,6 +11,8 @@
 // I/O, so it stays trivially testable (dashboard.test.ts). The route that feeds
 // it (query-param parsing, the D1 reads, the Access gate) is wired in http-app.ts.
 
+import { isCloseMarker } from "./pr-close";
+
 /** One execution row, pre-resolved with its tokened viewer links. */
 export interface DashboardRow {
   readonly id: string;
@@ -38,6 +40,12 @@ export interface DashboardRow {
    * — a non-null value means flare-dispatch drove a PR to fix a bug here.
    */
   readonly selfHealPrUrl: string | null;
+  /**
+   * The execution that superseded this one, or a pull-request close marker,
+   * from `run_supersessions.superseded_by`; `null` when nothing superseded it.
+   * A `cancelled` row with a value is labelled with it.
+   */
+  readonly supersededBy: string | null;
 }
 
 export interface DashboardData {
@@ -244,6 +252,15 @@ const STYLE = `
   footer a { margin-right: 1rem; }
 `;
 
+/** "superseded by <id>" (or "pull request closed") under a cancelled row's badge. */
+const supersededLabel = (row: DashboardRow): string => {
+  if (row.status !== "cancelled" || row.supersededBy === null) return "";
+  const text = isCloseMarker(row.supersededBy)
+    ? "pull request closed"
+    : `superseded by <span class="sha">${esc(row.supersededBy)}</span>`;
+  return ` <span class="when superseded">${text}</span>`;
+};
+
 const renderRow = (row: DashboardRow, nowMs: number): string => {
   const links: string[] = [];
   if (row.logsUrl !== null) {
@@ -267,7 +284,7 @@ const renderRow = (row: DashboardRow, nowMs: number): string => {
       ? ` <a class="badge selfheal" href="${esc(row.selfHealPrUrl)}" title="View flare-dispatch self-heal fix PRs">🩹 self-heal</a>`
       : "";
   return `<tr${clickable ? ' class="rowlink"' : ""}>
-    <td><span class="badge ${badgeClass(row.status)}">${esc(row.status)}</span></td>
+    <td><span class="badge ${badgeClass(row.status)}">${esc(row.status)}</span>${supersededLabel(row)}</td>
     <td class="run">${runCell}${selfHealPill}</td>
     <td><span class="repo">${esc(row.repo)}</span> <span class="sha">${esc(shortSha(row.sha))}</span></td>
     <td class="when">${esc(relativeTime(row.startedAt, nowMs))}</td>

apps/dispatcher/src/executions-read.ts

@@ -38,6 +38,12 @@ export type ExecutionRow = {
   readonly output_tokens: number | null;
   readonly vcpu_seconds: number | null;
   readonly model: string | null;
+  // --- Supersession (infra/migrations/0010; `listExecutions` only) -----------
+  /**
+   * The execution (or pull-request close marker) that superseded this one,
+   * from its `run_supersessions` row; NULL when nothing did or the row aged out.
+   */
+  readonly superseded_by?: string | null;
 };
 
 /** One row of the `steps` table. */
@@ -79,7 +85,8 @@ export type ListFilters = {
 
 /**
  * List executions newest-first, with optional `run`/`repo`/`status` filters and
- * a `(started_at, id)` keyset cursor. Returns at most `limit` rows.
+ * a `(started_at, id)` keyset cursor, each with the `superseded_by` of its
+ * `run_supersessions` row. Returns at most `limit` rows.
  */
 export const listExecutions = async (
   db: D1Database,
@@ -111,7 +118,11 @@ export const listExecutions = async (
     }
   }
   const whereSql = where.length > 0 ? `WHERE ${where.join(" AND ")}` : "";
-  const sql = `SELECT * FROM executions ${whereSql}
+  // The filter columns exist only on `executions`, so they stay unqualified.
+  const sql = `SELECT executions.*, run_supersessions.superseded_by
+                 FROM executions
+                 LEFT JOIN run_supersessions ON run_supersessions.execution_id = executions.id
+               ${whereSql}
                ORDER BY started_at DESC
                LIMIT ?`;
   binds.push(filters.limit);

apps/dispatcher/src/http-app.ts

@@ -323,6 +323,7 @@ const dashboardData = Effect.gen(function* () {
         logsUrl: Option.getOrNull(logsUrl),
         demosUrl: Option.getOrNull(demosUrl),
         selfHealPrUrl,
+        supersededBy: row.superseded_by ?? null,
       } satisfies DashboardRow;
     }),
   );

apps/dispatcher/src/opt-in.test.ts

@@ -0,0 +1,73 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { resolveOptIn } from "./opt-in";
+import { makeFakeKv } from "./test-helpers";
+
+const KEYS = ["check.command:acme/widgets:docs", "check.command:acme/widgets"] as const;
+
+afterEach(() => {
+  vi.restoreAllMocks();
+});
+
+describe("resolveOptIn", () => {
+  it("answers not-configured when no key is present", async () => {
+    const kv = makeFakeKv();
+    expect(await resolveOptIn({ configKv: kv.binding, runName: "check", keys: () => KEYS })).toBe(
+      "not-configured",
+    );
+  });
+
+  it("dispatches when any one key is present, the least specific included", async () => {
+    const kv = makeFakeKv();
+    kv.store.set("check.command:acme/widgets", "pnpm lint");
+    expect(await resolveOptIn({ configKv: kv.binding, runName: "check", keys: () => KEYS })).toBe(
+      "dispatch",
+    );
+  });
+
+  it("treats a present key as opted in whatever its value — the run body judges the value", async () => {
+    const kv = makeFakeKv();
+    kv.store.set("check.command:acme/widgets", "");
+    expect(await resolveOptIn({ configKv: kv.binding, runName: "check", keys: () => KEYS })).toBe(
+      "dispatch",
+    );
+  });
+
+  it("dispatches without a read when the inputs carry the command", async () => {
+    const get = vi.fn();
+    const configKv = { get } as unknown as KVNamespace;
+    expect(await resolveOptIn({ configKv, runName: "check", keys: () => [] })).toBe("dispatch");
+    expect(get).not.toHaveBeenCalled();
+  });
+
+  it("dispatches without computing keys on a deploy with no CONFIG_KV", async () => {
+    const keys = vi.fn(() => KEYS);
+    expect(await resolveOptIn({ configKv: undefined, runName: "check", keys })).toBe("dispatch");
+    expect(keys).not.toHaveBeenCalled();
+  });
+
+  it("fails open on a rejected read and logs the run and keys", async () => {
+    const error = vi.spyOn(console, "error").mockImplementation(() => {});
+    const configKv = {
+      get: async () => {
+        throw new Error("kv unavailable");
+      },
+    } as unknown as KVNamespace;
+    expect(await resolveOptIn({ configKv, runName: "check", keys: () => KEYS })).toBe("dispatch");
+    expect(String(error.mock.calls[0]?.[0])).toContain('run="check"');
+    expect(String(error.mock.calls[0]?.[0])).toContain("kv unavailable");
+  });
+
+  it("fails open when the run's optIn throws", async () => {
+    const error = vi.spyOn(console, "error").mockImplementation(() => {});
+    const kv = makeFakeKv();
+    const verdict = await resolveOptIn({
+      configKv: kv.binding,
+      runName: "check",
+      keys: () => {
+        throw new Error("input has no repo");
+      },
+    });
+    expect(verdict).toBe("dispatch");
+    expect(String(error.mock.calls[0]?.[0])).toContain("input has no repo");
+  });
+});

apps/dispatcher/src/opt-in.ts

@@ -0,0 +1,46 @@
+// FlareDispatch Dispatcher — receiver-side opt-in.
+//
+// A run may declare `optIn: ({ input }) => keys` (packages/core define-run.ts):
+// the CONFIG_KV keys whose presence means the repo opted in. The webhook
+// receiver reads them before it creates a Workflow, so a repo that never
+// configured the run costs no Workflow, check-run, admission slot or container
+// (specs/adr/0007-cost-efficient-execution.md § Dispatch only work).
+//
+// The read fails open. A KV error, a throwing `optIn`, or a deploy without
+// CONFIG_KV dispatches as if the run declared no opt-in: the run body still
+// resolves its own command and no-ops when there is none, so a failed read
+// costs one wasted run, never a configured repo's verdict.
+
+/** The receiver's decision for one triggered run. */
+export type OptInVerdict = "dispatch" | "not-configured";
+
+/**
+ * Decide whether to dispatch a run whose `optIn` names `keys`.
+ *
+ * No `configKv`, or `keys` empty (the inputs carry their own command) →
+ * `dispatch` with no read. Otherwise any present key → `dispatch`; none →
+ * `not-configured`. `keys` is a thunk so a throwing `optIn` lands in the same
+ * fail-open branch as a rejected read: it logs and answers `dispatch`, and
+ * never fails the delivery for the sibling triggers.
+ */
+export const resolveOptIn = async (args: {
+  readonly configKv: KVNamespace | undefined;
+  readonly runName: string;
+  readonly keys: () => readonly string[];
+}): Promise<OptInVerdict> => {
+  const { configKv, runName } = args;
+  if (configKv === undefined) return "dispatch";
+  let keys: readonly string[] = [];
+  try {
+    keys = args.keys();
+    if (keys.length === 0) return "dispatch";
+    const values = await Promise.all(keys.map((key) => configKv.get(key)));
+    return values.some((value) => value !== null) ? "dispatch" : "not-configured";
+  } catch (cause) {
+    const message = cause instanceof Error ? cause.message : String(cause);
+    console.error(
+      `[webhook] opt-in read failed run="${runName}" keys=${JSON.stringify(keys)}: ${message} — dispatching`,
+    );
+    return "dispatch";
+  }
+};