Merge pull request #148 from pikasTech/fix/code-queue-pg-rotation-crash

fix(code-queue): harden postgres rotation dirty flush

scripts/src/code-queue.ts

@@ -3864,12 +3864,54 @@ function blockerLikeFinalResponseSignals(task: Record<string, unknown>, summary:
     if (pattern.test(text) && !signals.includes(id)) signals.push(id);
   };
   add("blocked", /\b(blocked|blocker|cannot proceed|can't proceed|stuck|waiting for commander|needs authorization|need authorization|requires approval|permission denied)\b|阻塞|卡住|等待授权|需要授权|权限不足/iu);
-  add("infra-auth-network", /\b(auth|token|credential|github transient|dns|rate limit|429|timeout|timed out|unreachable|offline|proxy|tunnel|provider unavailable)\b|鉴权|令牌|网络|超时|不可达/iu);
+  add("infra-auth-network", /\b(auth|token|credential|github transient|dns|rate limit|429|timeout|timed out|unreachable|offline|proxy|tunnel|provider unavailable|postgres|database|connection_closed|socket\.write|null is not an object)\b|鉴权|令牌|网络|超时|不可达|数据库/iu);
   add("merge-or-test-failure", /\b(conflict|merge failed|tests? failed|typecheck failed|syntax failed|build failed|ci failed|e2e failed)\b|冲突|测试失败|构建失败|检查失败/iu);
   add("not-deployed", /\b(not deployed|not rebuilt|not rolled out|deploy skipped|rollout skipped|needs rollout|requires deploy)\b|未部署|未重建|未上线|需要部署/iu);
   return signals;
 }
 
+function commanderInfrastructureSignals(rawQueue: Record<string, unknown>): Record<string, unknown> {
+  const storage = asRecord(rawQueue.storage) ?? {};
+  const health = asRecord(storage.health) ?? {};
+  const rawSignals = asArray(health.signals).map((item) => asRecord(item)).filter((item): item is Record<string, unknown> => item !== null);
+  const boundedSignals = rawSignals.slice(0, 3).map((signal) => ({
+    id: asString(signal.id) || "code-queue-infrastructure-blocker",
+    category: asString(signal.category) || "infrastructure-blocker",
+    severity: asString(signal.severity) || "high",
+    retryable: asBoolean(signal.retryable),
+    bounded: signal.bounded !== false,
+    message: boundedInlineString(asString(signal.message), 240).text,
+    evidence: stringList(signal.evidence).slice(0, diagnosticsReasonPreviewLimit),
+    commanderAction: boundedInlineString(asString(signal.commanderAction), 260).text,
+    source: asString(signal.source) || "code-queue",
+  }));
+  const storageDegraded = health.degraded === true || storage.postgresReady === false || storage.lastError !== null && storage.lastError !== undefined;
+  return {
+    infrastructureBlocker: storageDegraded || boundedSignals.some((signal) => signal.category === "infrastructure-blocker"),
+    status: health.status ?? (storageDegraded ? "degraded" : "ready"),
+    source: "queue.storage.health",
+    signalCount: rawSignals.length,
+    omittedSignalCount: Math.max(0, rawSignals.length - boundedSignals.length),
+    bounded: true,
+    signals: boundedSignals,
+    storage: {
+      postgresReady: storage.postgresReady ?? health.postgresReady ?? null,
+      dirtyTaskCount: storage.dirtyTaskCount ?? health.dirtyTaskCount ?? null,
+      dirtyQueueCount: storage.dirtyQueueCount ?? health.dirtyQueueCount ?? null,
+      consecutiveFlushFailures: health.consecutiveFlushFailures ?? null,
+      lastFlushFailureAt: health.lastFlushFailureAt ?? null,
+      nextFlushRetryAt: health.nextFlushRetryAt ?? null,
+      clientRotationCount: health.clientRotationCount ?? null,
+      lastClientRotationAt: health.lastClientRotationAt ?? null,
+      lastErrorKind: health.lastErrorKind ?? null,
+      lastErrorTransient: health.lastErrorTransient ?? null,
+    },
+    actionable: storageDegraded
+      ? "Treat as Code Queue infrastructure-blocker; inspect storage health/logs and wait for bounded dirty-flush retry before duplicating or canceling business tasks."
+      : "No Code Queue storage infrastructure blocker reported in this overview page.",
+  };
+}
+
 function commanderAttentionReasons(
   task: Record<string, unknown>,
   summary: Record<string, unknown> | null,
@@ -4126,6 +4168,7 @@ function codexTasksCommanderResult(
   const rawQueue = asRecord(taskPage.queue) ?? {};
   const rawDiagnostics = asRecord(rawQueue.executionDiagnostics) ?? {};
   const diagnostics = supervisorExecutionDiagnostics(rawDiagnostics);
+  const infrastructure = commanderInfrastructureSignals(rawQueue);
   const activity = compactCodeQueueActivity(rawQueue, diagnostics, { snapshotRole: "supervisor-poll" });
   const commanderConcurrency = asRecord(activity.commanderConcurrency) ?? {};
   const runningTasks = sortRunningWatchTasks(allTasks);
@@ -4222,6 +4265,7 @@ function codexTasksCommanderResult(
       riskCounts: {
         attention: attentionCounts(attentionItems, returnedAttention),
         activeRisks: activeRiskTasks.length,
+        infrastructureBlocker: infrastructure.infrastructureBlocker === true ? 1 : 0,
         heartbeatRiskTaskIds: stringList(rawDiagnostics.heartbeatRiskTaskIds).length,
         staleRecoveryCandidateTaskIds: stringList(rawDiagnostics.staleRecoveryCandidateTaskIds).length,
         traceGapTaskIds: stringList(rawDiagnostics.traceGapTaskIds).length,
@@ -4230,6 +4274,7 @@ function codexTasksCommanderResult(
       },
       highPriorityIssues: commanderHighPriorityIssues(allTasks, summaries),
       classification: commanderClassificationCounts(allTasks, summaries),
+      infrastructure,
       executionDiagnostics: diagnostics,
       degraded,
       commands: {