diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts
index 37a3b6eb2..f3558f525 100644
--- a/src/queue/job-dispatch.ts
+++ b/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;
       }
diff --git a/src/queue/processors.ts b/src/queue/processors.ts
index 3254d6bf6..b0742bafd 100644
--- a/src/queue/processors.ts
+++ b/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);
   }
 }
 
diff --git a/src/review/pending-closure-watchdog.ts b/src/review/pending-closure-watchdog.ts
new file mode 100644
index 000000000..eeeef4d17
--- /dev/null
+++ b/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 };
+}
diff --git a/src/review/pr-outcome-reconciler.ts b/src/review/pr-outcome-reconciler.ts
new file mode 100644
index 000000000..cd0d50e7e
--- /dev/null
+++ b/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 };
+}
diff --git a/src/selfhost/backend-contracts.ts b/src/selfhost/backend-contracts.ts
index 3d65c6dd5..a108b6d1c 100644
--- a/src/selfhost/backend-contracts.ts
+++ b/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>;
diff --git a/src/selfhost/d1-adapter.ts b/src/selfhost/d1-adapter.ts
index 394bac1af..510f0bce4 100644
--- a/src/selfhost/d1-adapter.ts
+++ b/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
     },
diff --git a/src/selfhost/migrate.ts b/src/selfhost/migrate.ts
index 7efdb1f7a..cc5ccc969 100644
--- a/src/selfhost/migrate.ts
+++ b/src/selfhost/migrate.ts
@@ -73,16 +73,70 @@ function splitSqlStatements(sql: string): string[] {
   return statements;
 }
 
