From 97a647bc8d819cf25691764736d3b001ef1b066e Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 17 May 2026 11:12:41 +0000 Subject: [PATCH] fix(code-queue): harden active state recovery --- docs/reference/microservices.md | 4 +- .../microservices/code-queue-mgr/src/index.ts | 10 +- .../microservices/code-queue/src/index.ts | 120 +++++++++++++++++- .../microservices/code-queue/src/types.ts | 1 + 4 files changed, 125 insertions(+), 10 deletions(-) diff --git a/docs/reference/microservices.md b/docs/reference/microservices.md index 1c2eacac..0fb63560 100644 --- a/docs/reference/microservices.md +++ b/docs/reference/microservices.md @@ -161,7 +161,7 @@ Baidu Netdisk 在 UniDesk 语境中按纯后端服务管理：不得暴露百度 - 服务拆分语义：`code-queue-read` 只承载 GET/HEAD 查询、overview、任务详情、Trace/output/transcript、统计和只读健康，可多副本滚动更新；它必须设置 `CODE_QUEUE_SERVICE_ROLE=read` 与 `CODE_QUEUE_SCHEDULER_ENABLED=false`，且不得接受入队、queue 变更、已读、重试、移动、追加 prompt 或打断这类 mutation。`code-queue-write` 承载入队、queue 创建/合并/更新、已读、手动重试、移动等命令写入，初期保持单副本和 `CODE_QUEUE_SERVICE_ROLE=write`，只把命令和任务状态写入 PostgreSQL，不启动 agent 子进程。`code-queue-scheduler` 是唯一拥有 scheduler 和 active run 的执行服务，设置 `CODE_QUEUE_SERVICE_ROLE=scheduler` 与 `CODE_QUEUE_SCHEDULER_ENABLED=true`，负责从 PostgreSQL 热任务集轮询新写入任务、推进队列、启动 Codex/OpenCode、处理 running task 的 steer/interrupt、发送终态通知和暴露执行端 `/health`。普通 Service 负载均衡不得把 mutation 打到 read，也不得把 running task 控制打到 write。 - 实例语义：D601 是当前唯一 active 执行节点，`code-queue-scheduler` 以一个 scheduler Pod 承载长生命周期 Codex/OpenCode 子进程并轮询主 PostgreSQL 中由 `code-queue-mgr` 写入的 queued/retry_wait 任务。D518 不属于当前 Code Queue k3s 拓扑；在没有原生 k3s-agent 与稳定 Kubernetes 网络前，不得把 D518 写回 `expectedNodeIds` 或恢复 `code-queue-d518` standby。D601 scheduler 默认关闭 `CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED`；历史 OA Trace/STEP 回填必须通过显式 `/api/oa/backfill` 运维动作触发，不能在每次 Pod 重启时自动批量发布旧事件。 - 滚动更新边界：master `code-queue-mgr` 保证 D601 抖动或执行面滚动更新期间普通提交、queue 管理和历史读取仍可用；但当前 D601 scheduler Pod 内仍直接承载正在运行的 agent 子进程，scheduler Pod 被替换时 active task 仍会进入 restart-recovery/retry 语义，不能宣称 running task 零中断。真正的长期目标是继续把调度器和执行器拆开：scheduler 只负责 claim task 并创建 Kubernetes Job/Pod 或独立 worker，runner 把输出、状态、attempt、事件和通知写回 PostgreSQL/OA Event Flow/归档；只有这样 controller/scheduler 滚动更新才不会影响正在执行的任务。 -- Restart recovery：D601 scheduler 启动时必须把没有本地 active run 的 `running`/`judging` 任务恢复为 `retry_wait` 并先写回 PostgreSQL，再开启新一轮 scheduler 轮询；否则 PG 中残留的 running 会阻塞队列且不会被执行。health/overview 中的 `activeTaskIds` 只代表当前进程真实持有的 agent run；数据库里仍处于 `running`/`judging` 但没有本地 run 的任务只能作为 `orphanedActiveTaskIds` 暴露，不能计入 active run slot。 +- Restart recovery：D601 scheduler 启动时必须把没有本地 active run 的 `running`/`judging` 任务恢复为 `retry_wait` 并先写回 PostgreSQL，再开启新一轮 scheduler 轮询；同时必须清理 `queued`/`retry_wait`/terminal 任务残留的 `activeTurnId`，否则 PG 中残留的 running 或旧 turn id 会阻塞队列且不会被执行。health/overview 中的 `activeTaskIds` 只代表当前进程真实持有的 agent run；数据库里仍处于 `running`/`judging` 但没有本地 run 的任务只能作为 scheduler 侧 `orphanedActiveTaskIds` 暴露，不能计入 active run slot。主 server 直管 `code-queue-mgr` 只有 PostgreSQL 视角，不得把数据库中的 `running`/`judging` 误报为真实 active run；只能作为 `databaseActiveTaskIds`/`executionStateSource=postgres-control-plane` 这类控制面状态返回。 - 部署引用：Code Queue 镜像仍复用 `src/components/microservices/code-queue/Dockerfile`，Kubernetes 运行清单为 `src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml`，`config.json` 对外记录 k3s manifest `src/components/microservices/k3sctl-adapter/k3s/code-queue.k3s.json`；主 server 根目录 `docker-compose.yml` 不包含 `code-queue` service，旧 D601 direct Compose 文件只作为迁移/本地诊断参考，不是正式运行入口。 - 主服务依赖映射：Code Queue 仍以主 PostgreSQL 为权威数据库，但 D601 k3s Pod 不能依赖公网直连 `74.48.78.17:15432/4255`。Pod 内 `DATABASE_URL` 和 `OA_EVENT_FLOW_BASE_URL` 必须指向集群内 `d601-tcp-egress-gateway` Service，再由该 gateway 通过 D601 provider-gateway egress proxy 的 HTTP CONNECT 转发到主 PostgreSQL 和 OA Event Flow；新增 TCP 依赖时扩展 `TCP_EGRESS_ROUTES`，不得在业务容器里新增一次性公网直连或 ad hoc 隧道。D601 active 实例的 `CODE_QUEUE_NOTIFY_CLAUDEQQ_BASE_URL` 必须使用集群内 ClaudeQQ Service `http://claudeqq.unidesk.svc.cluster.local:3290`，并把 `claudeqq`/`claudeqq.unidesk.svc.cluster.local` 加入 `NO_PROXY`，避免任务完成通知被默认出网代理错误转发。旧 `http://host.docker.internal:3290` 只允许作为迁移期诊断，不得作为 Code Queue k3s Pod 的正式通知路径。这些端口映射只服务受控节点运行时，必须用防火墙或等价策略限制来源，不得成为浏览器或任意公网客户端入口。 - K8s 探针与启动维护：Kubernetes liveness/startup probe 必须使用轻量 `/live`，readiness 和用户服务健康使用 `/health`；`/health` 不得执行全量任务聚合、历史回填或长事务索引维护，历史任务总览应由 `/api/tasks/overview` 读取 PostgreSQL。启动时允许后台执行队列元数据 flush、通知 outbox 读取、任务表索引维护和 overview warmup，但这些维护不得阻塞 Bun server、readiness endpoint 或 frontend overview；通知表索引和大批量 OA backfill 不得作为默认启动副作用。 @@ -179,7 +179,7 @@ Baidu Netdisk 在 UniDesk 语境中按纯后端服务管理：不得暴露百度 - OpenCode 远程执行：`minimax-m2.7` 走 OpenCode JSON event port 时，本地和远程命令都必须显式执行 `opencode run ...`；远程 Docker exec 不得退化成 `exec run ...`，否则会在目标容器内变成 `bash: exec: run: not found`。OpenCode JSON stream 的终态判定以“当前进程退出码 + 当前 attempt 的最终 assistant response”为准：`exit=0` 且当前 attempt 产生非空最终回复时，即使上游没有发 `step_finish` 事件，也应视为正常 terminal；非零退出、无当前最终回复或传输关闭才进入 retry。每个 attempt 的 `finalResponse` 必须只来自当前 OpenCode/Codex turn，禁止在当前 turn 未产出最终回复时回退复用 task 上一次 `finalResponse`，否则会把旧任务内容误判为本轮完成。 - Codex 控制：服务内部启动 `codex app-server --listen stdio://`，用 JSON-RPC 调用 `thread/start`、`turn/start`、`turn/steer` 和 `turn/interrupt`，并监听 `turn/completed`、assistant delta、reasoning delta、command output delta、file diff delta 等通知生成前端可轮询的 transcript。 - 用户输入持久化：任务初始 prompt 以 `basePrompt/displayPrompt` 作为结构化来源，运行中追加的 `turn/steer` prompt 必须写入 `promptHistory`；transcript 构建时从这些结构化字段合成 `Submitted prompt` 和 `Steer prompt`，不能只依赖有 600 条上限的 raw output，否则长任务输出增长后会丢失关键人工指令。 -- 队列语义：`POST /api/tasks` 或 `/api/tasks/batch` 入队，服务始终只运行一个 Codex turn；当前任务真正终止后才推进下一个任务。`GET /api/tasks` 与 `GET /api/tasks/{id}` 返回队列、attempt、judge 和输出；`GET /api/tasks/{id}/summary` 返回按任务 ID 查询的结构化摘要，包括初始 prompt、最后 assistant message、工具调用摘要、attempt、judge、错误和耗时；CLI 入口是 `bun scripts/cli.ts codex task `。`GET|POST /api/tasks/{id}/judge?attempt=N` 与 CLI `bun scripts/cli.ts codex judge --attempt N` 用于单步复现指定 attempt 的 judge，必须复用真实队列 worker 的上下文构建、prompt 压缩、MiniMax 调用、JSON 去噪/repair 和 fallback 路径；`dryRun=1`/`--dry-run` 只输出 prompt/payload 和重建诊断，不调用 MiniMax。`POST /api/tasks/{id}/steer` 向运行中 turn 推入 prompt；`POST /api/tasks/{id}/interrupt` 或 `DELETE /api/tasks/{id}` 打断/取消；`POST /api/tasks/{id}/retry` 手动重试。队列 worker 必须隔离单个 task 的异常，不能因为某个 app-server、judge 异常或 judge 判定 `fail` 让后续 queued 任务停止；`fail` 只把当前任务标为 failed，随后必须继续扫描并推进下一个 queued/retry_wait 任务。当存在 queued/retry_wait 且 worker 空闲时，watchdog 必须自动重新调度。 +- 队列语义：`POST /api/tasks` 或 `/api/tasks/batch` 入队，服务始终只运行一个 Codex turn；当前任务真正终止后才推进下一个任务。`GET /api/tasks` 与 `GET /api/tasks/{id}` 返回队列、attempt、judge 和输出；`GET /api/tasks/{id}/summary` 返回按任务 ID 查询的结构化摘要，包括初始 prompt、最后 assistant message、工具调用摘要、attempt、judge、错误和耗时；CLI 入口是 `bun scripts/cli.ts codex task `。`GET|POST /api/tasks/{id}/judge?attempt=N` 与 CLI `bun scripts/cli.ts codex judge --attempt N` 用于单步复现指定 attempt 的 judge，必须复用真实队列 worker 的上下文构建、prompt 压缩、MiniMax 调用、JSON 去噪/repair 和 fallback 路径；`dryRun=1`/`--dry-run` 只输出 prompt/payload 和重建诊断，不调用 MiniMax。`POST /api/tasks/{id}/steer` 向运行中 turn 推入 prompt；`POST /api/tasks/{id}/interrupt` 或 `DELETE /api/tasks/{id}` 打断/取消；`POST /api/tasks/{id}/retry` 手动重试。队列 worker 必须隔离单个 task 的异常，不能因为某个 app-server、数据库 claim、judge 异常、judge 超时或 judge 判定 `fail` 让后续 queued 任务停止；`fail` 只把当前任务标为 failed，随后必须继续扫描并推进下一个 queued/retry_wait 任务。数据库 claim 必须有硬超时且失败时释放 active run slot；judge 必须有独立 watchdog，超时后走 fallback judge 并继续推进。当存在 queued/retry_wait 且 worker 空闲时，watchdog 必须自动重新调度。 - 稳定性与重启恢复：Code Queue 的第一目标是长期稳定可用；部署修复或运维排障时不得因为担心容器重启会打断任务而拒绝重启、重建或替换 active Pod。容器重启、服务进程重启和镜像替换后，队列、`promptHistory`、running/judging/retry_wait 任务和 active session 元数据必须从 PostgreSQL 恢复，并在已有 `codexThreadId` 可用时用 `thread/resume` 和 continuation prompt 无缝继续当前任务；如果原 app-server turn 已丢失，也必须把当前任务恢复到可 retry/continue 的状态，不能错误推进下一个任务或永久卡住。D601 侧重建必须走 `bun scripts/cli.ts codex deploy `；禁止先手工 `docker rm` 或只手工 `docker compose up` 再依赖后续命令补救，因为中断窗口会让 Pod/容器消失并触发 frontend/core 用户服务代理失败。重启后出现 active task 丢失、手动 steer/interrupt 记录丢失、running 任务卡死、误判完成、跳过当前任务、容器消失或阻塞队列，均属于 Code Queue 的 P0 核心缺陷，必须先修复并补充 restart-recovery 验收，不能把“避免重启”作为交付策略。 - 调度与 active run slot：Code Queue 必须把“queue processor 正在等待/退避/轮询”和“实际占用 Codex/OpenCode 子进程运行槽”分开建模；`CODE_QUEUE_MAX_ACTIVE_QUEUES` 只限制真实 active run slot，不能把 retry backoff、等待内存下降或等待前序任务的 `processingQueues` 计入 active slot，否则设置全局 active slot 上限时，一个空等队列会把其他 runnable queue 永久饿死。多个 queue 同时等待 active slot 时必须显式维护 FIFO waiter 队列，避免某个长 retry/backoff 队列刚释放 slot 就立刻重抢，导致更早进入等待的 `retry_wait` 任务长期饥饿；`/health` 必须同时暴露真实 `activeQueueIds`、`activeRunSlotCount`、等待中的 `processingQueueIds` 和 active slot waiters，排障时以 active run slot 与 waiter 顺序判断是否真的有任务在跑、谁应下一个启动。restart-recovery 后的 `retry_wait` 任务若缺失 `codexThreadId`/OpenCode session id，不得无限拒绝 retry；必须用紧凑 recovery prompt 和原始任务摘要重新开一个 agent thread/session，让任务继续推进并在 Trace 中留下 recovery 证据。任何修改 scheduler、retry backoff、queue move、manual retry、shutdown recovery 或内存等待逻辑时，都必须保留“空等 processor 不占 active run slot”、“等待者 FIFO 不饥饿”和“缺失 thread/session 可恢复”的自测或 live 验证。 - 内存优化过程与防回归：Code Queue 已迁移到 D601，但内存治理仍必须按“PostgreSQL 权威源优先、进程热状态最小化、容器硬上限兜底”的顺序设计。长期可复用的优化路径是：先确认任务、queue、readAt、promptHistory、active session 和通知 outbox 均可从 PostgreSQL 恢复；再把历史任务列表、详情、统计、Trace/output 和 `/health` 的只读查询改为 PostgreSQL 直读或聚合查询；随后只把 `queued`、`running`、`judging`、`retry_wait` 等调度必需任务载入 Bun 堆，并在 PostgreSQL 查询侧裁剪 hot `output`/`events`；最后用 dirty-only flush、append-only 输出归档、Codex SQLite 小批量导出、`bun --smol`、`mem_limit=600m`、`memswap_limit=1536m`、`NODE_OPTIONS=--max-old-space-size=768` 和 cgroup memory watchdog 作为运行时防线。PostgreSQL 到进程的单次读取足够快，不能为了减少 SQL 查询把全部历史 `task_json`、Trace、output 或统计摘要常驻内存；任何新增缓存都必须有默认较小的环境变量上限、明确淘汰策略、可从 PostgreSQL 或 append-only 归档重建，且不得影响重启恢复。新增或修改 `/api/tasks`、overview、stats、summary、transcript、output、trace、health、flush、scheduler 和通知路径时，禁止在常规请求中调用会物化全量历史任务 JSON 的代码，禁止启动后无条件重写全量历史 task JSON，禁止用未设上限的 `Map`/数组保存历史 output/event/Trace，`CODE_QUEUE_MAX_ACTIVE_QUEUES=0` 表示不按 queue 数量设置全局排队上限；如显式设置为正数，必须同时说明内存预算并补充内存压测验收。memory watchdog 必须以 cgroup working set 为主要判断，且在 swap 仍有余量时不得提前杀掉唯一 active run；否则 TypeScript/Playwright 这类短时高内存验证会被错误中断并让 retry 队列反复震荡。 diff --git a/src/components/microservices/code-queue-mgr/src/index.ts b/src/components/microservices/code-queue-mgr/src/index.ts index 6b8a6a40..659d9c56 100644 --- a/src/components/microservices/code-queue-mgr/src/index.ts +++ b/src/components/microservices/code-queue-mgr/src/index.ts @@ -698,6 +698,7 @@ function normalizeTask(value: unknown): QueueTask { nextPrompt: typeof record.nextPrompt === "string" ? record.nextPrompt : null, nextMode: record.nextMode === "initial" || record.nextMode === "retry" ? record.nextMode : null, }; + if (task.status !== "running" && task.status !== "judging") task.activeTurnId = null; task.stepCount = numberField(record.stepCount ?? record.llmStepCount, task.output.length); task.llmStepCount = task.stepCount; task.outputMaxSeq = Math.max(outputMaxSeq(task), numberField(record.outputMaxSeq, 0)); @@ -1254,7 +1255,7 @@ async function queueSummary(tasks?: QueueTask[], queueRecords?: QueueRecord[]): memo[task.status] = (memo[task.status] ?? 0) + 1; return memo; }, {}); - const orphanedActiveTaskIds = taskRows.filter((task) => task.status === "running" || task.status === "judging").map((task) => task.id).sort(); + const databaseActiveTaskIds = taskRows.filter((task) => task.status === "running" || task.status === "judging").map((task) => task.id).sort(); const queueList = Array.from(summaries.values()).sort((left, right) => String(left.id).localeCompare(String(right.id))); return { total: taskRows.length, @@ -1267,8 +1268,11 @@ async function queueSummary(tasks?: QueueTask[], queueRecords?: QueueRecord[]): activeRunSlotWaiters: [], activeTaskIds: [], activeTaskId: null, - orphanedActiveTaskIds, - orphanedActiveTaskCount: orphanedActiveTaskIds.length, + databaseActiveTaskIds, + databaseActiveTaskCount: databaseActiveTaskIds.length, + executionStateSource: "postgres-control-plane", + orphanedActiveTaskIds: [], + orphanedActiveTaskCount: 0, processing: false, counts, unreadTerminal: taskRows.reduce((total, task) => total + (terminalTaskUnread(task) ? 1 : 0), 0), diff --git a/src/components/microservices/code-queue/src/index.ts b/src/components/microservices/code-queue/src/index.ts index bf62b556..088816c6 100644 --- a/src/components/microservices/code-queue/src/index.ts +++ b/src/components/microservices/code-queue/src/index.ts @@ -10,6 +10,7 @@ import type { CgroupMemoryUsage, CodexEventSummary, CodexRunResult, + JudgeFailureDetails, JudgeProbeCase, JudgeResult, JsonValue, @@ -388,6 +389,7 @@ function readConfig(): RuntimeConfig { databaseUrl: envRequiredString("DATABASE_URL"), databasePoolMax: Math.max(1, Math.min(8, envNumber("CODE_QUEUE_DATABASE_POOL_MAX", envNumber("DATABASE_POOL_MAX", 2)))), databaseFlushIntervalMs: Math.max(100, Math.min(10_000, envNumber("CODE_QUEUE_DATABASE_FLUSH_INTERVAL_MS", 1000))), + databaseClaimTimeoutMs: Math.max(1000, Math.min(60_000, envNumber("CODE_QUEUE_DATABASE_CLAIM_TIMEOUT_MS", 15_000))), oaEventFlowBaseUrl: envString("OA_EVENT_FLOW_BASE_URL", "http://oa-event-flow:4255").replace(/\/+$/u, ""), notifyClaudeQqEnabled: envBool("CODE_QUEUE_NOTIFY_CLAUDEQQ_ENABLED", false), notifyClaudeQqBaseUrl: envString("CODE_QUEUE_NOTIFY_CLAUDEQQ_BASE_URL", "http://claudeqq.unidesk.svc.cluster.local:3290").replace(/\/+$/u, ""), @@ -928,6 +930,10 @@ function normalizeTaskMetric(value: unknown): number | null { return Number.isFinite(parsed) && parsed >= 0 ? Math.floor(parsed) : null; } +function taskCanHaveActiveTurn(status: TaskStatus): boolean { + return status === "running" || status === "judging"; +} + function normalizeTask(task: QueueTask): QueueTask { task.queueId = safeQueueId(task.queueId); task.queueEnteredAt = taskQueueEnteredAt(task); @@ -940,6 +946,7 @@ function normalizeTask(task: QueueTask): QueueTask { task.readAt = null; } task.activeTurnId ??= null; + if (!taskCanHaveActiveTurn(task.status)) task.activeTurnId = null; task.providerId = normalizeTaskProviderId(task.providerId); task.model ||= config.defaultModel; task.executionMode = normalizeCodeExecutionMode(task.executionMode); @@ -1525,7 +1532,15 @@ async function deleteTaskFromDatabase(taskId: string): Promise { async function claimTaskInDatabase(task: QueueTask, expectedQueueId: string): Promise { if (!databaseReady) return true; - const claimed = await sql.begin(async (client) => await upsertTaskToDatabase(client, task, { claimQueueId: expectedQueueId })); + const claimed = await withTimeout( + sql.begin(async (client) => { + const timeout = `${config.databaseClaimTimeoutMs}ms`; + await client`SELECT set_config('statement_timeout', ${timeout}, true), set_config('lock_timeout', ${timeout}, true)`; + return await upsertTaskToDatabase(client, task, { claimQueueId: expectedQueueId }); + }), + config.databaseClaimTimeoutMs, + `database claim timed out after ${config.databaseClaimTimeoutMs}ms`, + ); if (claimed) return true; const databaseTask = await loadTaskFromDatabase(task.id); if (databaseTask !== null) reconcileHotTaskFromDatabase(databaseTask); @@ -2068,6 +2083,14 @@ async function initDatabasePersistence(): Promise { AND COALESCE(task_json->>'readAt', '') <> '' AND (task_json->>'readAt') ~ '^\\d{4}-\\d{2}-\\d{2}T' `; + await sql` + UPDATE unidesk_code_queue_tasks + SET + active_turn_id = NULL, + task_json = jsonb_set(task_json, '{activeTurnId}', 'null'::jsonb, true) + WHERE status NOT IN ('running', 'judging') + AND active_turn_id IS NOT NULL + `; await sql`ALTER TABLE unidesk_code_queue_queues ADD COLUMN IF NOT EXISTS name TEXT NOT NULL DEFAULT ''`; await sql`ALTER TABLE unidesk_code_queue_workdirs ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`; } @@ -3183,6 +3206,63 @@ function retryBackoffMs(completedAttempts: number): number { return Math.min(retryBackoffMaxMs, retryBackoffBaseMs * (2 ** exponent)); } +async function withTimeout(promise: Promise, timeoutMs: number, message: string): Promise { + let timer: ReturnType | null = null; + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(message)), timeoutMs); + }); + try { + return await Promise.race([promise, timeout]); + } finally { + if (timer !== null) clearTimeout(timer); + } +} + +function fallbackJudgeFromWatchdog(task: QueueTask, result: CodexRunResult, error: unknown, startedAt: number): JudgeResult { + const durationMs = Date.now() - startedAt; + const message = error instanceof Error ? error.message : String(error); + const detail: JudgeFailureDetails = { + provider: "minimax", + stage: "unknown", + model: config.minimaxModel, + apiBase: config.minimaxApiBase, + occurredAt: nowIso(), + durationMs, + timeoutMs: config.judgeTimeoutMs, + timedOut: durationMs >= config.judgeTimeoutMs, + errorName: error instanceof Error ? error.name : "Error", + errorMessage: message, + }; + logger("warn", "judge_watchdog_fallback", { + taskId: task.id, + attempt: task.currentAttempt, + durationMs, + timeoutMs: config.judgeTimeoutMs, + error: safePreview(message, 1000), + }); + return { + decision: result.terminalStatus === "completed" ? "complete" : "retry", + confidence: 0.3, + reason: `MiniMax judge watchdog fallback: ${message}`, + source: "fallback", + failureDetails: detail, + raw: { minimaxFailure: detail as unknown as JsonValue }, + }; +} + +async function judgeTaskWithWatchdog(task: QueueTask, result: CodexRunResult): Promise { + const startedAt = Date.now(); + try { + return await withTimeout( + judgeTask(task, result), + config.judgeTimeoutMs + 5_000, + `judge task timed out after ${config.judgeTimeoutMs + 5_000}ms`, + ); + } catch (error) { + return fallbackJudgeFromWatchdog(task, result, error, startedAt); + } +} + async function sleepForRetryBackoff(task: QueueTask, delayMs: number): Promise { let remaining = delayMs; while (remaining > 0 && !task.cancelRequested && !shutdownRequested) { @@ -3198,6 +3278,18 @@ function taskHasLocalExecutionRecoveryClaim(task: QueueTask): boolean { return activeRunSlotReservations.has(queueIdOf(task)); } +function clearInactiveActiveTurnIds(): number { + let changed = 0; + for (const task of state.tasks) { + if (taskCanHaveActiveTurn(task.status) || task.activeTurnId === null) continue; + task.activeTurnId = null; + task.updatedAt = nowIso(); + markTaskDirty(task.id); + changed += 1; + } + return changed; +} + function queueActiveTasksForRestartRetry(reason: string, method: string, options: { onlyActiveRuns?: boolean } = {}): number { if (!config.schedulerEnabled || !serviceRoleAllowsScheduler(config.serviceRole)) return 0; let recovered = 0; @@ -3287,7 +3379,24 @@ async function runTask(task: QueueTask): Promise { task.readAt = null; task.finishedAt = null; task.updatedAt = startedAt; - if (!await claimTaskInDatabase(task, claimQueueId)) { + let claimed = false; + try { + claimed = await claimTaskInDatabase(task, claimQueueId); + } catch (error) { + releaseRunSlot(); + task.status = "retry_wait"; + task.activeTurnId = null; + task.finishedAt = null; + task.readAt = null; + task.updatedAt = nowIso(); + task.lastError = `Database claim failed before agent start: ${error instanceof Error ? error.message : String(error)}`; + appendOutput(task, "error", `${task.lastError}\n`, "database/claim"); + persistTaskState(task); + logger("error", "task_claim_failed", { taskId: task.id, queueId: claimQueueId, error: errorToJson(error) }); + await sleepForRetryBackoff(task, retryBackoffMs(Math.max(1, task.attempts.length))); + return; + } + if (!claimed) { releaseRunSlot(); return; } @@ -3310,7 +3419,7 @@ async function runTask(task: QueueTask): Promise { persistTaskState(task); if (task.cancelRequested) break; - const judge = await judgeTask(task, result); + const judge = await judgeTaskWithWatchdog(task, result); task.lastJudge = judge; const judgeOutput = appendOutput(task, judge.decision === "complete" ? "system" : judge.decision === "fail" ? "error" : "system", `judge=${judge.decision} confidence=${judge.confidence.toFixed(2)} source=${judge.source}: ${judge.reason}${judgeFailureDetailsForOutput(judge)}\n`, "judge"); const latestAttempt = task.attempts.at(-1); @@ -4938,12 +5047,13 @@ async function startDatabaseBackedRuntime(): Promise { logger("info", "service_started", { port: config.port, instanceId: config.instanceId, role: config.serviceRole, schedulerEnabled: config.schedulerEnabled, schedulerPollIntervalMs: config.schedulerPollIntervalMs, workdir: config.defaultWorkdir, defaultModel: config.defaultModel, judgeConfigured: config.minimaxApiKey.length > 0, storage: "postgres" }); const devReady = collectDevReady() as Record; logger(devReady.ok === true ? "info" : "warn", "dev_ready_check", devReady); + const inactiveActiveTurnsCleared = clearInactiveActiveTurnIds(); const startupRecovered = config.schedulerEnabled ? queueActiveTasksForRestartRetry("Service restarted while task was active", "startup") : 0; const startupRecoveryDirtyTaskCount = dirtyDatabaseTaskIds.size; if (startupRecovered > 0 || startupRecoveryDirtyTaskCount > 0) { - logger("warn", "startup_requeued_active_tasks", { recovered: startupRecovered, dirtyTaskCount: startupRecoveryDirtyTaskCount }); + logger("warn", "startup_requeued_active_tasks", { recovered: startupRecovered, inactiveActiveTurnsCleared, dirtyTaskCount: startupRecoveryDirtyTaskCount }); await flushDirtyTasksToDatabase(true); - logger("info", "startup_requeued_active_tasks_flushed", { recovered: startupRecovered, dirtyTaskCount: startupRecoveryDirtyTaskCount }); + logger("info", "startup_requeued_active_tasks_flushed", { recovered: startupRecovered, inactiveActiveTurnsCleared, dirtyTaskCount: startupRecoveryDirtyTaskCount }); } persistState(); serviceReady = true; diff --git a/src/components/microservices/code-queue/src/types.ts b/src/components/microservices/code-queue/src/types.ts index d2ed5653..396ddb86 100644 --- a/src/components/microservices/code-queue/src/types.ts +++ b/src/components/microservices/code-queue/src/types.ts @@ -75,6 +75,7 @@ export interface RuntimeConfig { databaseUrl: string; databasePoolMax: number; databaseFlushIntervalMs: number; + databaseClaimTimeoutMs: number; oaEventFlowBaseUrl: string; notifyClaudeQqEnabled: boolean; notifyClaudeQqBaseUrl: string;