fix: harden code queue postgres rotation

src/components/microservices/code-queue/src/database-resilience.ts

@@ -0,0 +1,190 @@
+import type { JsonValue } from "./types";
+
+export type DatabaseTransientKind =
+  | "deadlock"
+  | "connection-closed"
+  | "connection-reset"
+  | "connection-timeout"
+  | "connection-terminated"
+  | "socket-write-null"
+  | "unknown";
+
+export interface DatabaseTransientClassification {
+  transient: boolean;
+  retryable: boolean;
+  kind: DatabaseTransientKind | null;
+  source: "postgres-deadlock" | "postgres-client" | "unknown";
+  infrastructureBlocker: boolean;
+  message: string;
+  evidence: string[];
+  recommendedAction: string;
+}
+
+export interface DirtyFlushBatch {
+  taskIds: string[];
+  queueIds: string[];
+  takenAt: string;
+}
+
+export interface DatabaseStorageHealthInput {
+  postgresReady: boolean;
+  dirtyTaskCount: number;
+  dirtyQueueCount: number;
+  lastError: unknown;
+  flushInFlight: boolean;
+  clientRotationCount: number;
+  lastClientRotationAt: string | null;
+  lastClientRotationReason: string | null;
+  transientUncaughtSuppressedCount: number;
+  lastTransientUncaughtSuppressedAt: string | null;
+  consecutiveFlushFailures: number;
+  lastFlushFailureAt: string | null;
+  nextFlushRetryAt: string | null;
+}
+
+function errorField(error: unknown, key: "code" | "errno"): string {
+  if (typeof error !== "object" || error === null) return "";
+  const value = (error as Record<string, unknown>)[key];
+  return typeof value === "string" ? value : "";
+}
+
+export function databaseErrorText(error: unknown): string {
+  if (error instanceof Error) return `${error.name}: ${error.message}\n${error.stack ?? ""}\n${errorField(error, "code")}\n${errorField(error, "errno")}`;
+  return String(error);
+}
+
+function collectEvidence(text: string, patterns: RegExp[], limit = 5): string[] {
+  const evidence: string[] = [];
+  for (const pattern of patterns) {
+    for (const match of text.matchAll(pattern)) {
+      const value = String(match[0] ?? "").replace(/\s+/gu, " ").trim();
+      if (value.length > 0 && !evidence.includes(value)) evidence.push(value.slice(0, 220));
+      if (evidence.length >= limit) return evidence;
+    }
+  }
+  return evidence;
+}
+
+export function classifyTransientDatabaseError(error: unknown): DatabaseTransientClassification {
+  const text = databaseErrorText(error);
+  const lower = text.toLowerCase();
+  const deadlock = /deadlock detected/iu.test(text);
+  const code = `${errorField(error, "code")}\n${errorField(error, "errno")}`;
+  const postgresClientCode = /CONNECTION_CLOSED|CONNECTION_DESTROYED|CONNECTION_ENDED|CONNECT_TIMEOUT/iu.test(code);
+  const postgresClient = postgresClientCode || /node_modules\/postgres|postgres(?:@\d+(?:\.\d+){0,2})?\/src|postgres\/src|d601-tcp-egress-gateway|:15432|unidesk_code_queue|set_config/iu.test(text);
+  const combined = `${text}\n${code}`;
+  const socketWriteNull = /socket\.write|null is not an object|cannot read propert(?:y|ies) ['"]?write|evaluating ['"]socket\.write/iu.test(combined);
+  const connectionClosed = /CONNECTION_CLOSED|socket connection was closed|connection closed/iu.test(combined);
+  const connectionReset = /ECONNRESET|EPIPE|broken pipe/iu.test(combined);
+  const connectionTimeout = /ETIMEDOUT|CONNECT_TIMEOUT|Connection timeout|connect timed out/iu.test(combined);
+  const connectionTerminated = /Connection terminated|CONNECTION_DESTROYED|CONNECTION_ENDED/iu.test(combined);
+  const transient = deadlock || (postgresClient && (socketWriteNull || connectionClosed || connectionReset || connectionTimeout || connectionTerminated));
+  const kind: DatabaseTransientKind | null = deadlock
+    ? "deadlock"
+    : socketWriteNull && postgresClient
+      ? "socket-write-null"
+      : connectionClosed && postgresClient
+        ? "connection-closed"
+        : connectionReset && postgresClient
+          ? "connection-reset"
+          : connectionTimeout && postgresClient
+            ? "connection-timeout"
+            : connectionTerminated && postgresClient
+              ? "connection-terminated"
+              : transient
+                ? "unknown"
+                : null;
+  const evidence = collectEvidence(lower, [
+    /postgres(?:@\d+(?:\.\d+){0,2})?\/src\/connection\.js:\d+/giu,
+    /node_modules\/postgres\/src\/connection\.js:\d+/giu,
+    /connection_closed/giu,
+    /socket\.write/giu,
+    /null is not an object/giu,
+    /d601-tcp-egress-gateway[^\s'"]*/giu,
+    /:15432/giu,
+    /deadlock detected/giu,
+    /econnreset|epipe|etimedout|connect_timeout/giu,
+  ]);
+  return {
+    transient,
+    retryable: transient,
+    kind,
+    source: deadlock ? "postgres-deadlock" : postgresClient ? "postgres-client" : "unknown",
+    infrastructureBlocker: transient,
+    message: transient
+      ? "PostgreSQL client connection was transiently closed; rotate the client, keep dirty writes queued, and retry without crashing the scheduler."
+      : "Error is not recognized as a transient PostgreSQL client failure.",
+    evidence,
+    recommendedAction: transient
+      ? "Treat as Code Queue infrastructure, not task/business failure; inspect scheduler storage health and let bounded dirty-flush retry run before duplicating work."
+      : "Handle through the caller's normal error path.",
+  };
+}
+
+export function isTransientDatabaseError(error: unknown): boolean {
+  return classifyTransientDatabaseError(error).transient;
+}
+
+export function takeDirtyFlushBatch(taskIds: Set<string>, queueIds: Set<string>, nowIso: string): DirtyFlushBatch {
+  const batch = {
+    taskIds: Array.from(taskIds).sort(),
+    queueIds: Array.from(queueIds).sort(),
+    takenAt: nowIso,
+  };
+  taskIds.clear();
+  queueIds.clear();
+  return batch;
+}
+
+export function restoreDirtyFlushBatch(batch: DirtyFlushBatch, taskIds: Set<string>, queueIds: Set<string>): void {
+  for (const taskId of batch.taskIds) taskIds.add(taskId);
+  for (const queueId of batch.queueIds) queueIds.add(queueId);
+}
+
+export function databaseStorageHealth(input: DatabaseStorageHealthInput): JsonValue {
+  const classification = input.lastError === null ? null : classifyTransientDatabaseError(input.lastError);
+  const dirtyCount = Math.max(0, input.dirtyTaskCount) + Math.max(0, input.dirtyQueueCount);
+  const infrastructureBlocker = input.postgresReady !== true || input.lastError !== null || input.consecutiveFlushFailures > 0;
+  const status = input.postgresReady !== true
+    ? "unavailable"
+    : infrastructureBlocker
+      ? "degraded"
+      : input.flushInFlight || dirtyCount > 0
+        ? "recovering"
+        : "ready";
+  const signal = infrastructureBlocker
+    ? [{
+      id: classification?.kind === "socket-write-null" ? "postgres-client-socket-write-null" : "postgres-connection-rotation",
+      category: "infrastructure-blocker",
+      severity: input.postgresReady !== true ? "critical" : "high",
+      retryable: classification?.retryable ?? true,
+      bounded: true,
+      message: classification?.message ?? "Code Queue PostgreSQL storage is not ready or has pending dirty writes after a flush failure.",
+      evidence: classification?.evidence ?? [],
+      commanderAction: classification?.recommendedAction ?? "Inspect Code Queue storage health and scheduler logs before treating affected tasks as failed.",
+      source: "code-queue.storage.postgres",
+    } as Record<string, JsonValue>]
+    : [];
+  return {
+    status,
+    degraded: infrastructureBlocker,
+    infrastructureBlocker,
+    postgresReady: input.postgresReady,
+    dirtyTaskCount: input.dirtyTaskCount,
+    dirtyQueueCount: input.dirtyQueueCount,
+    dirtyCount,
+    flushInFlight: input.flushInFlight,
+    consecutiveFlushFailures: input.consecutiveFlushFailures,
+    lastFlushFailureAt: input.lastFlushFailureAt,
+    nextFlushRetryAt: input.nextFlushRetryAt,
+    clientRotationCount: input.clientRotationCount,
+    lastClientRotationAt: input.lastClientRotationAt,
+    lastClientRotationReason: input.lastClientRotationReason,
+    transientUncaughtSuppressedCount: input.transientUncaughtSuppressedCount,
+    lastTransientUncaughtSuppressedAt: input.lastTransientUncaughtSuppressedAt,
+    lastErrorKind: classification?.kind ?? null,
+    lastErrorTransient: classification?.transient ?? false,
+    lastErrorMessage: input.lastError === null ? null : databaseErrorText(input.lastError).split(/\r?\n/u)[0]?.slice(0, 500) ?? null,
+    signals: signal as unknown as JsonValue,
+  };
+}

src/components/microservices/code-queue/src/index.ts

@@ -119,6 +119,7 @@ import {
 } from "./queue-api";
 import { buildExecutionDiagnostics, buildSchedulerHeartbeat, normalizeSchedulerHeartbeatStaleMs, schedulerHeartbeatStaleMs, schedulerHeartbeatStaleMsMax, staleRecoveryCandidate } from "./execution-diagnostics";
 import { ReferenceTaskLookupError, configureReferences, injectReferencedTaskContext, taskReferenceIds } from "./references";
+import { classifyTransientDatabaseError, databaseStorageHealth, isTransientDatabaseError, restoreDirtyFlushBatch, takeDirtyFlushBatch } from "./database-resilience";
 import {
   applyOaTraceStatsToTaskJson,
   configureOaEvents,
@@ -236,8 +237,17 @@ function createSqlClient(): SqlClient {
 
 let databaseReady = false;
 let databaseLastError: string | null = null;
+let databaseLastErrorDetail: unknown = null;
 let databaseFlushTimer: ReturnType<typeof setTimeout> | null = null;
 let databaseFlushInFlight = false;
+let databaseFlushFailureCount = 0;
+let databaseLastFlushFailureAt: string | null = null;
+let databaseNextFlushRetryAt: string | null = null;
+let databaseClientRotationCount = 0;
+let databaseLastClientRotationAt: string | null = null;
+let databaseLastClientRotationReason: string | null = null;
+let databaseTransientUncaughtSuppressedCount = 0;
+let databaseLastTransientUncaughtSuppressedAt: string | null = null;
 const dirtyDatabaseTaskIds = new Set<string>();
 const dirtyDatabaseQueueIds = new Set<string>();
 const workdirRecords = new Map<string, WorkdirRecord>();
@@ -1220,29 +1230,27 @@ function errorToJson(error: unknown): JsonValue {
   return String(error);
 }
 
-function transientDatabaseErrorMessage(error: unknown): string {
-  const message = error instanceof Error ? error.message : String(error);
-  const stack = error instanceof Error ? error.stack ?? "" : "";
-  return `${message}\n${stack}`;
-}
-
-function isTransientDatabaseError(error: unknown): boolean {
-  const text = transientDatabaseErrorMessage(error);
-  if (/deadlock detected/iu.test(text)) return true;
-  const looksLikePostgresClient = /node_modules\/postgres|postgres\/src|d601-tcp-egress-gateway|:15432|unidesk_code_queue|set_config/iu.test(text);
-  if (!looksLikePostgresClient) return false;
-  return /CONNECTION_CLOSED|ECONNRESET|EPIPE|ETIMEDOUT|Connection terminated|socket connection was closed|socket\.write|null is not an object/iu.test(text);
-}
-
 function transientDatabaseRetryDelayMs(attempt: number): number {
   return Math.min(2000, 150 * Math.max(1, attempt));
 }
 
+function scheduleDatabaseFlushRetry(delayMs: number): void {
+  if (serviceRoleReadOnly(config.serviceRole)) return;
+  if (!databaseReady || shutdownRequested) return;
+  if (dirtyDatabaseTaskIds.size === 0 && dirtyDatabaseQueueIds.size === 0) return;
+  if (databaseFlushTimer !== null) return;
+  const boundedDelayMs = Math.max(100, Math.min(30_000, Math.floor(delayMs)));
+  scheduleDatabaseFlush(boundedDelayMs, { retry: true });
+}
+
 async function rotateSqlClientAfterTransientError(reason: string, error: unknown): Promise<void> {
-  databaseLastError = databaseErrorMessage(error);
+  setDatabaseLastError(error);
   if (sqlRotationInFlight !== null) return sqlRotationInFlight;
   const previous = sql;
   sql = createSqlClient();
+  databaseClientRotationCount += 1;
+  databaseLastClientRotationAt = nowIso();
+  databaseLastClientRotationReason = reason;
   sqlRotationInFlight = previous.end({ timeout: 1 })
     .catch((closeError) => logger("warn", "database_client_close_failed", { reason, error: errorToJson(closeError) }))
     .finally(() => {
@@ -1259,7 +1267,8 @@ async function withTransientDatabaseRetries<T>(operation: string, action: () =>
       return await action();
     } catch (error) {
       lastError = error;
-      if (!isTransientDatabaseError(error) || attempt >= attempts) throw error;
+      const classification = classifyTransientDatabaseError(error);
+      if (!classification.transient || attempt >= attempts) throw error;
       logger("warn", "database_transient_retry", { operation, attempt, attempts, delayMs: transientDatabaseRetryDelayMs(attempt), error: errorToJson(error) });
       await rotateSqlClientAfterTransientError(operation, error);
       await Bun.sleep(transientDatabaseRetryDelayMs(attempt));
@@ -1291,6 +1300,16 @@ function databaseErrorMessage(error: unknown): string {
   return String(error);
 }
 
+function setDatabaseLastError(error: unknown): void {
+  databaseLastError = databaseErrorMessage(error);
+  databaseLastErrorDetail = error;
+}
+
+function clearDatabaseLastError(): void {
+  databaseLastError = null;
+  databaseLastErrorDetail = null;
+}
+
 function markTaskDirty(taskId: string): void {
   dirtyDatabaseTaskIds.add(taskId);
   scheduleDatabaseFlush();
@@ -1319,14 +1338,15 @@ function runGarbageCollection(): void {
   if (typeof gc === "function") gc(true);
 }
 
-function scheduleDatabaseFlush(delayMs = config.databaseFlushIntervalMs): void {
+function scheduleDatabaseFlush(delayMs = config.databaseFlushIntervalMs, options: { retry?: boolean } = {}): void {
   if (serviceRoleReadOnly(config.serviceRole)) return;
   if (!databaseReady || (dirtyDatabaseTaskIds.size === 0 && dirtyDatabaseQueueIds.size === 0) || shutdownRequested) return;
   if (databaseFlushTimer !== null) return;
+  if (options.retry === true) databaseNextFlushRetryAt = new Date(Date.now() + Math.max(0, delayMs)).toISOString();
   databaseFlushTimer = setTimeout(() => {
     databaseFlushTimer = null;
     void flushDirtyTasksToDatabase().catch((error) => {
-      databaseLastError = databaseErrorMessage(error);
+      setDatabaseLastError(error);
       logger("error", "database_flush_failed", { error: errorToJson(error), dirtyTaskCount: dirtyDatabaseTaskIds.size });
     });
   }, delayMs);
@@ -2105,7 +2125,7 @@ function scheduleStartupDatabaseMaintenance(): void {
         durationMs: Math.round(performance.now() - started),
       });
     })().catch((error) => {
-      databaseLastError = databaseErrorMessage(error);
+      setDatabaseLastError(error);
       logger("warn", "database_startup_maintenance_failed", { error: errorToJson(error) });
     });
   }, 1000).unref?.();
@@ -2129,7 +2149,7 @@ async function findTaskForRead(taskId: string): Promise<QueueTask | null> {
     if (databaseTask === null) return hotTask;
     return shouldPreferHotTaskOverDatabase(hotTask, databaseTask) ? hotTask : databaseTask;
   } catch (error) {
-    databaseLastError = databaseErrorMessage(error);
+    setDatabaseLastError(error);
     logger("warn", "read_database_fallback", { taskId, error: errorToJson(error) });
     return hotTask;
   }
@@ -2144,7 +2164,7 @@ async function findTaskForMutation(taskId: string): Promise<QueueTask | null> {
     if (hotTask === null) return rememberHotTask(databaseTask);
     return shouldPreferHotTaskOverDatabase(hotTask, databaseTask) ? hotTask : reconcileHotTaskFromDatabase(databaseTask);
   } catch (error) {
-    databaseLastError = databaseErrorMessage(error);
+    setDatabaseLastError(error);
     logger("warn", "mutation_database_fallback", { taskId, error: errorToJson(error) });
     return hotTask;
   }
@@ -2170,11 +2190,10 @@ async function flushDirtyTasksToDatabase(force = false): Promise<void> {
     scheduleDatabaseFlush();
     return;
   }
-  const ids = Array.from(dirtyDatabaseTaskIds).sort();
-  const queueIds = Array.from(dirtyDatabaseQueueIds).sort();
+  const batch = takeDirtyFlushBatch(dirtyDatabaseTaskIds, dirtyDatabaseQueueIds, nowIso());
+  const ids = batch.taskIds;
+  const queueIds = batch.queueIds;
   if (ids.length === 0 && queueIds.length === 0) return;
-  dirtyDatabaseTaskIds.clear();
-  dirtyDatabaseQueueIds.clear();
   databaseFlushInFlight = true;
   let rejectedTaskIds: string[] = [];
   try {
@@ -2190,10 +2209,32 @@ async function flushDirtyTasksToDatabase(force = false): Promise<void> {
       }
       return currentRejectedTaskIds;
     }));
-    databaseLastError = null;
+    clearDatabaseLastError();
+    databaseFlushFailureCount = 0;
+    databaseLastFlushFailureAt = null;
+    databaseNextFlushRetryAt = null;
   } catch (error) {
-    for (const id of ids) dirtyDatabaseTaskIds.add(id);
-    for (const id of queueIds) dirtyDatabaseQueueIds.add(id);
+    restoreDirtyFlushBatch(batch, dirtyDatabaseTaskIds, dirtyDatabaseQueueIds);
+    setDatabaseLastError(error);
+    databaseFlushFailureCount += 1;
+    databaseLastFlushFailureAt = nowIso();
+    const classification = classifyTransientDatabaseError(error);
+    if (classification.transient) {
+      await rotateSqlClientAfterTransientError("flush-dirty-tasks-final", error);
+      const retryDelayMs = Math.min(30_000, Math.max(config.databaseFlushIntervalMs, transientDatabaseRetryDelayMs(databaseFlushFailureCount + 1) * Math.max(1, databaseFlushFailureCount)));
+      logger("error", "database_flush_degraded_retry_scheduled", {
+        operation: "flush-dirty-tasks",
+        force,
+        failureCount: databaseFlushFailureCount,
+        retryDelayMs,
+        dirtyTaskCount: dirtyDatabaseTaskIds.size,
+        dirtyQueueCount: dirtyDatabaseQueueIds.size,
+        classification: classification as unknown as JsonValue,
+        error: errorToJson(error),
+      });
+      scheduleDatabaseFlushRetry(retryDelayMs);
+      return;
+    }
     throw error;
   } finally {
     databaseFlushInFlight = false;
@@ -2368,7 +2409,7 @@ async function initDatabasePersistenceWithRetry(): Promise<void> {
       await initDatabasePersistence();
       return;
     } catch (error) {
-      databaseLastError = databaseErrorMessage(error);
+      setDatabaseLastError(error);
       const elapsedMs = Date.now() - started;
       logger("warn", "database_persistence_init_retry", { attempt, elapsedMs, error: errorToJson(error) });
       if (elapsedMs > 90_000 && !reportedLongWait) {
@@ -2923,7 +2964,23 @@ configureQueueApi({
   databaseLastError: () => databaseLastError,
   databaseReady: () => databaseReady,
   defaultQueueId,
+  dirtyDatabaseQueueCount: () => dirtyDatabaseQueueIds.size,
   dirtyDatabaseTaskCount: () => dirtyDatabaseTaskIds.size,
+  databaseStorageHealth: () => databaseStorageHealth({
+    postgresReady: databaseReady,
+    dirtyTaskCount: dirtyDatabaseTaskIds.size,
+    dirtyQueueCount: dirtyDatabaseQueueIds.size,
+    lastError: databaseLastErrorDetail ?? databaseLastError,
+    flushInFlight: databaseFlushInFlight,
+    clientRotationCount: databaseClientRotationCount,
+    lastClientRotationAt: databaseLastClientRotationAt,
+    lastClientRotationReason: databaseLastClientRotationReason,
+    transientUncaughtSuppressedCount: databaseTransientUncaughtSuppressedCount,
+    lastTransientUncaughtSuppressedAt: databaseLastTransientUncaughtSuppressedAt,
+    consecutiveFlushFailures: databaseFlushFailureCount,
+    lastFlushFailureAt: databaseLastFlushFailureAt,
+    nextFlushRetryAt: databaseNextFlushRetryAt,
+  }),
   jsonResponse,
   judgeFailRetryLimit,
   loadQueuesFromDatabase,
@@ -3489,9 +3546,24 @@ function latestAttemptExternalProvider429(task: QueueTask): boolean {
   return latestAttemptRunnerErrorClassification(task)?.externalProvider429 === true;
 }
 
+function latestAttemptStaleBadResume(task: QueueTask): boolean {
+  return latestAttemptRunnerErrorClassification(task)?.staleBadResume === true;
+}
+
+function clearStaleBadResumeThread(task: QueueTask): boolean {
+  if (!latestAttemptStaleBadResume(task) || task.codexThreadId === null) return false;
+  const previousThreadId = task.codexThreadId;
+  task.codexThreadId = null;
+  task.activeTurnId = null;
+  appendOutput(task, "system", `stale bad-resume thread ${previousThreadId} superseded; next retry will start a fresh Codex thread because app-server reported no rollout found for thread id\n`, "thread/resume-superseded");
+  logger("warn", "stale_bad_resume_thread_superseded", { taskId: task.id, previousThreadId });
+  return true;
+}
+
 function retryBackoffEvidence(task: QueueTask, completedAttempts: number, reason: string): JsonValue {
   const runnerClassification = latestAttemptRunnerErrorClassification(task);
   const externalProvider429 = runnerClassification?.externalProvider429 === true;
+  const staleBadResume = runnerClassification?.staleBadResume === true;
   const baseDelayMs = retryBackoffMs(completedAttempts);
   const jittered = externalProvider429
     ? retryBackoffWithJitterMs(task, completedAttempts, 30_000)
@@ -3505,9 +3577,12 @@ function retryBackoffEvidence(task: QueueTask, completedAttempts: number, reason
     delayMs: jittered.delayMs,
     maxDelayMs: retryBackoffMaxMs,
     externalProvider429,
+    staleBadResume,
     failureKind: runnerClassification?.failureKind ?? null,
     runnerDisposition: runnerClassification?.disposition ?? null,
-    commanderBenefit: "fewer failed runners and clearer retry/backoff evidence",
+    commanderBenefit: staleBadResume
+      ? "stale resume thread is superseded once and retried on a fresh Codex thread"
+      : "fewer failed runners and clearer retry/backoff evidence",
   };
 }
 
@@ -4072,6 +4147,7 @@ async function runTask(task: QueueTask): Promise<void> {
         const nextPrompt = judgeFailContinuationPrompt(task, judge, task.judgeFailCount);
         task.nextPrompt = nextPrompt;
         task.nextMode = "retry";
+        clearStaleBadResumeThread(task);
         setAttemptFeedbackPrompt(latestAttempt, nextPrompt, "judge-fail-retry", task.attempts.length + 1);
         task.updatedAt = nowIso();
         appendOutput(task, "system", `judge=fail treated as retry (${task.judgeFailCount}/${judgeFailRetryLimit}); appending continuation prompt to existing session\n`, "queue");
@@ -4135,6 +4211,7 @@ async function runTask(task: QueueTask): Promise<void> {
     const nextPrompt = retryPrompt(task, judge);
     task.nextPrompt = nextPrompt;
     task.nextMode = "retry";
+    clearStaleBadResumeThread(task);
     setAttemptFeedbackPrompt(latestAttempt, nextPrompt, judge.continuePrompt?.trim() ? "judge-continue-prompt" : "judge-retry-generated", task.attempts.length + 1);
     task.updatedAt = nowIso();
     persistTaskState(task);
@@ -4516,7 +4593,7 @@ function startSchedulerDatabasePoller(): void {
   const interval = setInterval(() => {
     if (!serviceReady || shutdownRequested) return;
     void refreshSchedulerTasksFromDatabase("poll").catch((error) => {
-      databaseLastError = databaseErrorMessage(error);
+      setDatabaseLastError(error);
       logger("warn", "scheduler_database_poll_failed", { error: errorToJson(error) });
     });
   }, config.schedulerPollIntervalMs);
@@ -4572,8 +4649,17 @@ function installShutdownHandlers(): void {
   process.once("SIGINT", stop);
   process.on("uncaughtException", (error) => {
     if (isTransientDatabaseError(error)) {
+      databaseTransientUncaughtSuppressedCount += 1;
+      databaseLastTransientUncaughtSuppressedAt = nowIso();
       void rotateSqlClientAfterTransientError("uncaught-exception", error).catch((closeError) => logger("warn", "database_client_rotate_uncaught_failed", { error: errorToJson(closeError) }));
-      logger("error", "transient_database_uncaught_exception_suppressed", { error: errorToJson(error) });
+      if (dirtyDatabaseTaskIds.size > 0 || dirtyDatabaseQueueIds.size > 0) scheduleDatabaseFlushRetry(config.databaseFlushIntervalMs);
+      logger("error", "transient_database_uncaught_exception_suppressed", {
+        suppressedCount: databaseTransientUncaughtSuppressedCount,
+        dirtyTaskCount: dirtyDatabaseTaskIds.size,
+        dirtyQueueCount: dirtyDatabaseQueueIds.size,
+        classification: classifyTransientDatabaseError(error) as unknown as JsonValue,
+        error: errorToJson(error),
+      });
       return;
     }
     logger("error", "uncaught_exception", { error: errorToJson(error) });
@@ -4581,8 +4667,17 @@ function installShutdownHandlers(): void {
   });
   process.on("unhandledRejection", (reason) => {
     if (isTransientDatabaseError(reason)) {
+      databaseTransientUncaughtSuppressedCount += 1;
+      databaseLastTransientUncaughtSuppressedAt = nowIso();
       void rotateSqlClientAfterTransientError("unhandled-rejection", reason).catch((error) => logger("warn", "database_client_rotate_rejection_failed", { error: errorToJson(error) }));
-      logger("error", "transient_database_unhandled_rejection_suppressed", { error: errorToJson(reason) });
+      if (dirtyDatabaseTaskIds.size > 0 || dirtyDatabaseQueueIds.size > 0) scheduleDatabaseFlushRetry(config.databaseFlushIntervalMs);
+      logger("error", "transient_database_unhandled_rejection_suppressed", {
+        suppressedCount: databaseTransientUncaughtSuppressedCount,
+        dirtyTaskCount: dirtyDatabaseTaskIds.size,
+        dirtyQueueCount: dirtyDatabaseQueueIds.size,
+        classification: classifyTransientDatabaseError(reason) as unknown as JsonValue,
+        error: errorToJson(reason),
+      });
       return;
     }
     logger("error", "unhandled_rejection", { error: errorToJson(reason) });
@@ -5965,7 +6060,7 @@ async function startDatabaseBackedRuntime(): Promise<void> {
   persistState();
   serviceReady = true;
   void refreshSchedulerTasksFromDatabase("startup").catch((error) => {
-    databaseLastError = databaseErrorMessage(error);
+    setDatabaseLastError(error);
     logger("warn", "scheduler_database_startup_refresh_failed", { error: errorToJson(error) });
   });
   startSchedulerDatabasePoller();
@@ -5979,6 +6074,6 @@ async function startDatabaseBackedRuntime(): Promise<void> {
 void initDatabasePersistenceWithRetry()
   .then(() => startDatabaseBackedRuntime())
   .catch((error) => {
-    databaseLastError = databaseErrorMessage(error);
+    setDatabaseLastError(error);
     logger("error", "database_persistence_init_failed", { error: errorToJson(error) });
   });

src/components/microservices/code-queue/src/queue-api.ts

@@ -24,7 +24,9 @@ export interface QueueApiContext {
   databaseLastError: () => string | null;
   databaseReady: () => boolean;
   defaultQueueId: string;
+  dirtyDatabaseQueueCount: () => number;
   dirtyDatabaseTaskCount: () => number;
+  databaseStorageHealth: () => JsonValue;
   jsonResponse: (body: unknown, status?: number) => Response;
   judgeFailRetryLimit: number;
   loadQueuesFromDatabase: () => Promise<QueueRecord[]>;
@@ -526,7 +528,9 @@ function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks()
       postgresConfigured: true,
       postgresReady: ctx().databaseReady(),
       dirtyTaskCount: ctx().dirtyDatabaseTaskCount(),
+      dirtyQueueCount: ctx().dirtyDatabaseQueueCount(),
       lastError: ctx().databaseLastError(),
+      health: ctx().databaseStorageHealth(),
       outputArchiveDir: ctx().config.outputArchiveDir,
       inMemoryOutputRecords: ctx().config.maxInMemoryOutputRecords,
       inMemoryEventRecords: ctx().config.maxInMemoryEventRecords,

src/components/microservices/code-queue/src/runner-error-classifier.ts

@@ -24,6 +24,7 @@ export interface RunnerErrorClassification {
   globalBlocker: boolean;
   singlePathObservation: boolean;
   externalProvider429: boolean;
+  staleBadResume: boolean;
   reason: string;
   evidence: string[];
   recommendedTriageCommand: string;
@@ -117,6 +118,12 @@ export function classifyRunnerError(message: string, providerId?: string | null)
     /exceeded retry limit/gu,
     /tokens per min|requests per min/gu,
   ]);
+  const staleBadResumeEvidence = collectEvidence(normalized, [
+    /no rollout found for thread id [a-z0-9._:-]+/gu,
+    /no rollout found for thread id/gu,
+    /rollout .*not found .*thread/gu,
+    /missing .*rollout .*thread/gu,
+  ]);
   const runnerLocalEvidence = collectEvidence(normalized, [
     /spawn .*enoent/gu,
     /command not found/gu,
@@ -133,8 +140,16 @@ export function classifyRunnerError(message: string, providerId?: string | null)
   let reason = "The runner failure needs provider triage before it can be treated as a global blocker.";
   let evidence: string[] = [];
   let externalProvider429 = false;
+  let staleBadResume = false;
 
-  if (externalProvider429Evidence.length > 0) {
+  if (staleBadResumeEvidence.length > 0) {
+    scope = "scheduler";
+    disposition = "service-degraded";
+    failureKind = "stale-bad-resume-thread-rollout-missing";
+    staleBadResume = true;
+    reason = "Codex app-server rejected resume because the saved thread id has no rollout; the scheduler should stop reusing that stale resume thread and reopen bounded recovery on a fresh thread.";
+    evidence = staleBadResumeEvidence;
+  } else if (externalProvider429Evidence.length > 0) {
     scope = "external-provider";
     disposition = "retryable-transient";
     failureKind = "external-provider-rate-limit";
@@ -187,6 +202,7 @@ export function classifyRunnerError(message: string, providerId?: string | null)
     globalBlocker: false,
     singlePathObservation: true,
     externalProvider429,
+    staleBadResume,
     reason,
     evidence,
     recommendedTriageCommand: providerTriageCommand(providerId, raw, scope),