fix(queue): mark superseded webhook_events rows instead of leaving them stuck at 'queued' forever

src/db/repositories.ts

@@ -4954,7 +4954,11 @@ export async function recordWebhookEvent(
     installationId?: number | undefined;
     repositoryFullName?: string | undefined;
     payloadHash: string;
-    status: "queued" | "processed" | "error";
+    // "superseded": a coalescable delivery (e.g. a pr-refresh) whose queue row was overwritten by a later
+    // redelivery sharing the same job_key before either was claimed — written directly by the self-host queue
+    // backends (pg-queue.ts / sqlite-queue.ts) at coalesce time, not through this function, but included here so
+    // the full set of terminal statuses this column can hold is documented in one place (#audit-webhook-supersede-trace).
+    status: "queued" | "processed" | "error" | "superseded";
     errorSummary?: string;
   },
 ): Promise<void> {

src/selfhost/pg-queue.ts

@@ -819,6 +819,38 @@ export function createPgQueue(
     return due.length;
   }
 
+  // #audit-webhook-supersede-trace: best-effort, never blocks a coalesce on a write hiccup -- the row it marks
+  // is purely an audit trace (webhook_events), not the actual job data, so a failure here must not resurrect the
+  // "abort the whole enqueue" class of bug this whole issue exists to close. `oldPayload` is the row's payload
+  // BEFORE it gets overwritten by the coalesce; `incomingMessage` is what it's about to become. Only a
+  // github-webhook delivery has a webhook_events row at all (rag-index-repo etc. never do), and only when the
+  // superseded id genuinely differs from the surviving one (defense-in-depth against a same-id no-op).
+  async function markSupersededWebhookEvent(oldPayload: string, incomingMessage: JobMessage): Promise<void> {
+    let old: { type?: unknown; deliveryId?: unknown } | null;
+    try {
+      old = JSON.parse(oldPayload) as { type?: unknown; deliveryId?: unknown };
+    } catch {
+      return;
+    }
+    if (old?.type !== "github-webhook" || typeof old.deliveryId !== "string") return;
+    const incomingDeliveryId = incomingMessage.type === "github-webhook" ? incomingMessage.deliveryId : undefined;
+    if (old.deliveryId === incomingDeliveryId) return;
+    try {
+      await pool.query(
+        `UPDATE webhook_events SET status='superseded', processed_at=$2 WHERE delivery_id=$1 AND status='queued'`,
+        [old.deliveryId, new Date().toISOString()],
+      );
+    } catch (error) {
+      console.error(
+        JSON.stringify({
+          level: "error",
+          event: "webhook_supersede_mark_failed",
+          error: errorMessageWithCause(error),
+        }),
+      );
+    }
+  }
+
   async function enqueue(
     message: JobMessage,
     delaySeconds: number,
@@ -920,11 +952,17 @@ export function createPgQueue(
     if (key) {
       const existing = (
         await pool.query(
-          `SELECT id FROM ${TABLE} WHERE status='pending' AND job_key=$1 ORDER BY priority DESC, run_after DESC, id LIMIT 1`,
+          `SELECT id, payload FROM ${TABLE} WHERE status='pending' AND job_key=$1 ORDER BY priority DESC, run_after DESC, id LIMIT 1`,
           [key],
         )
-      ).rows[0] as { id: string } | undefined;
+      ).rows[0] as { id: string; payload: string } | undefined;
       if (existing) {
+        // #audit-webhook-supersede-trace: the row about to be overwritten below may itself be a github-webhook
+        // delivery (e.g. a "PR opened" pr-refresh coalesce) whose own webhook_events row was written as 'queued'
+        // BEFORE it ever reached this coalesce -- overwriting the payload here discards that delivery's id
+        // forever, so nothing would ever advance its webhook_events row past 'queued'. Mark it superseded FIRST,
+        // while the OLD payload (and its deliveryId) is still readable.
+        await markSupersededWebhookEvent(existing.payload, message);
         // See the supersededKeyPrefix branch above: created_at is preserved across a coalesced re-enqueue so the
         // maintenance trickle clock reflects genuine wait time, not the most recent re-request.
         await pool.query(

src/selfhost/sqlite-queue.ts

@@ -483,6 +483,38 @@ export function createSqliteQueue(
     }
   }
 
+  // #audit-webhook-supersede-trace: best-effort, never blocks a coalesce on a write hiccup -- the row it marks
+  // is purely an audit trace (webhook_events), not the actual job data, so a failure here must not resurrect the
+  // "abort the whole enqueue" class of bug this whole issue exists to close. `oldPayload` is the row's payload
+  // BEFORE it gets overwritten by the coalesce; `incomingMessage` is what it's about to become. Only a
+  // github-webhook delivery has a webhook_events row at all (rag-index-repo etc. never do), and only when the
+  // superseded id genuinely differs from the surviving one (defense-in-depth against a same-id no-op).
+  function markSupersededWebhookEvent(oldPayload: string, incomingMessage: JobMessage): void {
+    let old: { type?: unknown; deliveryId?: unknown } | null;
+    try {
+      old = JSON.parse(oldPayload) as { type?: unknown; deliveryId?: unknown };
+    } catch {
+      return;
+    }
+    if (old?.type !== "github-webhook" || typeof old.deliveryId !== "string") return;
+    /* v8 ignore next -- defensive: jobCoalesceKey partitions its key format strictly by message.type, so a
+     * github-webhook job_key can only ever be matched by another github-webhook message; the non-webhook arm
+     * is unreachable through this call site, not load-bearing. */
+    const incomingDeliveryId = incomingMessage.type === "github-webhook" ? incomingMessage.deliveryId : undefined;
+    if (old.deliveryId === incomingDeliveryId) return;
+    try {
+      driver.query(`UPDATE webhook_events SET status='superseded', processed_at=? WHERE delivery_id=? AND status='queued'`, [new Date().toISOString(), old.deliveryId]);
+    } catch (error) {
+      console.error(
+        JSON.stringify({
+          level: "error",
+          event: "webhook_supersede_mark_failed",
+          error: errorMessageWithCause(error),
+        }),
+      );
+    }
+  }
+
   function enqueue(message: JobMessage, delaySeconds: number): void {
     const now = Date.now();
     const payload = JSON.stringify(message);
@@ -572,10 +604,16 @@ export function createSqliteQueue(
     }
     if (key) {
       const existing = driver.query(
-        `SELECT id FROM ${TABLE} WHERE status='pending' AND job_key=? ORDER BY priority DESC, run_after DESC, id LIMIT 1`,
+        `SELECT id, payload FROM ${TABLE} WHERE status='pending' AND job_key=? ORDER BY priority DESC, run_after DESC, id LIMIT 1`,
         [key],
-      ).rows[0] as { id: number } | undefined;
+      ).rows[0] as { id: number; payload: string } | undefined;
       if (existing) {
+        // #audit-webhook-supersede-trace: the row about to be overwritten below may itself be a github-webhook
+        // delivery (e.g. a "PR opened" pr-refresh coalesce) whose own webhook_events row was written as 'queued'
+        // BEFORE it ever reached this coalesce -- overwriting the payload here discards that delivery's id
+        // forever, so nothing would ever advance its webhook_events row past 'queued'. Mark it superseded FIRST,
+        // while the OLD payload (and its deliveryId) is still readable.
+        markSupersededWebhookEvent(existing.payload, message);
         // See the supersededKeyPrefix branch above: created_at is preserved across a coalesced re-enqueue so the
         // maintenance trickle clock reflects genuine wait time, not the most recent re-request.
         driver.query(

test/unit/selfhost-pg-queue.test.ts

@@ -558,6 +558,31 @@ describe("createPgQueue (durable #977)", () => {
     );
   });
 
+  it("REGRESSION (#audit-webhook-supersede-trace): marks the superseded delivery's webhook_events row instead of leaving it stuck at 'queued' forever", async () => {
+    const m = makePool();
+    const q = createPgQueue(m.pool, async () => undefined);
+    await q.init();
+    m.fn.mockResolvedValueOnce({ rows: [{ id: "existing", payload: JSON.stringify(ciWebhook("ci-1", "check_run")) }], rowCount: 1 });
+
+    await q.binding.send(ciWebhook("ci-2", "check_run"), { delaySeconds: 1 });
+
+    expect(m.pool.query).toHaveBeenCalledWith(
+      expect.stringContaining("UPDATE webhook_events SET status='superseded'"),
+      ["ci-1", expect.any(String)],
+    );
+  });
+
+  it("does NOT mark superseded when the coalesced-away job was not a github-webhook delivery (e.g. a scheduled sweep re-arm)", async () => {
+    const m = makePool();
+    const q = createPgQueue(m.pool, async () => undefined);
+    await q.init();
+    m.fn.mockResolvedValueOnce({ rows: [{ id: "existing", payload: JSON.stringify({ type: "refresh-registry", requestedBy: "schedule" }) }], rowCount: 1 });
+
+    await q.binding.send(msg("refresh-registry"));
+
+    expect(m.pool.query).not.toHaveBeenCalledWith(expect.stringContaining("UPDATE webhook_events SET status='superseded'"), expect.anything());
+  });
+
   it("does not reset created_at when coalescing a re-enqueue into an existing pending row (regression for #selfhost-runtime-drift)", async () => {
     // created_at anchors the maintenance trickle's age clock (maintenance-admission.ts). If a coalesced
     // re-enqueue reset it, a periodic scheduler re-requesting the same still-pending maintenance job faster

test/unit/selfhost-sqlite-queue.test.ts

@@ -871,6 +871,94 @@ describe("createSqliteQueue (durable #980)", () => {
     });
   });
 
+  it("REGRESSION (#audit-webhook-supersede-trace): marks a superseded pr-refresh delivery's webhook_events row instead of leaving it stuck at 'queued' forever", async () => {
+    const driver = makeDriver();
+    driver.exec(`
+      CREATE TABLE webhook_events (
+        delivery_id TEXT PRIMARY KEY,
+        event_name TEXT NOT NULL,
+        action TEXT,
+        installation_id INTEGER,
+        repository_full_name TEXT,
+        payload_hash TEXT NOT NULL,
+        status TEXT NOT NULL,
+        error_summary TEXT,
+        received_at TEXT NOT NULL,
+        processed_at TEXT
+      );
+    `);
+    driver.query(
+      "INSERT INTO webhook_events (delivery_id, event_name, payload_hash, status, received_at) VALUES (?, 'pull_request', 'hash-1', 'queued', '2026-07-06T00:00:00.000Z')",
+      ["pr-1"],
+    );
+    const q = createSqliteQueue(driver, async () => undefined);
+
+    await q.binding.send(prWebhook("pr-1"), { delaySeconds: 60 });
+    await q.binding.send(prWebhook("pr-2"), { delaySeconds: 1 }); // coalesces into pr-1's row, overwriting its payload
+
+    const rows = driver.query("SELECT payload FROM _selfhost_jobs ORDER BY id", []).rows as Array<{ payload: string }>;
+    expect(rows).toHaveLength(1); // coalesced into one row
+    expect(JSON.parse(rows[0]!.payload).deliveryId).toBe("pr-2"); // pr-2's payload survives -- pr-1's is gone from the queue
+    const event = driver.query("SELECT status FROM webhook_events WHERE delivery_id = ?", ["pr-1"]).rows[0] as { status: string } | undefined;
+    expect(event?.status).toBe("superseded"); // pr-1's trace row is marked, not left stuck at 'queued' forever
+  });
+
+  it("fails safe when the webhook_events table itself errors: the coalesce still completes and only logs a warning", async () => {
+    const driver = makeDriver();
+    // No webhook_events table -- the UPDATE inside markSupersededWebhookEvent will throw "no such table", which
+    // must be caught and logged, never allowed to abort the coalesce/enqueue itself.
+    const q = createSqliteQueue(driver, async () => undefined);
+    const errors = vi.spyOn(console, "error").mockImplementation(() => undefined);
+
+    await q.binding.send(prWebhook("pr-1"), { delaySeconds: 60 });
+    await q.binding.send(prWebhook("pr-2"), { delaySeconds: 1 });
+
+    const rows = driver.query("SELECT payload FROM _selfhost_jobs ORDER BY id", []).rows as Array<{ payload: string }>;
+    expect(rows).toHaveLength(1); // the coalesce itself still completed despite the missing table
+    expect(JSON.parse(rows[0]!.payload).deliveryId).toBe("pr-2");
+    expect(errors.mock.calls.some((call) => String(call[0]).includes("webhook_supersede_mark_failed"))).toBe(true);
+    errors.mockRestore();
+  });
+
+  it("does not mark superseded (no-op) when the coalesce is against the SAME deliveryId (defense-in-depth)", async () => {
+    const driver = makeDriver();
+    driver.exec(`
+      CREATE TABLE webhook_events (
+        delivery_id TEXT PRIMARY KEY,
+        event_name TEXT NOT NULL,
+        payload_hash TEXT NOT NULL,
+        status TEXT NOT NULL,
+        received_at TEXT NOT NULL
+      );
+    `);
+    driver.query(
+      "INSERT INTO webhook_events (delivery_id, event_name, payload_hash, status, received_at) VALUES (?, 'pull_request', 'hash-1', 'queued', '2026-07-06T00:00:00.000Z')",
+      ["same-id"],
+    );
+    const q = createSqliteQueue(driver, async () => undefined);
+
+    await q.binding.send(prWebhook("same-id"), { delaySeconds: 60 });
+    await q.binding.send(prWebhook("same-id"), { delaySeconds: 1 }); // re-coalesces against its own delivery id
+
+    const event = driver.query("SELECT status FROM webhook_events WHERE delivery_id = ?", ["same-id"]).rows[0] as { status: string } | undefined;
+    expect(event?.status).toBe("queued"); // untouched -- there is nothing genuinely superseded here
+  });
+
+  it("tolerates an unparseable existing payload (a corrupted row) without throwing", async () => {
+    const driver = makeDriver();
+    const q = createSqliteQueue(driver, async () => undefined);
+    driver.query(
+      `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key) VALUES (?, 'pending', 0, ?, ?, 5, ?)`,
+      ["not valid json{", Date.now() + 60_000, Date.now(), `github-webhook:pr-refresh:jsonbored/gittensory#1629@${"a".repeat(40)}`],
+    );
+
+    await expect(q.binding.send(prWebhook("pr-2"), { delaySeconds: 1 })).resolves.toBeUndefined();
+
+    const rows = driver.query("SELECT payload FROM _selfhost_jobs ORDER BY id", []).rows as Array<{ payload: string }>;
+    expect(rows).toHaveLength(1); // still coalesced into one row despite the corrupted existing payload
+    expect(JSON.parse(rows[0]!.payload).deliveryId).toBe("pr-2");
+  });
+
   it("lets a pending full RAG index absorb later repo incrementals", async () => {
     const driver = makeDriver();
     const q = createSqliteQueue(driver, async () => undefined);