fix(selfhost): durability — queue leases, atomic migrations, and two repair scans for state a crash used to lose

src/queue/job-dispatch.ts

@@ -51,6 +51,8 @@ import { isDecisionAuditEnabled, runDecisionAuditSample } from "../review/decisi
 import { isRiskControlEnabled, runRiskControlRecalibration } from "../review/risk-control-wire";
 import { runRetentionPrune } from "./retention";
 import { sweepStaleApprovalQueue } from "../services/agent-approval-queue";
+import { reconcileMissingPrOutcomes } from "../review/pr-outcome-reconciler";
+import { sweepStrandedPendingClosures } from "../review/pending-closure-watchdog";
 // The 15 handlers below have no reason to move -- each is only reachable via this dispatcher (or, for
 // mapWithConcurrency, ALSO used by other still-in-processors.ts code), so they stay put and are exported
 // there purely for this one-directional import-back (processors.ts itself never calls processJob).
@@ -273,13 +275,30 @@ export async function processJob(env: Env, message: JobMessage): Promise<void> {
     }
     case "agent-regate-sweep":
       if (!message.repoFullName && message.requestedBy !== "test") {
-        // #9032: piggyback the approval-queue staleness pass on the sweep's own fan-out tick rather than adding
-        // a job type and a cron entry for a bounded DB scan. Best-effort and deliberately BEFORE the fan-out:
-        // a failure here must not cost the tick its re-gate work, which is the sweep's actual job.
+        // Three bounded repair scans ride the sweep's own fan-out tick rather than each earning a job type and
+        // a cron entry. All are best-effort and deliberately BEFORE the fan-out: none may cost the tick its
+        // re-gate work, which is the sweep's actual job.
+        //
+        // #9032 reminds and expires approval-queue rows nobody has acted on.
+        //
+        // #9026 backfills pr_outcome rows lost when a process died between the GitHub mutation and the
+        // bookkeeping write. Those losses skew toward the gate's MISTAKES, so leaving them out biases published
+        // accuracy upward -- this is calibration ground truth, not ordinary telemetry.
+        //
+        // #9031 re-enqueues a pending-closure Pass 2 whose single delayed job was lost, which otherwise strands
+        // the PR permanently: flagged, un-mergeable, un-approvable, and sweep-ineligible.
         const staleness = await sweepStaleApprovalQueue(env).catch(() => null);
         if (staleness && (staleness.reminded > 0 || staleness.expired > 0)) {
           console.log(JSON.stringify({ event: "approval_queue_staleness_swept", ...staleness }));
         }
+        const outcomes = await reconcileMissingPrOutcomes(env).catch(() => null);
+        if (outcomes && outcomes.backfilled > 0) {
+          console.log(JSON.stringify({ event: "pr_outcomes_reconciled", ...outcomes }));
+        }
+        const stranded = await sweepStrandedPendingClosures(env).catch(() => null);
+        if (stranded && stranded.requeued > 0) {
+          console.log(JSON.stringify({ event: "pending_closure_verifications_requeued", ...stranded }));
+        }
         await fanOutAgentRegateSweepJobs(env, message.requestedBy);
         return;
       }

src/queue/processors.ts

@@ -301,6 +301,7 @@ import {
 } from "../services/agent-action-executor";
 import { activeMergeBlockedSha } from "../services/merge-failure";
 import { applyLowConfidenceHoldCap } from "../review/low-confidence-hold-cap";
+import { recordPendingClosureFlag } from "../review/pending-closure-watchdog";
 import { loadIssueQualityReportMap } from "../services/issue-quality";
 import { generateAndSendReviewRecap } from "../services/review-recap";
 import {
@@ -3676,6 +3677,11 @@ async function runAgentMaintenancePlanAndExecute(
         ? env.JOBS.send(verifyJob, { delaySeconds })
         : env.JOBS.send(verifyJob)
     ).catch(() => undefined);
+    // #9031: the durable half of the same intent. The enqueue above is a single queue message, and until now
+    // it was the ONLY thing that could finish this sequence -- lose it and the PR sits flagged forever, since
+    // the flag suppresses merge/approve and #never-endless-reregate makes the PR sweep-ineligible. Recording
+    // the deadline lets sweepStrandedPendingClosures re-enqueue Pass 2 if it never ran.
+    await recordPendingClosureFlag(env, { repoFullName, pullNumber: pr.number, installationId }, Date.now() + delaySeconds * 1000);
   }
 }
 

src/review/pending-closure-watchdog.ts