+/** SQL-literal escape for the migration name written into the ledger inside the transactional script. The
+ *  names are our own repo filenames (`NNNN_*.sql`), not user input, but the doubling keeps the script correct
+ *  for any name rather than relying on that. */
+function sqlQuote(value: string): string {
+  return `'${value.replace(/'/g, "''")}'`;
+}
+
+/**
+ * Apply pending self-host migrations.
+ *
+ * #9027 — each file now runs as ONE transaction, with its ledger row committed inside that same transaction, so
+ * a file is all-or-nothing. Before this, statements ran individually and the file was recorded only after the
+ * last one succeeded, so a crash mid-file re-ran the WHOLE file on the next boot. That is fine for pure DDL and
+ * actively harmful for the several migrations carrying real DML — the table-rebuild `INSERT ... SELECT`
+ * pattern, plus UPDATE/DELETE steps — where a re-run either double-inserts or raises a PK conflict which, not
+ * matching the tolerated "already exists" shape, throws and bricks boot. With deploys currently ending in
+ * SIGKILL under load (#9007), a mid-migration kill is not hypothetical.
+ *
+ * The pre-existing drift tolerance is preserved, not replaced. A database that already drifted — a column a
+ * migration adds is somehow present — must still heal itself rather than fail to boot, and that self-healing
+ * only works per-statement. So a file that fails atomically is retried through the original tolerant path.
+ * The two orders matter: atomic first means a genuine mid-file crash is the case that gets protected, and the
+ * tolerant fallback only runs for a database that was already inconsistent before this boot began.
+ */
 export async function runSelfHostMigrations(db: D1Database, dir: string): Promise<number> {
   await db.exec("CREATE TABLE IF NOT EXISTS _selfhost_migrations (name TEXT PRIMARY KEY, applied_at TEXT NOT NULL)");
   const existing = await db.prepare("SELECT name FROM _selfhost_migrations").all<{ name: string }>();
   const applied = new Set(existing.results.map((r) => r.name));
   const files = readdirSync(dir).filter((f) => f.endsWith(".sql")).sort();
+  // Real Cloudflare D1 has no transactional multi-statement surface; only the two self-host adapters implement
+  // this. Absent it, behavior is exactly the pre-#9027 path.
+  const execTransaction = (db as unknown as { execTransaction?: (sql: string) => Promise<void> }).execTransaction;
   let count = 0;
   for (const file of files) {
     if (applied.has(file)) continue;
     const sql = readFileSync(join(dir, file), "utf8");
-    for (const statement of splitSqlStatements(sql)) {
+    const statements = splitSqlStatements(sql);
+    const ledger = `INSERT INTO _selfhost_migrations (name, applied_at) VALUES (${sqlQuote(file)}, ${sqlQuote(new Date().toISOString())});`;
+    if (execTransaction) {
+      try {
+        // splitSqlStatements keeps each statement's own trailing `;` but returns a final unterminated tail
+        // verbatim, so re-terminate before concatenating — otherwise the ledger INSERT fuses onto the last
+        // statement and the whole file fails to parse.
+        const script = statements.map((statement) => (statement.endsWith(";") ? statement : `${statement};`)).join("\n");
+        await execTransaction.call(db, `${script}\n${ledger}`);
+        count += 1;
+        continue;
+      } catch (error) {
+        // ONLY drift falls through. Any other failure rethrows here, with the transaction already rolled back --
+        // so boot dies loudly against a database in exactly the state it started in. Retrying a genuinely broken
+        // file through the per-statement path would re-apply its valid statements after the rollback and leave
+        // precisely the partial state this fix exists to prevent.
+        if (!/duplicate column|already exists/i.test(errorMessage(error))) throw error;
+        console.warn(
+          JSON.stringify({
+            level: "warn",
+            event: "selfhost_migration_transaction_failed",
+            file,
+            message: errorMessage(error),
+          }),
+        );
+      }
+    }
+    for (const statement of statements) {
       try {
         await db.exec(statement);
       } catch (error) {
diff --git a/src/selfhost/pg-adapter.ts b/src/selfhost/pg-adapter.ts
index 1d666fa48..565145394 100644
--- a/src/selfhost/pg-adapter.ts
+++ b/src/selfhost/pg-adapter.ts
@@ -90,6 +90,28 @@ export function createPgAdapter(pool: Pool): D1Database {
       await pool.query(translateDdl(sql));
       return { count: (sql.match(/;/g) ?? []).length || 1, duration: 0 };
     },
+    async execTransaction(sql: string) {
+      // #9027: a dedicated client, so BEGIN/COMMIT/ROLLBACK are guaranteed to be the SAME session — the thing
+      // `exec` above cannot promise, since each of its calls takes whatever connection the pool hands out.
+      // Postgres DDL is transactional, so a migration file wrapped this way is genuinely all-or-nothing.
+      // Mirrors batch()'s transaction handling above, including releasing the client on every path so a failed
+      // transaction can never be returned to the pool still open.
+      const client = await pool.connect();
+      try {
+        await client.query("BEGIN");
+        await client.query(translateDdl(sql));
+        await client.query("COMMIT");
+      } catch (error) {
+        try {
+          await client.query("ROLLBACK");
+        } catch {
+          /* ignore -- the original error is the one worth reporting */
+        }
+        throw error;
+      } finally {
+        client.release();
+      }
+    },
     async dump() {
       return new ArrayBuffer(0); // unused; present for D1 surface completeness
     },
diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts
index 7ee5bdd6a..1f460cd8c 100644
--- a/src/selfhost/pg-queue.ts
+++ b/src/selfhost/pg-queue.ts
@@ -565,24 +565,65 @@ export function createPgQueue(
     return result.rowCount ?? 0;
   }
 
+  /**
+   * Boot recovery: re-pend jobs a previous process left mid-flight.
+   *
+   * #9023 — this used to re-pend EVERY row in `processing`, unconditionally. On Postgres the queue is
+   * explicitly multi-instance (that is what the FOR UPDATE SKIP LOCKED claim exists for), so "every processing
+   * row" is not "my crashed predecessor's work" — during any overlapped or rolling deploy it also includes jobs
+   * a SIBLING instance is running RIGHT NOW. Both processes then ran the same PR pass concurrently, producing
+   * duplicate gate check-runs and verdict thrash.
+   *
+   * The lease cutoff is the fix: only a row whose lease has actually expired can belong to a dead process, and
+   * that is exactly the condition reclaimExpiredProcessingJobs already encodes, so boot and steady-state now
+   * agree on what "abandoned" means instead of using two different definitions. A live sibling heartbeats its
+   * lease (see heartbeatProcessingLease), so its rows are never eligible however long the job legitimately runs.
+   */
   async function recoverProcessingJobs(): Promise<number> {
+    // A disabled timeout means leases never expire, so nothing can be proven abandoned -- recovering anything
+    // would be a guess, and the guess is the bug. The runtime reaper is disabled in that configuration too.
+    if (processingTimeoutMs <= 0) return 0;
+    const now = Date.now();
     const res = await pool.query(
-      `SELECT id, payload, job_key FROM ${TABLE} WHERE status='processing'`,
+      `SELECT id, payload, job_key FROM ${TABLE} WHERE status='processing' AND run_after<=$1`,
+      [now - processingTimeoutMs],
     );
     let changed = 0;
-    const now = Date.now();
     const maxJitter = queueRecoveryJitterMs();
     for (const row of res.rows as Array<{ id: string; payload: string; job_key?: string | null }>) {
       const runAfter = now + deterministicJitterMs(`${row.job_key ?? ""}:${row.id}:${row.payload}`, maxJitter);
-      await pool.query(`UPDATE ${TABLE} SET status='pending', run_after=$1 WHERE id=$2`, [
-        runAfter,
-        row.id,
-      ]);
-      changed += 1;
+      // AND status='processing' makes this a compare-and-swap: between the SELECT above and this UPDATE a
+      // sibling may have finished the job (row deleted) or reclaimed it itself. Without the guard this could
+      // resurrect a completed job as pending. Mirrors reclaimExpiredProcessingJobs' own guard.
+      const update = await pool.query(
+        `UPDATE ${TABLE} SET status='pending', run_after=$1 WHERE id=$2 AND status='processing'`,
+        [runAfter, row.id],
+      );
+      changed += update.rowCount ?? 0;
     }
     return changed;
   }
 
+  /**
+   * Refresh the lease on a job this process is actively running (#9023).
+   *
+   * `run_after` doubles as the lease timestamp: the claim sets it to now, and a row is reclaimable once it is
+   * older than `processingTimeoutMs`. Without a heartbeat that made the timeout a hard ceiling on job DURATION,
+   * not on liveness — a genuinely long pass (a large PR, a slow AI review, a rate-limited GitHub window) got
+   * reclaimed and double-run by a sibling purely for taking too long. The `activeJobIds` check that was meant to
+   * prevent this only covers THIS process's in-memory set, which is precisely the wrong scope for a queue whose
+   * whole point is that several processes share it.
+   *
+   * Best-effort by design: a failed heartbeat just means the lease ages normally, which is the pre-#9023
+   * behavior. Losing the row (deleted on completion, or dead-lettered) makes this a harmless no-op via the
+   * status guard.
+   */
+  async function heartbeatProcessingLease(id: string): Promise<void> {
+    await pool
+      .query(`UPDATE ${TABLE} SET run_after=$1 WHERE id=$2 AND status='processing'`, [Date.now(), id])
+      .catch(() => undefined);
+  }
+
   // Dead-letter auto-retry (#audit-rate-headroom): a job dies once `attempts >= maxRetries` (see the
   // max-retries branch in processOne below). Reviving it here only clears `status`/`run_after`/`last_error`
   // -- `attempts` is left untouched, so it already satisfies `attempts >= maxRetries` and will die again
@@ -1114,6 +1155,13 @@ export function createPgQueue(
     if (!job) return false;
     activeJobIds.add(job.id);
     const claimedAt = Date.now();
+    // #9023: hold the lease open for as long as this process is genuinely working on the job. Without this the
+    // lease timeout is a ceiling on job DURATION rather than on liveness, so a legitimately slow pass gets
+    // reclaimed and double-run by a sibling. Refreshed at a third of the timeout so two refreshes can be lost
+    // (a blip, a paused event loop) before the lease is actually at risk.
+    const heartbeat = setInterval(() => void heartbeatProcessingLease(job.id), Math.max(1000, Math.floor(processingTimeoutMs / 3)));
+    // Never let the timer hold the process open: shutdown must not wait on a heartbeat tick.
+    heartbeat.unref?.();
     try {
       let message: JobMessage;
       try {
@@ -1468,6 +1516,7 @@ export function createPgQueue(
       }
       return true;
     } finally {
+      clearInterval(heartbeat);
       activeJobIds.delete(job.id);
       if (job.backgroundSlotReserved)
         activeBackground = Math.max(0, activeBackground - 1);
@@ -1630,17 +1679,39 @@ export function createPgQueue(
     const now = Date.now();
     const cutoff = now - processingTimeoutMs;
     const res = await pool.query(
-      `SELECT id, payload, job_key FROM ${TABLE} WHERE status='processing' AND run_after<=$1`,
+      `SELECT id, payload, job_key, attempts FROM ${TABLE} WHERE status='processing' AND run_after<=$1`,
       [cutoff],
     );
     let changed = 0;
     const maxJitter = queueRecoveryJitterMs();
-    for (const row of res.rows as Array<{ id: string; payload: string; job_key?: string | null }>) {
+    for (const row of res.rows as Array<{ id: string; payload: string; job_key?: string | null; attempts: number | string }>) {
       if (activeJobIds.has(row.id)) continue;
+      // #9023: a reclaim now COSTS an attempt. It never did, so a job that reliably wedges past its lease --
+      // an infinite loop, a hang against an unresponsive dependency -- was requeued forever: reclaimed, re-run,
+      // wedged, reclaimed, with `attempts` frozen at whatever the last real failure left it. It could never
+      // reach the dead-letter threshold that exists precisely to stop a poison job from burning the queue.
+      // Charging the attempt makes a chronic wedger die on the same budget as a chronic thrower.
+      const attempts = Number(row.attempts) + 1;
+      if (attempts >= maxRetries) {
+        const update = await pool.query(
+          `UPDATE ${TABLE} SET status='dead', attempts=$1, last_error=$2, dead_at=$3 WHERE id=$4 AND status='processing'`,
+          [attempts, "processing lease expired repeatedly; dead-lettered", Date.now(), row.id],
+        );
+        if ((update.rowCount ?? 0) > 0) {
+          await recordQueueMetric("loopover_jobs_dead_total");
+          capturePostHogError(new Error("self-host queue job wedged past its lease repeatedly"), {
+            kind: "job_dead",
+            reason: "processing_lease_exhausted",
+            jobId: row.id,
+            attempts,
+          }, "processing_lease_exhausted");
+        }
+        continue;
+      }
       const runAfter = now + deterministicJitterMs(`${row.job_key ?? ""}:${row.id}:${row.payload}`, maxJitter);
       const update = await pool.query(
-        `UPDATE ${TABLE} SET status='pending', run_after=$1, last_error=COALESCE(last_error, $2) WHERE id=$3 AND status='processing'`,
-        [runAfter, "processing lease expired; requeued", row.id],
+        `UPDATE ${TABLE} SET status='pending', run_after=$1, attempts=$2, last_error=COALESCE(last_error, $3) WHERE id=$4 AND status='processing'`,
+        [runAfter, attempts, "processing lease expired; requeued", row.id],
       );
       changed += update.rowCount ?? 0;
     }
diff --git a/test/unit/docs-beta-onboarding.test.ts b/test/unit/docs-beta-onboarding.test.ts
index 28c6fe87a..ff41cb307 100644
--- a/test/unit/docs-beta-onboarding.test.ts
+++ b/test/unit/docs-beta-onboarding.test.ts
@@ -51,7 +51,11 @@ describe("docs beta onboarding page", () => {
     expect(source).toMatch(/official Gittensor product surface/i);
     expect(source).toMatch(/official Gittensor frontend/i);
     expect(source).toMatch(/independent of/i);
-    expect(source).toMatch(/base-agent/i);
+    // Deliberately NOT asserting "base-agent" any more. #9117 repositioned the product away from "a
+    // deterministic base-agent for the Gittensor ecosystem" toward an agent stack for both sides of the pull
+    // request on ANY GitHub repo, which is a narrowing this test had no business vetoing -- it exists to stop
+    // LoopOver claiming to BE the official Gittensor frontend, and the four assertions here cover that on
+    // their own. Pinning the old wording only made the guard fire on an intended product change.
     expect(source).not.toMatch(/the official Gittensor frontend/i);
   });
 });
diff --git a/test/unit/job-dispatch.test.ts b/test/unit/job-dispatch.test.ts
index 125ae0e9c..2c786a7ca 100644
--- a/test/unit/job-dispatch.test.ts
+++ b/test/unit/job-dispatch.test.ts
@@ -144,3 +144,46 @@ describe("agent-regate-sweep also sweeps the stale approval queue (#9032)", () =
     await expect(processJob(env, { type: "agent-regate-sweep", requestedBy: "schedule" })).resolves.toBeUndefined();
   });
 });
+
+// #9026 / #9031: two bounded repair scans ride the sweep's own fan-out tick rather than each earning a job type
+// and a cron entry. Both must be best-effort — neither may cost the tick its re-gate work, which is the sweep's
+// actual job — and both must be quiet when there is nothing to repair.
+describe("agent-regate-sweep runs the durability repair scans (#9026, #9031)", () => {
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  async function sweepWith(stubs: { outcomes?: unknown; stranded?: unknown; outcomesThrows?: boolean; strandedThrows?: boolean }): Promise<Record<string, unknown>[]> {
+    const logs: string[] = [];
+    vi.spyOn(console, "log").mockImplementation((...args: unknown[]) => void logs.push(String(args[0])));
+    const reconciler = await import("../../src/review/pr-outcome-reconciler");
+    const watchdog = await import("../../src/review/pending-closure-watchdog");
+    const outcomeSpy = vi.spyOn(reconciler, "reconcileMissingPrOutcomes");
+    const strandedSpy = vi.spyOn(watchdog, "sweepStrandedPendingClosures");
+    if (stubs.outcomesThrows) outcomeSpy.mockRejectedValue(new Error("db down"));
+    else outcomeSpy.mockResolvedValue((stubs.outcomes ?? { scanned: 0, backfilled: 0 }) as never);
+    if (stubs.strandedThrows) strandedSpy.mockRejectedValue(new Error("db down"));
+    else strandedSpy.mockResolvedValue((stubs.stranded ?? { scanned: 0, requeued: 0 }) as never);
+
+    await processJob(createTestEnv(), { type: "agent-regate-sweep", requestedBy: "schedule" });
+    return logs.map((line) => JSON.parse(line) as Record<string, unknown>);
+  }
+
+  it("reports what each scan repaired", async () => {
+    const logs = await sweepWith({ outcomes: { scanned: 5, backfilled: 3 }, stranded: { scanned: 2, requeued: 1 } });
+    expect(logs.find((log) => log.event === "pr_outcomes_reconciled")).toMatchObject({ scanned: 5, backfilled: 3 });
+    expect(logs.find((log) => log.event === "pending_closure_verifications_requeued")).toMatchObject({ scanned: 2, requeued: 1 });
+  });
+
+  it("stays quiet when a scan looked but repaired nothing", async () => {
+    const logs = await sweepWith({ outcomes: { scanned: 9, backfilled: 0 }, stranded: { scanned: 4, requeued: 0 } });
+    expect(logs.some((log) => log.event === "pr_outcomes_reconciled")).toBe(false);
+    expect(logs.some((log) => log.event === "pending_closure_verifications_requeued")).toBe(false);
+  });
+
+  it("completes the tick even when both scans throw", async () => {
+    const logs = await sweepWith({ outcomesThrows: true, strandedThrows: true });
+    expect(logs.some((log) => log.event === "pr_outcomes_reconciled")).toBe(false);
+    expect(logs.some((log) => log.event === "pending_closure_verifications_requeued")).toBe(false);
+  });
+});
diff --git a/test/unit/pending-closure-watchdog.test.ts b/test/unit/pending-closure-watchdog.test.ts
new file mode 100644
index 000000000..a9303624f
--- /dev/null
+++ b/test/unit/pending-closure-watchdog.test.ts
@@ -0,0 +1,144 @@
+import { describe, expect, it, vi } from "vitest";
+import { upsertInstallation, upsertPullRequestFromGitHub, upsertRepositoryFromGitHub } from "../../src/db/repositories";
+import {
+  PENDING_CLOSURE_GRACE_MS,
+  PENDING_CLOSURE_LOOKBACK_MS,
+  PENDING_CLOSURE_REQUEUE_INTERVAL_MS,
+  recordPendingClosureFlag,
+  sweepStrandedPendingClosures,
+} from "../../src/review/pending-closure-watchdog";
+import { createTestEnv } from "../helpers/d1";
+
+function envWithQueue(): { env: Env; sent: unknown[] } {
+  const sent: unknown[] = [];
+  const env = createTestEnv();
+  (env as unknown as { JOBS: { send: (msg: unknown) => Promise<void> } }).JOBS = {
+    send: async (msg: unknown) => void sent.push(msg),
+  };
+  return { env, sent };
+}
+
+async function seedPr(env: Env, number: number, state: string): Promise<void> {
+  await upsertInstallation(env, {
+    installation: { id: 88, account: { login: "alice", id: 88, type: "User" }, repository_selection: "selected", permissions: { metadata: "read" }, events: ["pull_request"] },
+  });
+  await upsertRepositoryFromGitHub(env, { name: "repo", full_name: "alice/repo", private: false, owner: { login: "alice" } }, 88);
+  await upsertPullRequestFromGitHub(env, "alice/repo", { number, title: "PR", state, user: { login: "bob" }, head: { sha: "s1" }, labels: [], body: "b" });
+}
+
+// #9031: Pass 1 applies the pending-closure label and enqueues ONE delayed job to run Pass 2. That message was
+// the only thing that could finish the sequence — and the documented "next sweep / CI event" backstop is
+// vacuous, because #never-endless-reregate makes an already-regated PR permanently sweep-ineligible while the
+// flag itself suppresses merge and approve. Lose the message and the PR is stranded for good.
+describe("sweepStrandedPendingClosures (#9031)", () => {
+  const PAST_DUE = (now: number) => now - PENDING_CLOSURE_GRACE_MS - 60_000;
+
+  it("re-enqueues Pass 2 for a flag whose deadline passed with the PR still open", async () => {
+    const { env, sent } = envWithQueue();
+    await seedPr(env, 11, "open");
+    const now = Date.now();
+    await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 11, installationId: 88 }, PAST_DUE(now));
+
+    expect(await sweepStrandedPendingClosures(env, now)).toEqual({ scanned: 1, requeued: 1 });
+    expect(sent).toEqual([
+      { type: "recapture-preview", deliveryId: "linked-issue-verify:alice/repo#11", repoFullName: "alice/repo", prNumber: 11, installationId: 88, attempt: 0 },
+    ]);
+  });
+
+  it("waits out the grace period, so a job merely queued behind a busy backlog is not mistaken for a lost one", async () => {
+    const { env, sent } = envWithQueue();
+    await seedPr(env, 12, "open");
+    const now = Date.now();
+    await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 12, installationId: 88 }, now - 1000);
+
+    expect(await sweepStrandedPendingClosures(env, now)).toEqual({ scanned: 1, requeued: 0 });
+    expect(sent).toEqual([]);
+  });
+
+  it("leaves a PR that already reached a terminal state alone", async () => {
+    const { env, sent } = envWithQueue();
+    await seedPr(env, 13, "closed");
+    const now = Date.now();
+    await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 13, installationId: 88 }, PAST_DUE(now));
+
+    expect(await sweepStrandedPendingClosures(env, now)).toEqual({ scanned: 1, requeued: 0 });
+    expect(sent).toEqual([]);
+  });
+
+  it("does not rescue the same PR on every tick — a PR stuck for another reason is retried periodically", async () => {
+    const { env, sent } = envWithQueue();
+    await seedPr(env, 14, "open");
+    const now = Date.now();
+    await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 14, installationId: 88 }, PAST_DUE(now));
+
+    expect((await sweepStrandedPendingClosures(env, now)).requeued).toBe(1);
+    expect((await sweepStrandedPendingClosures(env, now + 60_000)).requeued).toBe(0);
+    // Past the interval it tries again, rather than giving up on a PR that is genuinely still stranded.
+    expect((await sweepStrandedPendingClosures(env, now + PENDING_CLOSURE_REQUEUE_INTERVAL_MS + 1000)).requeued).toBe(1);
+    expect(sent).toHaveLength(2);
+  });
+
+  it("ignores flags older than the lookback window", async () => {
+    const { env } = envWithQueue();
+    await seedPr(env, 15, "open");
+    const now = Date.now();
+    await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 15, installationId: 88 }, PAST_DUE(now));
+
+    expect(await sweepStrandedPendingClosures(env, now + PENDING_CLOSURE_LOOKBACK_MS + 60_000)).toEqual({ scanned: 0, requeued: 0 });
+  });
+
+  it("falls back to the flag's own timestamp when the recorded deadline is unreadable", async () => {
+    const { env, sent } = envWithQueue();
+    await seedPr(env, 16, "open");
+    await env.DB.prepare("INSERT INTO audit_events (id, event_type, actor, target_key, outcome, detail, metadata_json, created_at) VALUES (?,?,?,?,?,?,?,?)")
+      .bind(
+        crypto.randomUUID(),
+        "agent.linked_issue.pending_closure_flagged",
+        "loopover",
+        "alice/repo#16",
+        "queued",
+        "flagged",
+        JSON.stringify({ repoFullName: "alice/repo", pullNumber: 16, installationId: 88, dueAt: "not-a-date" }),
+        new Date(Date.now() - 3 * PENDING_CLOSURE_GRACE_MS).toISOString(),
+      )
+      .run();
+
+    // Refusing to rescue a real flag because its metadata is malformed would recreate the exact stranding
+    // this watchdog exists to end.
+    expect((await sweepStrandedPendingClosures(env)).requeued).toBe(1);
+    expect(sent).toHaveLength(1);
+  });
+
+  it("skips a flag whose metadata cannot identify a PR at all", async () => {
+    const { env, sent } = envWithQueue();
+    await env.DB.prepare("INSERT INTO audit_events (id, event_type, actor, target_key, outcome, detail, metadata_json, created_at) VALUES (?,?,?,?,?,?,?,?)")
+      .bind(crypto.randomUUID(), "agent.linked_issue.pending_closure_flagged", "loopover", "alice/repo#17", "queued", "flagged", JSON.stringify({ repoFullName: "alice/repo" }), new Date().toISOString())
+      .run();
+
+    expect(await sweepStrandedPendingClosures(env)).toEqual({ scanned: 1, requeued: 0 });
+    expect(sent).toEqual([]);
+  });
+
+  it("does not count a rescue whose enqueue failed", async () => {
+    const { env } = envWithQueue();
+    await seedPr(env, 18, "open");
+    const now = Date.now();
+    await recordPendingClosureFlag(env, { repoFullName: "alice/repo", pullNumber: 18, installationId: 88 }, PAST_DUE(now));
+    (env as unknown as { JOBS: { send: () => Promise<void> } }).JOBS = { send: async () => Promise.reject(new Error("queue down")) };
+
+    expect(await sweepStrandedPendingClosures(env, now)).toEqual({ scanned: 1, requeued: 0 });
+    // And nothing was recorded, so the next tick retries rather than believing it already rescued this PR.
+    expect((await sweepStrandedPendingClosures(env, now + 1000)).scanned).toBe(1);
+  });
+
+  it("returns an empty result instead of throwing when the scan fails", async () => {
+    const { env } = envWithQueue();
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    const repositories = await import("../../src/db/repositories");
+    vi.spyOn(repositories, "listAuditEventsByType").mockRejectedValue(new Error("db unavailable"));
+
+    expect(await sweepStrandedPendingClosures(env)).toEqual({ scanned: 0, requeued: 0 });
+    expect(warn.mock.calls.some(([line]) => String(line).includes("pending_closure_sweep_scan_failed"))).toBe(true);
+    vi.restoreAllMocks();
+  });
+});
diff --git a/test/unit/pr-outcome-reconciler.test.ts b/test/unit/pr-outcome-reconciler.test.ts
new file mode 100644
index 000000000..8155c3550
--- /dev/null
+++ b/test/unit/pr-outcome-reconciler.test.ts
@@ -0,0 +1,137 @@
+import { describe, expect, it, vi } from "vitest";
+import { upsertInstallation, upsertPullRequestFromGitHub, upsertRepositoryFromGitHub } from "../../src/db/repositories";
+import { PR_OUTCOME_RECONCILE_LIMIT, PR_OUTCOME_RECONCILE_LOOKBACK_MS, reconcileMissingPrOutcomes } from "../../src/review/pr-outcome-reconciler";
+import { createTestEnv } from "../helpers/d1";
+
+async function seedRepo(env: Env): Promise<void> {
+  await upsertInstallation(env, {
+    installation: { id: 55, account: { login: "alice", id: 55, type: "User" }, repository_selection: "selected", permissions: { metadata: "read" }, events: ["pull_request"] },
+  });
+  await upsertRepositoryFromGitHub(env, { name: "repo", full_name: "alice/repo", private: false, owner: { login: "alice" } }, 55);
+}
+
+async function seedPr(env: Env, number: number, opts: { state: string; mergedAt?: string | null }): Promise<void> {
+  await upsertPullRequestFromGitHub(env, "alice/repo", {
+    number,
+    title: `PR ${number}`,
+    state: opts.state,
+    user: { login: "bob" },
+    head: { sha: `sha${number}` },
+    labels: [],
+    body: "b",
+    ...(opts.mergedAt ? { merged_at: opts.mergedAt } : {}),
+  });
+}
+
+async function addReviewAudit(env: Env, number: number, eventType: string): Promise<void> {
+  await env.DB.prepare("INSERT INTO review_audit (id, project, target_id, event_type, decision, created_at) VALUES (?, ?, ?, ?, ?, ?)")
+    .bind(crypto.randomUUID(), "alice/repo", `alice/repo#${number}`, eventType, "closed", new Date().toISOString())
+    .run();
+}
+
+async function outcomeRows(env: Env, number: number): Promise<number> {
+  const row = await env.DB.prepare("SELECT COUNT(*) AS c FROM review_audit WHERE target_id = ? AND event_type = 'pr_outcome'")
+    .bind(`alice/repo#${number}`)
+    .first<{ c: number }>();
+  return Number(row?.c ?? 0);
+}
+
+// #9026: pr_outcome is the realized ground truth the fleet calibration export INNER JOINs gate_decision
+// against, so a PR missing one vanishes from calibration entirely — neither numerator nor denominator. Both
+// writers are in-process and best-effort, and nothing scanned for the gap: the repair sweep only visits OPEN
+// PRs. The losses skew toward the gate's mistakes (a superseded close is by definition a wrong close), so their
+// absence biased published accuracy UPWARD.
+describe("reconcileMissingPrOutcomes (#9026)", () => {
+  it("backfills a closed PR that has a gate decision but no outcome", async () => {
+    const env = createTestEnv();
+    await seedRepo(env);
+    await seedPr(env, 1, { state: "closed" });
+    await addReviewAudit(env, 1, "gate_decision");
+
+    expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 1, backfilled: 1 });
+    expect(await outcomeRows(env, 1)).toBe(1);
+  });
+
+  it("records a merged PR as merged, not closed — GitHub reports both as state=closed", async () => {
+    const env = createTestEnv();
+    await seedRepo(env);
+    await seedPr(env, 2, { state: "closed", mergedAt: new Date().toISOString() });
+    await addReviewAudit(env, 2, "gate_decision");
+
+    await reconcileMissingPrOutcomes(env);
+    const row = await env.DB.prepare("SELECT decision FROM review_audit WHERE target_id = ? AND event_type = 'pr_outcome'")
+      .bind("alice/repo#2")
+      .first<{ decision: string }>();
+    // Recording a merge as a plain close would invert the very judgment calibration is scoring.
+    expect(row?.decision).toBe("merged");
+  });
+
+  it("leaves open PRs alone — they have no realized outcome yet", async () => {
+    const env = createTestEnv();
+    await seedRepo(env);
+    await seedPr(env, 3, { state: "open" });
+    await addReviewAudit(env, 3, "gate_decision");
+
+    expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 0, backfilled: 0 });
+  });
+
+  it("ignores a closed PR the gate never evaluated — no prediction means nothing to pair an outcome with", async () => {
+    const env = createTestEnv();
+    await seedRepo(env);
+    await seedPr(env, 4, { state: "closed" });
+
+    expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 0, backfilled: 0 });
+    expect(await outcomeRows(env, 4)).toBe(0);
+  });
+
+  it("skips a PR whose outcome was already recorded, and stays idempotent across runs", async () => {
+    const env = createTestEnv();
+    await seedRepo(env);
+    await seedPr(env, 5, { state: "closed" });
+    await addReviewAudit(env, 5, "gate_decision");
+
+    await reconcileMissingPrOutcomes(env);
+    expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 0, backfilled: 0 });
+    expect(await outcomeRows(env, 5)).toBe(1);
+  });
+
+  it("ignores anything older than the lookback window", async () => {
+    const env = createTestEnv();
+    await seedRepo(env);
+    await seedPr(env, 6, { state: "closed" });
+    await addReviewAudit(env, 6, "gate_decision");
+
+    expect(await reconcileMissingPrOutcomes(env, Date.now() + PR_OUTCOME_RECONCILE_LOOKBACK_MS + 60_000)).toEqual({ scanned: 0, backfilled: 0 });
+  });
+
+  it("returns an empty result instead of throwing when the scan fails — a repair pass must not break its tick", async () => {
+    const env = createTestEnv();
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    vi.spyOn(env.DB, "prepare").mockImplementation(() => {
+      throw new Error("db unavailable");
+    });
+
+    expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 0, backfilled: 0 });
+    expect(warn.mock.calls.some(([line]) => String(line).includes("pr_outcome_reconcile_scan_failed"))).toBe(true);
+    vi.restoreAllMocks();
+  });
+
+  it("counts a scanned row it could not write, rather than reporting it as backfilled", async () => {
+    const env = createTestEnv();
+    await seedRepo(env);
+    await seedPr(env, 7, { state: "closed" });
+    await addReviewAudit(env, 7, "gate_decision");
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    const outcomes = await import("../../src/review/outcomes-wire");
+    vi.spyOn(outcomes, "recordTerminalActionOutcome").mockRejectedValue(new Error("write failed"));
+
+    expect(await reconcileMissingPrOutcomes(env)).toEqual({ scanned: 1, backfilled: 0 });
+    expect(warn.mock.calls.some(([line]) => String(line).includes("pr_outcome_reconcile_write_failed"))).toBe(true);
+    vi.restoreAllMocks();
+  });
+
+  it("bounds each run so a large backlog drains across runs instead of blocking one", () => {
+    expect(PR_OUTCOME_RECONCILE_LIMIT).toBeGreaterThan(0);
+    expect(PR_OUTCOME_RECONCILE_LIMIT).toBeLessThanOrEqual(1000);
+  });
+});
diff --git a/test/unit/selfhost-migration-atomicity.test.ts b/test/unit/selfhost-migration-atomicity.test.ts
new file mode 100644
index 000000000..022b7d35b
--- /dev/null
+++ b/test/unit/selfhost-migration-atomicity.test.ts
@@ -0,0 +1,106 @@
+import { mkdtempSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { DatabaseSync } from "node:sqlite";
+import { describe, expect, it, vi } from "vitest";
+import { createD1Adapter, nodeSqliteDriver } from "../../src/selfhost/d1-adapter";
+import { runSelfHostMigrations } from "../../src/selfhost/migrate";
+
+function sqliteDb(): { db: ReturnType<typeof createD1Adapter>; raw: DatabaseSync } {
+  const raw = new DatabaseSync(":memory:");
+  return { db: createD1Adapter(nodeSqliteDriver(raw as never)), raw };
+}
+
+function migrationDir(files: Record<string, string>): string {
+  const dir = mkdtempSync(join(tmpdir(), "gtmig-atomic-"));
+  for (const [name, sql] of Object.entries(files)) writeFileSync(join(dir, name), sql);
+  return dir;
+}
+
+// #9027: each statement ran individually and the file was recorded applied only after the LAST one succeeded,
+// so a crash mid-file re-ran the WHOLE file next boot. Harmless for pure DDL; actively destructive for the
+// several migrations carrying real DML (the table-rebuild INSERT ... SELECT pattern, plus UPDATE/DELETE steps),
+// where a re-run either double-inserts or raises a PK conflict that — not matching the tolerated "already
+// exists" shape — throws and bricks boot. With deploys currently ending in SIGKILL under load (#9007), a
+// mid-migration kill is not hypothetical.
+describe("self-host migrations are all-or-nothing per file (#9027)", () => {
+  it("rolls the whole file back when a later statement fails, leaving no partial DDL and no ledger row", async () => {
+    const { db, raw } = sqliteDb();
+    const dir = migrationDir({
+      "0001_partial.sql": "CREATE TABLE keep (id INTEGER);\nCREATE TABLE dropme (id INTEGER);\nTHIS IS NOT VALID SQL;",
+    });
+
+    await expect(runSelfHostMigrations(db, dir)).rejects.toThrow();
+
+    const tables = raw.prepare("SELECT name FROM sqlite_master WHERE type='table'").all() as Array<{ name: string }>;
+    // Neither table survives: before this fix `keep` and `dropme` were both left behind, and the file was NOT
+    // recorded — so the next boot re-ran it from the top against a schema that had already half-changed.
+    expect(tables.map((t) => t.name)).not.toContain("dropme");
+    expect(tables.map((t) => t.name)).not.toContain("keep");
+  });
+
+  it("does not double-apply DML when a file is retried after a crash", async () => {
+    const { db, raw } = sqliteDb();
+    raw.exec("CREATE TABLE src (id INTEGER PRIMARY KEY); INSERT INTO src (id) VALUES (1), (2);");
+    // The table-rebuild shape several real migrations use, with a failing tail standing in for the crash.
+    const dir = migrationDir({
+      "0001_rebuild.sql": "CREATE TABLE dst (id INTEGER PRIMARY KEY);\nINSERT INTO dst (id) SELECT id FROM src;\nBROKEN;",
+    });
+
+    await expect(runSelfHostMigrations(db, dir)).rejects.toThrow();
+
+    const dstExists = (raw.prepare("SELECT COUNT(*) AS c FROM sqlite_master WHERE type='table' AND name='dst'").get() as { c: number }).c;
+    // The INSERT ... SELECT was rolled back with everything else, so a retry starts from a clean slate rather
+    // than inserting the same rows a second time.
+    expect(dstExists).toBe(0);
+  });
+
+  it("records the ledger row inside the same transaction, so 'applied' and 'actually applied' cannot diverge", async () => {
+    const { db, raw } = sqliteDb();
+    const dir = migrationDir({ "0001_ok.sql": "CREATE TABLE ok (id INTEGER);" });
+
+    expect(await runSelfHostMigrations(db, dir)).toBe(1);
+    const ledger = raw.prepare("SELECT name FROM _selfhost_migrations").all() as Array<{ name: string }>;
+    expect(ledger.map((row) => row.name)).toEqual(["0001_ok.sql"]);
+    expect(await runSelfHostMigrations(db, dir)).toBe(0);
+  });
+
+  it("still heals a drifted database through the per-statement fallback, and says so", async () => {
+    const { db, raw } = sqliteDb();
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    const dir = migrationDir({
+      "0001_add_x.sql": "CREATE TABLE t (id INTEGER);\nALTER TABLE t ADD COLUMN x INTEGER;",
+      // A renumbered-migration collision re-adds the same column: the atomic attempt fails, and the tolerant
+      // path must still record the file rather than crash-looping the boot.
+      "0002_readd_x.sql": "ALTER TABLE t ADD COLUMN x INTEGER;",
+    });
+
+    expect(await runSelfHostMigrations(db, dir)).toBe(2);
+    expect(warn.mock.calls.some(([line]) => String(line).includes("selfhost_migration_transaction_failed"))).toBe(true);
+    const ledger = raw.prepare("SELECT COUNT(*) AS c FROM _selfhost_migrations").get() as { c: number };
+    expect(ledger.c).toBe(2);
+    vi.restoreAllMocks();
+  });
+
+  it("applies a file whose final statement has no trailing semicolon", async () => {
+    const { db, raw } = sqliteDb();
+    // The ledger INSERT is concatenated onto the file's statements, so an unterminated tail would otherwise
+    // fuse onto it and fail to parse.
+    const dir = migrationDir({ "0001_tail.sql": "CREATE TABLE tail (id INTEGER)" });
+
+    expect(await runSelfHostMigrations(db, dir)).toBe(1);
+    const tables = raw.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='tail'").all();
+    expect(tables).toHaveLength(1);
+  });
+
+  it("falls back to the pre-existing path when the adapter has no transactional surface (real D1)", async () => {
+    const { db, raw } = sqliteDb();
+    const dir = migrationDir({ "0001_plain.sql": "CREATE TABLE plain (id INTEGER);" });
+    // Real Cloudflare D1 exposes no multi-statement transaction; migrations must still apply there.
+    const withoutTransaction = { ...(db as unknown as Record<string, unknown>) };
+    delete withoutTransaction.execTransaction;
+
+    expect(await runSelfHostMigrations(withoutTransaction as never, dir)).toBe(1);
+    expect(raw.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='plain'").all()).toHaveLength(1);
+  });
+});
diff --git a/test/unit/selfhost-pg-adapter.test.ts b/test/unit/selfhost-pg-adapter.test.ts
index ba9ef0a85..c5d3b3597 100644
--- a/test/unit/selfhost-pg-adapter.test.ts
+++ b/test/unit/selfhost-pg-adapter.test.ts
@@ -110,3 +110,58 @@ describe("createPgAdapter (#977 self-host D1-over-Postgres)", () => {
     expect(await createPgAdapter(makeMockPool([])).dump()).toBeInstanceOf(ArrayBuffer);
   });
 });