@@ -0,0 +1,119 @@
+import { countRecentAuditEventsForActorAndTarget, getPullRequest, listAuditEventsByType, recordAuditEvent } from "../db/repositories";
+import { errorMessage } from "../utils/json";
+
+/**
+ * #9031 — rescue a PR stranded mid-way through the flag-then-close double-check.
+ *
+ * Pass 1 applies the pending-closure label and enqueues a single delayed `recapture-preview` job to run Pass 2.
+ * Until then that one queue message was the ONLY thing that could finish the sequence. Its documented backstop —
+ * "the next sweep / CI event" — is largely vacuous: #never-endless-reregate makes an already-regated PR
+ * permanently sweep-ineligible, and the flag itself suppresses merge and approve via `linkedIssueCloseInFlight`.
+ * So if the queue lost the job (a crash mid-flight, #9007), the PR sat flagged with pending-closure plus
+ * changes-requested, going nowhere, waiting on a contributor webhook that may never come — and because the
+ * violation memory is permanent, `clearLinkedIssueFlag` could never fire either.
+ *
+ * The fix the issue asks for is a deadline something re-checks, rather than a sequence that depends on one
+ * message surviving. Pass 1 now records an audit event carrying that deadline, and this watchdog re-enqueues
+ * Pass 2 for any flag past it. `audit_events` is the right store for this: durable, already queried by type,
+ * and repo-agnostic — unlike the label, which is per-repo configurable and would need settings resolution
+ * before a cross-repo scan could even recognize it.
+ *
+ * The re-enqueued job is the same message Pass 1 sends, so a redundant one is harmless: Pass 2 is a re-review
+ * that re-derives everything from live state.
+ */
+
+export const PENDING_CLOSURE_FLAGGED_EVENT = "agent.linked_issue.pending_closure_flagged";
+export const PENDING_CLOSURE_REQUEUED_EVENT = "agent.linked_issue.pending_closure_requeued";
+
+/** Grace beyond the flag's own deadline before the watchdog steps in, so a job merely waiting its turn behind a
+ *  busy queue is never mistaken for a lost one. */
+export const PENDING_CLOSURE_GRACE_MS = 10 * 60 * 1000;
+
+/** How far back to scan. A flag older than this is past rescuing by re-enqueue — the PR has moved on, or a
+ *  human has dealt with it — and the scan stays cheap by not carrying that history forever. */
+export const PENDING_CLOSURE_LOOKBACK_MS = 7 * 24 * 60 * 60 * 1000;
+
+/** Minimum spacing between rescues for the same PR, so a PR that is stuck for a reason the re-enqueue cannot
+ *  fix gets retried periodically instead of on every single sweep tick. */
+export const PENDING_CLOSURE_REQUEUE_INTERVAL_MS = 60 * 60 * 1000;
+
+export type PendingClosureSweepResult = { scanned: number; requeued: number };
+
+/** Pass 1's durable record that a flag was applied and when its verification is due. Best-effort, exactly like
+ *  the enqueue it accompanies — but the two failing together is far less likely than the single message being
+ *  lost, which is the whole point of having both. */
+export async function recordPendingClosureFlag(
+  env: Env,
+  target: { repoFullName: string; pullNumber: number; installationId: number },
+  dueAtMs: number,
+): Promise<void> {
+  await recordAuditEvent(env, {
+    eventType: PENDING_CLOSURE_FLAGGED_EVENT,
+    actor: "loopover",
+    targetKey: `${target.repoFullName}#${target.pullNumber}`,
+    outcome: "queued",
+    detail: "pending-closure flag applied; Pass 2 verification scheduled",
+    metadata: { repoFullName: target.repoFullName, pullNumber: target.pullNumber, installationId: target.installationId, dueAt: new Date(dueAtMs).toISOString() },
+  }).catch(() => undefined);
+}
+
+export async function sweepStrandedPendingClosures(env: Env, nowMs: number = Date.now()): Promise<PendingClosureSweepResult> {
+  let flags: Awaited<ReturnType<typeof listAuditEventsByType>> = [];
+  try {
+    flags = await listAuditEventsByType(env, PENDING_CLOSURE_FLAGGED_EVENT, new Date(nowMs - PENDING_CLOSURE_LOOKBACK_MS).toISOString(), 500);
+  } catch (error) {
+    console.warn(JSON.stringify({ level: "warn", event: "pending_closure_sweep_scan_failed", message: errorMessage(error).slice(0, 160) }));
+    return { scanned: 0, requeued: 0 };
+  }
+
+  let requeued = 0;
+  for (const flag of flags) {
+    const repoFullName = typeof flag.metadata.repoFullName === "string" ? flag.metadata.repoFullName : null;
+    const pullNumber = typeof flag.metadata.pullNumber === "number" ? flag.metadata.pullNumber : null;
+    const installationId = typeof flag.metadata.installationId === "number" ? flag.metadata.installationId : null;
+    if (repoFullName === null || pullNumber === null || installationId === null) continue;
+    const dueAt = typeof flag.metadata.dueAt === "string" ? Date.parse(flag.metadata.dueAt) : NaN;
+    // An unreadable deadline falls back to the event's own timestamp: the flag is still real, and refusing to
+    // rescue it because its metadata is malformed would recreate the exact stranding this exists to end.
+    const deadline = Number.isFinite(dueAt) ? dueAt : Date.parse(flag.createdAt);
+    if (!Number.isFinite(deadline) || nowMs < deadline + PENDING_CLOSURE_GRACE_MS) continue;
+
+    // Only an OPEN PR can still be stranded. A closed/merged one already reached a terminal state, whether by
+    // Pass 2, a maintainer, or the contributor.
+    const pr = await getPullRequest(env, repoFullName, pullNumber).catch(() => null);
+    if (!pr || pr.state !== "open") continue;
+
+    const targetKey = `${repoFullName}#${pullNumber}`;
+    const recent = await countRecentAuditEventsForActorAndTarget(
+      env,
+      "loopover",
+      PENDING_CLOSURE_REQUEUED_EVENT,
+      targetKey,
+      new Date(nowMs - PENDING_CLOSURE_REQUEUE_INTERVAL_MS).toISOString(),
+    ).catch(() => 1);
+    if (recent > 0) continue;
+
+    const verifyJob = {
+      type: "recapture-preview" as const,
+      deliveryId: `linked-issue-verify:${repoFullName}#${pullNumber}`,
+      repoFullName,
+      prNumber: pullNumber,
+      installationId,
+      attempt: 0,
+    };
+    const sent = await env.JOBS.send(verifyJob)
+      .then(() => true)
+      .catch(() => false);
+    if (!sent) continue;
+    requeued += 1;
+    await recordAuditEvent(env, {
+      eventType: PENDING_CLOSURE_REQUEUED_EVENT,
+      actor: "loopover",
+      targetKey,
+      outcome: "queued",
+      detail: "pending-closure verification re-enqueued after its deadline passed with the PR still flagged",
+      metadata: { repoFullName, pullNumber, installationId, deadline: new Date(deadline).toISOString() },
+    }).catch(() => undefined);
+  }
+  return { scanned: flags.length, requeued };
+}

src/review/pr-outcome-reconciler.ts

@@ -0,0 +1,94 @@
+import { recordTerminalActionOutcome } from "./outcomes-wire";
+import { errorMessage } from "../utils/json";
+
+/**
+ * #9026 — backfill `pr_outcome` rows that were never written.
+ *
+ * `pr_outcome` is the realized ground truth the fleet calibration export inner-joins `gate_decision` against, so
+ * a PR missing one does not merely lose a data point: it vanishes from calibration entirely, contributing to
+ * neither numerator nor denominator. It is written two ways, both in-process and both best-effort —
+ * `recordTerminalActionOutcome` immediately after the bot's own merge/close mutation, and `recordPrOutcome` from
+ * the inbound `pull_request.closed` webhook.
+ *
+ * Neither survives a kill at the wrong moment. A process that dies between `mergePullRequest`/`closePullRequest`
+ * and the record call loses the direct write, and on the next pass the PR is already terminal so the planner
+ * plans nothing and the write never fires. The webhook that would have caught it was delivered while the
+ * container was down, and GitHub does not redeliver. Nothing scanned for the gap: the repair sweep only visits
+ * OPEN PRs. #8823 narrowed the window; it did not close it.
+ *
+ * The rows this loses are not a random sample. A superseded close is by definition a wrong close, so the losses
+ * skew toward the gate's mistakes and their absence biases published accuracy UPWARD — which is why this matters
+ * more than ordinary telemetry loss, and why it is worth a reconciler rather than a wider window.
+ *
+ * Scoped to PRs that actually carry a `gate_decision`, because those are exactly the ones calibration reads; a
+ * closed PR the gate never evaluated has no prediction to pair an outcome with. `recordTerminalActionOutcome` is
+ * already idempotent, so a row that raced in between this scan and the write is a no-op.
+ */
+
+/** How far back to look. Long enough to cover a multi-day outage, bounded so the scan stays cheap on every
+ *  run — anything older has already been exported and is beyond repair for the current calibration window. */
+export const PR_OUTCOME_RECONCILE_LOOKBACK_MS = 14 * 24 * 60 * 60 * 1000;
+
+/** Cap per run. The scan is ordered oldest-first so a large backlog drains deterministically across runs
+ *  instead of the same newest slice being retried forever. */
+export const PR_OUTCOME_RECONCILE_LIMIT = 200;
+
+export type PrOutcomeReconcileResult = { scanned: number; backfilled: number };
+
+/**
+ * Find terminal PRs with a gate decision but no recorded outcome, and write the outcome. Returns what it did so
+ * the caller can log it. Best-effort throughout: this is a repair pass, and a repair pass that can itself break
+ * the tick it runs on is worse than the gap it closes.
+ */
+export async function reconcileMissingPrOutcomes(env: Env, nowMs: number = Date.now()): Promise<PrOutcomeReconcileResult> {
+  const since = new Date(nowMs - PR_OUTCOME_RECONCILE_LOOKBACK_MS).toISOString();
+  let rows: Array<{ repoFullName: string; number: number; mergedAt: string | null }> = [];
+  try {
+    const result = await env.DB.prepare(
+      `SELECT pr.repo_full_name AS repoFullName, pr.number AS number, pr.merged_at AS mergedAt
+         FROM pull_requests AS pr
+        WHERE pr.state != 'open'
+          AND pr.updated_at >= ?1
+          AND EXISTS (
+            SELECT 1 FROM review_audit AS gate
+             WHERE gate.target_id = pr.repo_full_name || '#' || pr.number
+               AND gate.event_type = 'gate_decision'
+          )
+          AND NOT EXISTS (
+            SELECT 1 FROM review_audit AS outcome
+             WHERE outcome.target_id = pr.repo_full_name || '#' || pr.number
+               AND outcome.event_type = 'pr_outcome'
+          )
+        ORDER BY pr.updated_at
+        LIMIT ?2`,
+    )
+      .bind(since, PR_OUTCOME_RECONCILE_LIMIT)
+      .all<{ repoFullName: string; number: number; mergedAt: string | null }>();
+    rows = result.results ?? [];
+  } catch (error) {
+    console.warn(JSON.stringify({ level: "warn", event: "pr_outcome_reconcile_scan_failed", message: errorMessage(error).slice(0, 160) }));
+    return { scanned: 0, backfilled: 0 };
+  }
+
+  let backfilled = 0;
+  for (const row of rows) {
+    // `merged_at` is the authoritative signal, not the state string: GitHub reports a merged PR as `closed`
+    // too, and recording a merge as a plain close would invert the very judgment calibration is scoring.
+    const decision = row.mergedAt ? "merged" : "closed";
+    try {
+      await recordTerminalActionOutcome(env, row.repoFullName, row.number, decision);
+      backfilled += 1;
+    } catch (error) {
+      console.warn(
+        JSON.stringify({
+          level: "warn",
+          event: "pr_outcome_reconcile_write_failed",
+          repo: row.repoFullName,
+          pr: row.number,
+          message: errorMessage(error).slice(0, 160),
+        }),
+      );
+    }
+  }
+  return { scanned: rows.length, backfilled };
+}