+
+// #9027: migrations need all-or-nothing execution per file, and `exec` cannot give it — each `exec` call takes
+// whatever connection the pool hands out, so a BEGIN and its COMMIT issued as separate calls can land on
+// different connections and mean nothing. This surface exists to make the transaction a single session.
+describe("execTransaction runs a script as one transaction (#9027)", () => {
+  it("wraps the script in BEGIN/COMMIT on a single dedicated client", async () => {
+    const pool = makeMockPool();
+    const db = createPgAdapter(pool) as unknown as { execTransaction(sql: string): Promise<void> };
+
+    await db.execTransaction("CREATE TABLE t (id INTEGER);");
+
+    expect(pool.queries.map((q) => q.sql)).toEqual(["BEGIN", "CREATE TABLE t (id INTEGER);", "COMMIT"]);
+  });
+
+  it("rolls back and rethrows when the script fails", async () => {
+    const queries: string[] = [];
+    let released = 0;
+    const client = {
+      async query(sql: string) {
+        queries.push(String(sql));
+        if (String(sql).includes("BROKEN")) throw new Error("syntax error");
+        return { rows: [], rowCount: 0 };
+      },
+      release() {
+        released += 1;
+      },
+    };
+    const pool = { async connect() { return client as unknown as PoolClient; } } as unknown as Pool;
+    const db = createPgAdapter(pool) as unknown as { execTransaction(sql: string): Promise<void> };
+
+    await expect(db.execTransaction("BROKEN;")).rejects.toThrow("syntax error");
+    expect(queries).toEqual(["BEGIN", "BROKEN;", "ROLLBACK"]);
+    // A failed transaction must never be returned to the pool still open.
+    expect(released).toBe(1);
+  });
+
+  it("still releases the client and reports the ORIGINAL error when the rollback itself fails", async () => {
+    let released = 0;
+    const client = {
+      async query(sql: string) {
+        if (String(sql) === "BEGIN") return { rows: [], rowCount: 0 };
+        throw new Error(String(sql) === "ROLLBACK" ? "connection lost" : "syntax error");
+      },
+      release() {
+        released += 1;
+      },
+    };
+    const pool = { async connect() { return client as unknown as PoolClient; } } as unknown as Pool;
+    const db = createPgAdapter(pool) as unknown as { execTransaction(sql: string): Promise<void> };
+
+    // The rollback failure must not mask why the migration actually failed.
+    await expect(db.execTransaction("BROKEN;")).rejects.toThrow("syntax error");
+    expect(released).toBe(1);
+  });
+});
diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts
index 9d605ab15..137d8360d 100644
--- a/test/unit/selfhost-pg-queue.test.ts
+++ b/test/unit/selfhost-pg-queue.test.ts
@@ -281,7 +281,66 @@ describe("createPgQueue (durable #977)", () => {
     const q = createPgQueue(m.pool, async () => undefined);
     await q.init();
     expect(m.pool.query).toHaveBeenCalledWith(expect.stringContaining("CREATE TABLE IF NOT EXISTS _selfhost_jobs"));
-    expect(m.pool.query).toHaveBeenCalledWith(expect.stringContaining("status='processing'"));
+    expect(m.pool.query).toHaveBeenCalledWith(expect.stringContaining("status='processing'"), expect.anything());
+  });
+
+  // #9023: boot recovery used to re-pend EVERY processing row unconditionally. On Postgres the queue is
+  // deliberately multi-instance, so during any overlapped or rolling deploy "every processing row" includes
+  // jobs a SIBLING is running right now -- both processes then ran the same PR pass, producing duplicate gate
+  // check-runs and verdict thrash. Only a row whose lease has actually expired can belong to a dead process.
+  describe("boot recovery respects the processing lease (#9023)", () => {
+    async function initWithTimeout(timeoutMs: string | undefined, fn: ReturnType<typeof vi.fn>): Promise<void> {
+      const old = process.env.QUEUE_PROCESSING_TIMEOUT_MS;
+      if (timeoutMs === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS;
+      else process.env.QUEUE_PROCESSING_TIMEOUT_MS = timeoutMs;
+      try {
+        await createPgQueue({ query: fn } as unknown as Pool, async () => undefined).init();
+      } finally {
+        if (old === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS;
+        else process.env.QUEUE_PROCESSING_TIMEOUT_MS = old;
+      }
+    }
+
+    it("scopes the recovery scan to rows older than the lease, not to every processing row", async () => {
+      const seen: Array<{ sql: string; params: unknown[] }> = [];
+      const fn = vi.fn().mockImplementation(async (sql: unknown, params?: unknown[]) => {
+        seen.push({ sql: String(sql), params: params ?? [] });
+        return { rows: [], rowCount: 0 };
+      });
+      await initWithTimeout("60000", fn);
+
+      const scan = seen.find((call) => call.sql.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1"));
+      expect(scan).toBeDefined();
+      // The cutoff is "now minus the lease", the same definition the runtime reaper uses -- boot and
+      // steady-state must not disagree about what counts as abandoned.
+      expect(Number(scan?.params[0])).toBeLessThanOrEqual(Date.now() - 60000);
+    });
+
+    it("re-pends only under a compare-and-swap, so a sibling finishing first cannot be resurrected", async () => {
+      const seen: string[] = [];
+      const fn = vi.fn().mockImplementation(async (sql: unknown) => {
+        const q = String(sql);
+        seen.push(q);
+        if (q.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1")) {
+          return { rows: [{ id: "7", payload: JSON.stringify({ type: "agent-regate-pr" }), job_key: null }], rowCount: 1 };
+        }
+        return { rows: [], rowCount: 0 };
+      });
+      await initWithTimeout("60000", fn);
+
+      expect(seen.some((q) => q.includes("SET status='pending', run_after=$1 WHERE id=$2 AND status='processing'"))).toBe(true);
+    });
+
+    it("recovers nothing when the lease is disabled — with no lease, nothing can be PROVEN abandoned", async () => {
+      const seen: string[] = [];
+      const fn = vi.fn().mockImplementation(async (sql: unknown) => {
+        seen.push(String(sql));
+        return { rows: [], rowCount: 0 };
+      });
+      await initWithTimeout("0", fn);
+
+      expect(seen.some((q) => q.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1"))).toBe(false);
+    });
   });
 
   it("init() handles null rowCount from the recovery query (rowCount ?? 0 nullish arm)", async () => {
@@ -2943,18 +3002,195 @@ describe("createPgQueue (durable #977)", () => {
     expect(logged.some((line) => line.includes("selfhost_queue_pump_crashed") && line.includes("connection terminated unexpectedly"))).toBe(true);
   });
 
+  // #9023: a reclaim never charged an attempt, so a job that reliably wedges past its lease -- an infinite
+  // loop, a hang against an unresponsive dependency -- was requeued forever with `attempts` frozen at whatever
+  // the last real failure left it. It could never reach the dead-letter threshold that exists precisely to stop
+  // a poison job from burning the queue.
+  describe("reclaimed jobs consume their retry budget (#9023)", () => {
+    async function drainWithProcessingRow(row: Record<string, unknown>): Promise<string[]> {
+      const old = process.env.QUEUE_PROCESSING_TIMEOUT_MS;
+      process.env.QUEUE_PROCESSING_TIMEOUT_MS = "1";
+      const statements: string[] = [];
+      try {
+        // Boot recovery now shares the reaper's lease-scoped scan (#9023), so the row is only offered AFTER
+        // init() -- otherwise this would exercise boot recovery, not the runtime reclaim.
+        let booted = false;
+        let served = false;
+        const fn = vi.fn().mockImplementation(async (sql: unknown) => {
+          const q = String(sql);
+          statements.push(q);
+          if (booted && q.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1")) {
+            if (served) return { rows: [], rowCount: 0 };
+            served = true;
+            return { rows: [row], rowCount: 1 };
+          }
+          return { rows: [], rowCount: 1 };
+        });
+        const q = createPgQueue({ query: fn } as unknown as Pool, async () => undefined);
+        await q.init();
+        booted = true;
+        statements.length = 0;
+        await q.drain();
+      } finally {
+        if (old === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS;
+        else process.env.QUEUE_PROCESSING_TIMEOUT_MS = old;
+      }
+      return statements;
+    }
+
+    it("requeues with attempts incremented while the job still has budget", async () => {
+      const statements = await drainWithProcessingRow({ id: "5", payload: JSON.stringify({ type: "agent-regate-pr" }), job_key: null, attempts: 0 });
+      expect(statements.some((q) => q.includes("SET status='pending', run_after=$1, attempts=$2"))).toBe(true);
+    });
+
+    it("dead-letters a job that has wedged past its lease as many times as the retry cap allows", async () => {
+      const statements = await drainWithProcessingRow({ id: "6", payload: JSON.stringify({ type: "agent-regate-pr" }), job_key: null, attempts: 4 });
+      expect(statements.some((q) => q.includes("SET status='dead', attempts=$1") && q.includes("AND status='processing'"))).toBe(true);
+      // It must NOT also be requeued -- that would be the forever-loop this fixes, just one attempt later.
+      expect(statements.some((q) => q.includes("SET status='pending', run_after=$1, attempts=$2"))).toBe(false);
+    });
+  });
+
+  // #9023: `run_after` doubles as the lease timestamp, so without a heartbeat the lease timeout was a ceiling on
+  // job DURATION rather than on liveness -- a legitimately long pass (a large PR, a slow AI review, a
+  // rate-limited GitHub window) got reclaimed and double-run by a sibling purely for taking too long. The
+  // activeJobIds guard only covers THIS process's in-memory set, which is the wrong scope for a shared queue.
+  it("heartbeats the lease of a job it is actively running (#9023)", async () => {
+    const old = process.env.QUEUE_PROCESSING_TIMEOUT_MS;
+    process.env.QUEUE_PROCESSING_TIMEOUT_MS = "3000"; // → a 1s heartbeat interval
+    vi.useFakeTimers();
+    try {
+      const m = makePool();
+      m.enqueueJob("9", { type: "agent-regate-pr" }, 0);
+      let release: (() => void) | undefined;
+      const running = new Promise<void>((resolve) => {
+        release = resolve;
+      });
+      const q = createPgQueue(m.pool, async () => running);
+      await q.init();
+      const drained = q.drain();
+      await vi.advanceTimersByTimeAsync(2500);
+
+      const poolCalls = (): unknown[][] => (m.fn as unknown as ReturnType<typeof vi.fn>).mock.calls;
+      const heartbeats = poolCalls().filter((call: unknown[]) => String(call[0]).includes("SET run_after=$1 WHERE id=$2 AND status='processing'"));
+      expect(heartbeats.length).toBeGreaterThanOrEqual(1);
+
+      release?.();
+      await vi.advanceTimersByTimeAsync(10);
+      await drained;
+      // The timer must be cleared on completion, or a finished job keeps renewing a lease it no longer holds.
+      const after = poolCalls().length;
+      await vi.advanceTimersByTimeAsync(5000);
+      expect(poolCalls().length).toBe(after);
+    } finally {
+      vi.useRealTimers();
+      if (old === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS;
+      else process.env.QUEUE_PROCESSING_TIMEOUT_MS = old;
+    }
+  });
+
+  // The pg driver can return a null rowCount for some UPDATE results; the recovery/reclaim counters must read
+  // that as "nothing changed" rather than crashing or over-reporting.
+  describe("#9023 recovery paths tolerate a null rowCount", () => {
+    async function runWith(fn: ReturnType<typeof vi.fn>, afterInit: (q: ReturnType<typeof createPgQueue>) => Promise<void>): Promise<void> {
+      const old = process.env.QUEUE_PROCESSING_TIMEOUT_MS;
+      process.env.QUEUE_PROCESSING_TIMEOUT_MS = "1";
+      try {
+        const q = createPgQueue({ query: fn } as unknown as Pool, async () => undefined);
+        await q.init();
+        await afterInit(q);
+      } finally {
+        if (old === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS;
+        else process.env.QUEUE_PROCESSING_TIMEOUT_MS = old;
+      }
+    }
+
+    it("counts nothing recovered at boot when the compare-and-swap reports a null rowCount", async () => {
+      const logs: string[] = [];
+      vi.spyOn(console, "log").mockImplementation((...args: unknown[]) => void logs.push(String(args[0])));
+      const fn = vi.fn().mockImplementation(async (sql: unknown) => {
+        const q = String(sql);
+        if (q.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1")) {
+          return { rows: [{ id: "1", payload: JSON.stringify({ type: "agent-regate-pr" }), job_key: null, attempts: 0 }], rowCount: 1 };
+        }
+        if (q.includes("SET status='pending', run_after=$1 WHERE id=$2 AND status='processing'")) return { rows: [], rowCount: null };
+        return { rows: [], rowCount: 1 };
+      });
+      await runWith(fn, async () => undefined);
+
+      expect(logs.map((line) => JSON.parse(line) as Record<string, unknown>).some((log) => log.event === "selfhost_queue_recovered")).toBe(false);
+      vi.restoreAllMocks();
+    });
+
+    it("does not report a dead-letter it did not actually win", async () => {
+      let booted = false;
+      let served = false;
+      const fn = vi.fn().mockImplementation(async (sql: unknown) => {
+        const q = String(sql);
+        if (booted && q.includes("FROM _selfhost_jobs WHERE status='processing' AND run_after<=$1")) {
+          if (served) return { rows: [], rowCount: 0 };
+          served = true;
+          return { rows: [{ id: "2", payload: JSON.stringify({ type: "agent-regate-pr" }), job_key: null, attempts: 4 }], rowCount: 1 };
+        }
+        // A sibling won the row first, so the guarded UPDATE matched nothing.
+        if (q.includes("SET status='dead', attempts=$1")) return { rows: [], rowCount: null };
+        return { rows: [], rowCount: 1 };
+      });
+      await runWith(fn, async (q) => {
+        booted = true;
+        await q.drain();
+      });
+      expect(served).toBe(true);
+    });
+  });
+
+  it("#9023 heartbeat failures are absorbed — a lost renewal just lets the lease age normally", async () => {
+    const old = process.env.QUEUE_PROCESSING_TIMEOUT_MS;
+    process.env.QUEUE_PROCESSING_TIMEOUT_MS = "3000";
+    vi.useFakeTimers();
+    try {
+      const m = makePool();
+      m.enqueueJob("10", { type: "agent-regate-pr" }, 0);
+      let release: (() => void) | undefined;
+      const running = new Promise<void>((resolve) => {
+        release = resolve;
+      });
+      const original = m.pool.query.bind(m.pool);
+      (m.pool as unknown as { query: (sql: string, params?: unknown[]) => Promise<unknown> }).query = async (sql: string, params?: unknown[]) => {
+        if (String(sql).includes("SET run_after=$1 WHERE id=$2 AND status='processing'")) throw new Error("connection lost");
+        return original(sql, params as never);
+      };
+      const q = createPgQueue(m.pool, async () => running);
+      await q.init();
+      const drained = q.drain();
+      await vi.advanceTimersByTimeAsync(2500);
+      release?.();
+      await vi.advanceTimersByTimeAsync(10);
+      // The job still completes: a heartbeat is an optimization, never a precondition.
+      await expect(drained).resolves.toBeUndefined();
+    } finally {
+      vi.useRealTimers();
+      if (old === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS;
+      else process.env.QUEUE_PROCESSING_TIMEOUT_MS = old;
+    }
+  });
+
   it("pump absorbs a reclaimExpiredProcessingJobs() pool failure instead of crashing the process (regression for #2498)", async () => {
     const oldTimeout = process.env.QUEUE_PROCESSING_TIMEOUT_MS;
     process.env.QUEUE_PROCESSING_TIMEOUT_MS = "1";
     try {
+      // #9023 made boot recovery use the SAME lease-scoped scan as the reaper, so the injection has to start
+      // AFTER init() -- otherwise this exercises a failing boot, not the pump absorbing a reaper failure.
+      let failReaper = false;
       const fn = vi.fn().mockImplementation(async (sql: unknown) => {
-        if (String(sql).includes("WHERE status='processing' AND run_after<=$1")) throw new Error("connection terminated unexpectedly");
+        if (failReaper && String(sql).includes("WHERE status='processing' AND run_after<=$1")) throw new Error("connection terminated unexpectedly");
         return { rows: [], rowCount: 0 };
       });
       const pool = { query: fn } as unknown as Pool;
       const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined);
       const q = createPgQueue(pool, async () => undefined);
       await q.init();
+      failReaper = true;
 
       await expect(q.drain()).resolves.toBeUndefined();