src/selfhost/backend-contracts.ts

@@ -88,6 +88,13 @@ export interface SelfHostD1Database {
   prepare(query: string): SelfHostD1PreparedStatement;
   batch(statements: SelfHostD1PreparedStatement[]): Promise<Array<{ results: unknown[]; success: true; meta: Record<string, unknown> }>>;
   exec(query: string): Promise<{ count: number; duration: number }>;
+  /** Run a multi-statement script as ONE all-or-nothing transaction (#9027). `exec` cannot give this: on
+   *  Postgres it takes an arbitrary pooled connection per call, so a BEGIN and its COMMIT issued as separate
+   *  `exec`s can land on different connections and mean nothing. Migrations need it because several of them
+   *  carry non-idempotent DML (the table-rebuild `INSERT ... SELECT` pattern, plus UPDATE/DELETE steps) that a
+   *  crash mid-file would otherwise re-run in full on the next boot — double-inserting, or failing on a PK
+   *  conflict that is not "already exists" and so bricking boot outright. */
+  execTransaction(query: string): Promise<void>;
   /** @deprecated present only because both self-host adapters still implement it for D1 surface completeness;
    *  real D1 no longer uses it either (see d1-adapter.ts / pg-adapter.ts). */
   dump(): Promise<ArrayBuffer>;

src/selfhost/d1-adapter.ts

@@ -92,6 +92,24 @@ export function createD1Adapter(driver: SqliteDriver): D1Database {
       driver.exec(sql); // runs one or more statements (used for migrations)
       return { count: (sql.match(/;/g) ?? []).length || 1, duration: 0 };
     },
+    async execTransaction(sql: string) {
+      // #9027: SQLite is a single connection here, so BEGIN/COMMIT issued as separate exec calls are already
+      // the same session — unlike Postgres, where that is exactly the trap. Still explicit rather than relying
+      // on the driver: node:sqlite's exec() runs statements sequentially with no implicit transaction, so a
+      // crash partway through a migration file would leave it half-applied without this.
+      driver.exec("BEGIN");
+      try {
+        driver.exec(sql);
+        driver.exec("COMMIT");
+      } catch (error) {
+        try {
+          driver.exec("ROLLBACK");
+        } catch {
+          /* ignore -- the original error is the one worth reporting */
+        }
+        throw error;
+      }
+    },
     async dump() {
       return new ArrayBuffer(0); // unused by loopover; present for D1 surface completeness
     },