From d90c9d10e375b8fe1562b50f3ef63ab03ae10a84 Mon Sep 17 00:00:00 2001 From: Lyon <88232613+pikasTech@users.noreply.github.com> Date: Sat, 23 May 2026 21:33:49 +0800 Subject: [PATCH] fix: foreground commander queue concurrency Foreground codex queues commander concurrency summary and fixtures. --- docs/reference/code-queue-supervision.md | 4 +- .../code-queue-queues-shape-contract-test.ts | 293 ++++++++++++++++++ scripts/src/check.ts | 4 + scripts/src/code-queue.ts | 257 ++++++++++++++- scripts/src/help.ts | 9 +- 5 files changed, 558 insertions(+), 9 deletions(-) diff --git a/docs/reference/code-queue-supervision.md b/docs/reference/code-queue-supervision.md index 2c12ac19..67b8383f 100644 --- a/docs/reference/code-queue-supervision.md +++ b/docs/reference/code-queue-supervision.md @@ -275,7 +275,7 @@ replacement runner 只用于方向明显错误、质量不可接受、原 task - `bun scripts/cli.ts codex tasks --view commander --limit N`：host commander 轮询的推荐入口。输出是有界 action map，必须直接显示 `activeRunners.count`、计数来源、split-brain/heartbeat 处置、queued/retry_wait 精确计数、terminal-unread 总数和已省略行数、active 风险数、stale/heartbeat/trace gap、`finalResponse` 已出现但仍非终态的 awaiting terminal/judge、blocker-like final response、HWLAB#7/#99/#116/#164/#317 与 UniDesk#20/#118 命中、任务分类和下一步 drill-down 命令。默认不得输出完整 prompt、完整 final response、raw output、完整 trace 或 raw overview；需要详情只能按 task id 使用 `codex task`、`codex task --trace`、`codex output`、`codex read` 或 `rawOverview` 命令渐进展开。 - `bun scripts/cli.ts codex tasks --view supervisor --limit N`：查看默认低噪声监督视图，包括 `activeRunning`、running、完成未读、少量最近完成、queued/runnable、activity、commanderConcurrency、execution diagnostics、任务分类和下一步 drill-down 命令。默认行只保留 task id、队列、短 prompt/body 预览和原始字符数；`--limit` 是扫描/分页预算，不是返回几十条肥行的开关，CLI effective limit 安全上限为 100，输出必须用 `filters.requestedLimit`、`filters.effectiveLimit`、`filters.limitCapped`、`source.requestedLimit` 和 `source.effectiveLimit` 区分用户请求、CLI cap 和 overview 源拉取预算；例如 `--limit 260` 应明确显示 requested=260、effective=100、source=200，`running.returned` 只是低噪声返回行数。`show/detail/trace/output/full/read` 放在 section template 中，避免每条任务重复刷屏，需要更多内容再按 taskId 展开。刚执行 `codex submit` 后也可以先读 submit 返回的 `submitted.taskStates[]`、`queue.countContext`、`queue.activity.effectiveActiveTaskCount` 和 `queue.stateDisclosure`；若某个 id preview 有 `idsUnavailable=true`，不要把它当成空队列，按 `queue.listPreviewPolicy.rawCommand` 或本 supervisor 命令继续查。 -- `bun scripts/cli.ts codex queues`：查看低噪声队列计数、activity、commanderConcurrency、active task id、完成未读队列、runnable 队列和控制面诊断；需要完整队列行视图时加 `--full`，但 `--full` 仍默认分页，继续用 `--limit N`、`--page N` 或 `--offset N` 渐进展开。summary 和 full 都使用稳定 JSON path `.data.queues.items[]` 读取队列行，并从 `.data.queues.commanderConcurrency`、`.data.queues.activity`、`.data.queues.counts` 与 `.data.queues.executionDiagnostics` 读取全局活跃计数和执行诊断；完整 upstream 只通过输出中的 raw command 显式获取。 +- `bun scripts/cli.ts codex queues`：默认是 commander-first 队列态势摘要，`--commander` 是显式同义开关。输出前部固定使用 `.data.queues.commander`，先给出 `activeRunnerCount`、`source`、`target=15`、`slotDeficit`、`queuedCount`、`runningTasks`、`heartbeat.fresh`、`heartbeat.risk`、`heartbeat.staleRecoveryCandidates`、active/runnable queue 小页和 drill-down 命令；历史 queue item 列表保留在 `.data.queues.items[]`，但只是分页的次要行。需要完整队列行视图时加 `--full`，但 `--full` 仍默认分页，继续用 `--limit N`、`--page N` 或 `--offset N` 渐进展开。summary 和 full 都使用稳定 JSON path `.data.queues.items[]` 读取队列行，并从 `.data.queues.commander`、`.data.queues.commanderConcurrency`、`.data.queues.activity`、`.data.queues.counts` 与 `.data.queues.executionDiagnostics` 读取全局活跃计数和执行诊断；完整 upstream 只通过输出中的 raw command 显式获取。若 `/api/queues` 没有返回 task row，`runningTasks.items[].name` 会是 `null` 且 `nameSource=not-returned-by-api-queues`，此时按返回的 `codex task ` 或 supervisor 命令展开，不要假设任务没有名称。 - `bun scripts/cli.ts codex unread --limit N`：查看完成未读审阅积压的默认 triage，按 repo、issue、status 和 queue 汇总，并给出有界最新任务和 drill-down/read 命令；默认不输出 raw prompt、final response、trace 或 output。 - `bun scripts/cli.ts codex unread mark-read --repo owner/name --issue N --limit N --confirm`：批量已读入口，必须显式 `mark-read` 和 `--confirm`，否则结构化失败且不 POST `/read`。 - `bun scripts/cli.ts codex tasks --unread --limit N`：兼容查看完成未读审阅积压；`--unread` 与 `--unread-only` 等价，不能被静默忽略。 @@ -303,7 +303,7 @@ commander 视图的任务分类必须是确定性字段，至少区分 `business stale-active 恢复和 `/api/scheduler/reconcile?staleMs=...` 诊断入口的 heartbeat stale 阈值必须按安全下限归一化：缺省和低于默认 5 分钟的值都按 5 分钟处理，过大值按 24 小时上限截断，并在结构化响应中返回 `requestedStaleMs*`、`staleMsAdjusted`、`staleMsAdjustmentReason`、`minStaleMs` 和 `maxStaleMs`。任何 `staleMs=0` 或过低阈值都不能把仍有 fresh scheduler heartbeat 的任务判成 stale/recoverable。 -`codex queues`、`codex tasks --view commander` 和默认 supervisor 视图的 `activity` / `commanderConcurrency` 是指挥官并发治理的主读数。并发决策固定使用 `commanderConcurrency.activeRunnerCount` 或 commander `activeRunners.count`，它等于 `activity.effectiveActiveTaskCount`；15 并发策略的可补窗口按 `15 - activeRunnerCount` 计算，不能用 `activeQueueIds.length` 或 scheduler-local slot 数替代。`effectiveActiveTaskCount` 表示用于调度判断的有效活跃任务数；`databaseRunningTaskCount` 来自 PostgreSQL 中 `running` 状态计数；`databaseActiveTaskCount` 覆盖 running/judging 等数据库活跃任务；`heartbeatFreshActiveTaskCount` 表示 heartbeat-fresh 的有效 runner 数；`schedulerLocalActiveQueueCount` 和 `schedulerLocalActiveRunSlotCount` 只表示当前控制面本地可见 active run slots。`activeQueueIds` 与 `activeQueueCount` 是 scheduler-local 字段，可能在 `counts.running>0` 且 heartbeat 新鲜时为 0；看到这种组合时应按 `activity.effectiveActiveTaskCount`、`activity.heartbeatFreshActiveTaskCount` 和 `splitBrainLive` 决策，不得把空 `activeQueueIds` 当作零并发或停摆证据。`commanderConcurrency.splitBrainDisposition=live-count-as-active` 表示 split-brain 仍是 live 且应计入 active runner；`attentionRequired=true` 表示需要人工看一眼或重新 poll，`interventionRequired=true` 才表示当前输出已经足以进入高风险介入路径。单次 heartbeat risk、stale recovery candidates 或 `recommendedAction=investigate-heartbeat-risk` 应先落到 `attentionRequired=true` 加 `re-poll supervisor before recovery`，不得直接等价为恢复授权。 +`codex queues`、`codex tasks --view commander` 和默认 supervisor 视图的 `activity` / `commanderConcurrency` 是指挥官并发治理的主读数。并发决策固定使用 `commanderConcurrency.activeRunnerCount`、`.data.queues.commander.activeRunnerCount` 或 commander `activeRunners.count`，它等于 `activity.effectiveActiveTaskCount`；15 并发策略的可补窗口按 `15 - activeRunnerCount` 计算，CLI 也会直接给出 `.data.queues.commander.slotDeficit`，不能用 `activeQueueIds.length` 或 scheduler-local slot 数替代。`effectiveActiveTaskCount` 表示用于调度判断的有效活跃任务数；`databaseRunningTaskCount` 来自 PostgreSQL 中 `running` 状态计数；`databaseActiveTaskCount` 覆盖 running/judging 等数据库活跃任务；`heartbeatFreshActiveTaskCount` 表示 heartbeat-fresh 的有效 runner 数；`schedulerLocalActiveQueueCount` 和 `schedulerLocalActiveRunSlotCount` 只表示当前控制面本地可见 active run slots。`activeQueueIds` 与 `activeQueueCount` 是 scheduler-local 字段，可能在 `counts.running>0` 且 heartbeat 新鲜时为 0；看到这种组合时应按 `activity.effectiveActiveTaskCount`、`activity.heartbeatFreshActiveTaskCount`、`.data.queues.commander.runningTasks` 和 `splitBrainLive` 决策，不得把空 `activeQueueIds` 当作零并发或停摆证据。`commanderConcurrency.splitBrainDisposition=live-count-as-active` 表示 split-brain 仍是 live 且应计入 active runner；`attentionRequired=true` 表示需要人工看一眼或重新 poll，`interventionRequired=true` 才表示当前输出已经足以进入高风险介入路径。单次 heartbeat risk、stale recovery candidates 或 `recommendedAction=investigate-heartbeat-risk` 应先落到 `attentionRequired=true` 加 `re-poll supervisor before recovery`，不得直接等价为恢复授权。 单次 `provider is not online`、SSH 超时、proxy 超时或 registry 请求失败只能证明“当前观察路径失败”，不能单独升级为 D601 全局离线、CI/CD 全局阻塞或业务任务不可推进。指挥官和 runner 必须用多信号裁决运行面状态，至少区分以下观察面： diff --git a/scripts/code-queue-queues-shape-contract-test.ts b/scripts/code-queue-queues-shape-contract-test.ts index 484eecd0..cf2ee188 100644 --- a/scripts/code-queue-queues-shape-contract-test.ts +++ b/scripts/code-queue-queues-shape-contract-test.ts @@ -140,6 +140,224 @@ function splitBrainLiveResponse(): JsonRecord { }; } +function heartbeatRiskResponse(): JsonRecord { + const staleTaskIds = manyIds("task-stale", 4); + return { + ok: true, + status: 200, + body: { + ok: true, + queue: { + total: 6, + queueCount: 3, + activeQueueIds: [], + activeTaskIds: [], + queuedTaskIds: ["task-queued-risk"], + counts: { running: 4, queued: 1, retry_wait: 1 }, + unreadTerminal: 0, + executionDiagnostics: { + state: "stale-active", + effectiveLiveness: "at-risk", + recommendedAction: "investigate-heartbeat-risk", + now: "2026-05-20T00:30:00.000Z", + databaseActiveTaskCount: 4, + databaseActiveTaskIds: staleTaskIds, + schedulerActiveRunSlotCount: 0, + schedulerActiveTaskIds: [], + activeHeartbeatCount: 4, + activeHeartbeatTaskIds: staleTaskIds, + heartbeatFreshTaskIds: [], + heartbeatExpiredTaskIds: staleTaskIds, + heartbeatMissingTaskIds: [], + staleRecoveryCandidateTaskIds: staleTaskIds, + heartbeatRiskTaskIds: staleTaskIds, + lastSchedulerHeartbeatAt: "2026-05-20T00:15:00.000Z", + lastObservedAgentEventAt: "2026-05-20T00:14:30.000Z", + reasons: ["heartbeat expired for database-active tasks"], + }, + }, + queues: [ + { + id: "risk-a", + name: "Risk A", + total: 3, + counts: { running: 3 }, + unreadTerminal: 0, + activeTaskId: null, + runnableTaskId: null, + updatedAt: "2026-05-20T00:20:00.000Z", + }, + { + id: "risk-b", + name: "Risk B", + total: 1, + counts: { running: 1 }, + unreadTerminal: 0, + activeTaskId: null, + runnableTaskId: null, + updatedAt: "2026-05-20T00:19:00.000Z", + }, + { + id: "risk-queued", + name: "Risk Queued", + total: 2, + counts: { queued: 1, retry_wait: 1 }, + unreadTerminal: 0, + activeTaskId: null, + runnableTaskId: "task-queued-risk", + updatedAt: "2026-05-20T00:18:00.000Z", + }, + ], + }, + }; +} + +function noActiveResponse(): JsonRecord { + return { + ok: true, + status: 200, + body: { + ok: true, + queue: { + total: 2, + queueCount: 2, + activeQueueIds: [], + activeTaskIds: [], + queuedTaskIds: [], + counts: { succeeded: 2 }, + unreadTerminal: 1, + executionDiagnostics: { + state: "healthy", + effectiveLiveness: "healthy", + recommendedAction: "none", + databaseActiveTaskCount: 0, + databaseActiveTaskIds: [], + schedulerActiveRunSlotCount: 0, + schedulerActiveTaskIds: [], + activeHeartbeatCount: 0, + activeHeartbeatTaskIds: [], + heartbeatFreshTaskIds: [], + heartbeatExpiredTaskIds: [], + heartbeatMissingTaskIds: [], + staleRecoveryCandidateTaskIds: [], + heartbeatRiskTaskIds: [], + }, + }, + queues: [ + { + id: "done-a", + name: "Done A", + total: 1, + counts: { succeeded: 1 }, + unreadTerminal: 1, + activeTaskId: null, + runnableTaskId: null, + updatedAt: "2026-05-20T00:05:00.000Z", + }, + { + id: "done-b", + name: "Done B", + total: 1, + counts: { succeeded: 1 }, + unreadTerminal: 0, + activeTaskId: null, + runnableTaskId: null, + updatedAt: "2026-05-20T00:04:00.000Z", + }, + ], + }, + }; +} + +function activeBelowTargetResponse(): JsonRecord { + return { + ok: true, + status: 200, + body: { + ok: true, + queue: { + total: 8, + queueCount: 3, + activeQueueIds: ["feature-a", "feature-b"], + activeTaskIds: ["task-active-a", "task-active-b"], + queuedTaskIds: ["task-queued-a", "task-queued-b", "task-queued-c"], + counts: { running: 2, queued: 3, retry_wait: 1, succeeded: 2 }, + unreadTerminal: 0, + executionDiagnostics: { + state: "healthy", + effectiveLiveness: "healthy", + recommendedAction: "none", + databaseActiveTaskCount: 2, + databaseActiveTaskIds: ["task-active-a", "task-active-b"], + schedulerActiveRunSlotCount: 2, + schedulerActiveTaskIds: ["task-active-a", "task-active-b"], + activeHeartbeatCount: 2, + activeHeartbeatTaskIds: ["task-active-a", "task-active-b"], + heartbeatFreshTaskIds: ["task-active-a", "task-active-b"], + heartbeatExpiredTaskIds: [], + heartbeatMissingTaskIds: [], + staleRecoveryCandidateTaskIds: [], + heartbeatRiskTaskIds: [], + lastSchedulerHeartbeatAt: "2026-05-20T01:00:00.000Z", + lastObservedAgentEventAt: "2026-05-20T01:00:05.000Z", + }, + }, + queues: [ + { + id: "feature-a", + name: "Feature A", + total: 3, + counts: { running: 1, queued: 1, succeeded: 1 }, + unreadTerminal: 0, + activeTaskId: "task-active-a", + runnableTaskId: null, + updatedAt: "2026-05-20T01:01:00.000Z", + }, + { + id: "feature-b", + name: "Feature B", + total: 3, + counts: { running: 1, retry_wait: 1, succeeded: 1 }, + unreadTerminal: 0, + activeTaskId: "task-active-b", + runnableTaskId: "task-queued-b", + updatedAt: "2026-05-20T01:00:30.000Z", + }, + { + id: "feature-c", + name: "Feature C", + total: 2, + counts: { queued: 2 }, + unreadTerminal: 0, + activeTaskId: null, + runnableTaskId: "task-queued-c", + updatedAt: "2026-05-20T01:00:10.000Z", + }, + ], + tasks: [ + { + id: "task-active-a", + queueId: "feature-a", + status: "running", + displayPrompt: "Implement feature A polish and contract tests", + }, + { + id: "task-active-b", + queueId: "feature-b", + status: "running", + displayPrompt: "Fix feature B queue handling", + }, + { + id: "task-queued-a", + queueId: "feature-a", + status: "queued", + displayPrompt: "Queued feature A follow-up", + }, + ], + }, + }; +} + function assertQueuesShape(label: string, result: unknown, expectedView: string): void { const data = asRecord(result); const queues = asRecord(data.queues); @@ -175,6 +393,12 @@ function assertQueuesShape(label: string, result: unknown, expectedView: string) assertCondition(activity.activeQueueIdsScope === "scheduler-local-active-run-slots", `${label} activity should label activeQueueIds scope`, activity); assertCondition(Array.isArray(queues.activeTaskIds), `${label} activeTaskIds should be present`, queues); assertCondition(Array.isArray(queues.queuedTaskIds), `${label} queuedTaskIds should be present`, queues); + const commander = asRecord(queues.commander); + assertCondition(commander.activeRunnerCount === 1, `${label} commander block should be near the front and expose active runner count`, commander); + assertCondition(commander.target === 15 && commander.slotDeficit === 14, `${label} commander block should expose 15 target slot deficit`, commander); + assertCondition(commander.queuedCount === 2, `${label} commander block should expose queued count`, commander); + assertCondition(asArray(asRecord(commander.runningTasks).items).length >= 1, `${label} commander runningTasks should expose active task ids`, commander); + assertCondition(asArray(asRecord(asRecord(commander.heartbeat).fresh).items).includes("task-running"), `${label} commander heartbeat should foreground fresh ids`, commander); } function assertSplitBrainLiveActivity(label: string, result: unknown): void { @@ -209,6 +433,50 @@ function assertSplitBrainLiveActivity(label: string, result: unknown): void { assertCondition(String(commanderConcurrency.decisionRule ?? "").includes("15 - activeRunnerCount"), `${label} should give 15-concurrency arithmetic`, commanderConcurrency); assertCondition(String(activity.activeQueueIdsNote ?? "").includes("zero local queue ids does not mean zero active runners"), `${label} activity note should prevent zero-active misread`, activity); assertCondition(String(activity.interpretation ?? "").includes("continue supervision"), `${label} activity interpretation should keep supervision action`, activity); + const commander = asRecord(queues.commander); + assertCondition(commander.activeRunnerCount === 8, `${label} commander should foreground heartbeat-effective active count`, commander); + assertCondition(commander.slotDeficit === 7 && commander.target === 15, `${label} commander should compute below-target slot deficit`, commander); + assertCondition(asRecord(commander.runningTasks).count === 8, `${label} commander runningTasks should preserve live active count`, commander); + assertCondition(asRecord(asRecord(commander.heartbeat).fresh).count === 8, `${label} commander heartbeat should preserve fresh heartbeat count`, commander); + assertCondition(asRecord(asRecord(commander.heartbeat).risk).count === 0, `${label} commander heartbeat risk should be zero for live split-brain`, commander); +} + +function assertHeartbeatRiskCommander(label: string, result: unknown): void { + const queues = asRecord(asRecord(result).queues); + const commander = asRecord(queues.commander); + const heartbeat = asRecord(commander.heartbeat); + const risk = asRecord(heartbeat.risk); + const stale = asRecord(heartbeat.staleRecoveryCandidates); + assertCondition(commander.activeRunnerCount === 4, `${label} should count database-active stale tasks as active before recovery`, commander); + assertCondition(commander.slotDeficit === 11 && commander.target === 15, `${label} should compute slot deficit from target 15`, commander); + assertCondition(commander.queuedCount === 2, `${label} should expose queued + retry_wait count`, commander); + assertCondition(commander.attentionRequired === true && commander.interventionRequired === false, `${label} should require attention but not high-risk intervention from one snapshot`, commander); + assertCondition(heartbeat.effectiveLiveness === "at-risk", `${label} heartbeat liveness should be at-risk`, heartbeat); + assertCondition(risk.count === 4 && asArray(risk.items).length > 0, `${label} should expose bounded heartbeat risk ids`, risk); + assertCondition(stale.count === 4 && String(stale.command ?? "").includes("dryRun=1"), `${label} should expose stale recovery candidates and dry-run command`, stale); +} + +function assertNoActiveCommander(label: string, result: unknown): void { + const commander = asRecord(asRecord(asRecord(result).queues).commander); + assertCondition(commander.activeRunnerCount === 0, `${label} should expose zero active runners`, commander); + assertCondition(commander.slotDeficit === 15 && commander.slotDeficitState === "below-target", `${label} should expose full 15-slot deficit`, commander); + assertCondition(commander.queuedCount === 0, `${label} queued count should be zero`, commander); + assertCondition(asRecord(commander.runningTasks).count === 0, `${label} running task list should be empty`, commander); +} + +function assertBelowTargetCommander(label: string, result: unknown): void { + const commander = asRecord(asRecord(asRecord(result).queues).commander); + const runningTasks = asRecord(commander.runningTasks); + const runningItems = asArray(runningTasks.items).map(asRecord); + const activeQueues = asRecord(commander.activeQueues); + const runnableQueues = asRecord(commander.runnableQueues); + assertCondition(commander.activeRunnerCount === 2, `${label} should foreground active runner count`, commander); + assertCondition(commander.slotDeficit === 13 && commander.target === 15, `${label} should expose deficit below target`, commander); + assertCondition(commander.queuedCount === 4, `${label} should include queued and retry_wait`, commander); + assertCondition(runningItems.some((item) => item.id === "task-active-a" && item.name === "Implement feature A polish and contract tests"), `${label} should include active task names when upstream task rows are available`, runningTasks); + assertCondition(runningItems.some((item) => asArray(item.queueIds).includes("feature-a") && asArray(item.queueNames).includes("Feature A")), `${label} should include active task queues`, runningTasks); + assertCondition(activeQueues.count === 2 && runnableQueues.count === 3, `${label} should keep active and queued queue rows separate`, commander); + assertCondition(String(asRecord(commander.commands).running ?? "").includes("--status running,judging"), `${label} should expose running drill-down command`, commander); } export function runCodeQueueQueuesShapeContract(): JsonRecord { @@ -220,11 +488,28 @@ export function runCodeQueueQueuesShapeContract(): JsonRecord { assertCondition(path === "/api/microservices/code-queue/proxy/api/queues", "codex queues should use stable proxy path", { path }); return splitBrainLiveResponse(); }; + const heartbeatRiskFetcher = (path: string): JsonRecord => { + assertCondition(path === "/api/microservices/code-queue/proxy/api/queues", "codex queues should use stable proxy path", { path }); + return heartbeatRiskResponse(); + }; + const noActiveFetcher = (path: string): JsonRecord => { + assertCondition(path === "/api/microservices/code-queue/proxy/api/queues", "codex queues should use stable proxy path", { path }); + return noActiveResponse(); + }; + const activeBelowTargetFetcher = (path: string): JsonRecord => { + assertCondition(path === "/api/microservices/code-queue/proxy/api/queues", "codex queues should use stable proxy path", { path }); + return activeBelowTargetResponse(); + }; const summary = codexQueuesQueryForTest([], fetcher); assertQueuesShape("summary", summary, "summary"); const summaryQueues = asRecord(asRecord(summary).queues); assertCondition(summaryQueues.deprecatedFullArray === undefined, "summary should not expose deprecated full array compatibility field", summaryQueues); + assertCondition(summaryQueues.summaryMode === "commander-first", "summary should default to commander-first mode", summaryQueues); + + const explicitCommander = codexQueuesQueryForTest(["--commander"], fetcher); + const explicitCommanderQueues = asRecord(asRecord(explicitCommander).queues); + assertCondition(explicitCommanderQueues.summaryMode === "commander-first", "--commander should keep commander-first mode", explicitCommanderQueues); const full = codexQueuesQueryForTest(["--full"], fetcher); assertQueuesShape("full", full, "full"); @@ -255,6 +540,12 @@ export function runCodeQueueQueuesShapeContract(): JsonRecord { assertSplitBrainLiveActivity("split-brain summary", splitSummary); const splitFull = codexQueuesQueryForTest(["--full"], splitBrainFetcher); assertSplitBrainLiveActivity("split-brain full", splitFull); + const heartbeatRisk = codexQueuesQueryForTest([], heartbeatRiskFetcher); + assertHeartbeatRiskCommander("heartbeat-risk summary", heartbeatRisk); + const noActive = codexQueuesQueryForTest([], noActiveFetcher); + assertNoActiveCommander("no-active summary", noActive); + const belowTarget = codexQueuesQueryForTest([], activeBelowTargetFetcher); + assertBelowTargetCommander("active-below-target summary", belowTarget); return { ok: true, @@ -268,6 +559,8 @@ export function runCodeQueueQueuesShapeContract(): JsonRecord { "offset pagination", "split-brain live activity counts distinguish scheduler-local queues, DB running tasks, and heartbeat-fresh runners", "commander concurrency block names the active runner count and 15-concurrency rule", + "queues commander-first summary foregrounds active runner count, target 15 deficit, running task ids/names/queues, heartbeat risk, stale recovery candidates, and queued count", + "queues commander fixtures cover split-brain live, heartbeat risk, no active, and active below target", ], }; } diff --git a/scripts/src/check.ts b/scripts/src/check.ts index f37cd20f..c6b0713d 100644 --- a/scripts/src/check.ts +++ b/scripts/src/check.ts @@ -47,6 +47,7 @@ const syntaxFiles = [ "scripts/code-queue-cli-read-terminal-contract-test.ts", "scripts/code-queue-gh-auth-redaction-contract-test.ts", "scripts/d601-recovery-guardrails-contract-test.ts", + "scripts/code-queue-queues-shape-contract-test.ts", "scripts/microservice-health-output-contract-test.ts", "scripts/code-queue-supervisor-disclosure-contract-test.ts", "scripts/code-queue-commander-view-contract-test.ts", @@ -354,6 +355,7 @@ export function runChecks(config: UniDeskConfig, options: CheckOptions = default fileItem("scripts/code-queue-submit-summary-contract-test.ts"), fileItem("scripts/code-queue-submit-routing-contract-test.ts"), fileItem("scripts/code-queue-gh-auth-redaction-contract-test.ts"), + fileItem("scripts/code-queue-queues-shape-contract-test.ts"), fileItem("scripts/code-queue-supervisor-disclosure-contract-test.ts"), fileItem("scripts/code-queue-commander-view-contract-test.ts"), fileItem("scripts/host-codex-commander-skeleton-contract-test.ts"), @@ -408,6 +410,7 @@ export function runChecks(config: UniDeskConfig, options: CheckOptions = default items.push(commandItem("code-queue:submit-summary-contract", ["bun", "scripts/code-queue-submit-summary-contract-test.ts"], 30_000)); items.push(commandItem("code-queue:submit-routing-contract", ["bun", "scripts/code-queue-submit-routing-contract-test.ts"], 30_000)); items.push(commandItem("code-queue:gh-auth-redaction-contract", ["bun", "scripts/code-queue-gh-auth-redaction-contract-test.ts"], 30_000)); + items.push(commandItem("code-queue:queues-shape-contract", ["bun", "scripts/code-queue-queues-shape-contract-test.ts"], 30_000)); items.push(commandItem("code-queue:supervisor-disclosure-contract", ["bun", "scripts/code-queue-supervisor-disclosure-contract-test.ts"], 30_000)); items.push(commandItem("code-queue:commander-view-contract", ["bun", "scripts/code-queue-commander-view-contract-test.ts"], 30_000)); items.push(commandItem("host-codex-commander:skeleton-contract", ["bun", "scripts/host-codex-commander-skeleton-contract-test.ts"], 30_000)); @@ -450,6 +453,7 @@ export function runChecks(config: UniDeskConfig, options: CheckOptions = default items.push(skippedItem("code-queue:submit-summary-contract", "Code Queue submit summary contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("code-queue:submit-routing-contract", "Code Queue submit routing contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("code-queue:gh-auth-redaction-contract", "Code Queue GitHub auth output redaction contract is opt-in with script checks", "--scripts-typecheck or --full")); + items.push(skippedItem("code-queue:queues-shape-contract", "Code Queue queues shape contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("code-queue:supervisor-disclosure-contract", "Code Queue supervisor disclosure contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("code-queue:commander-view-contract", "Code Queue commander view contract is opt-in with script checks", "--scripts-typecheck or --full")); items.push(skippedItem("host-codex-commander:skeleton-contract", "host Codex commander skeleton contract is opt-in with script checks", "--scripts-typecheck or --full")); diff --git a/scripts/src/code-queue.ts b/scripts/src/code-queue.ts index 0dc84256..2fb00f62 100644 --- a/scripts/src/code-queue.ts +++ b/scripts/src/code-queue.ts @@ -37,6 +37,7 @@ const commanderRecentCompletedLimit = 3; const commanderPromptPreviewChars = 96; const commanderBodyPreviewChars = 120; const commanderIssueTaskPreviewLimit = 4; +const commanderConcurrencyTarget = 15; const unreadTriageCountLimit = 12; const diagnosticsIdPreviewLimit = 3; const diagnosticsReasonPreviewLimit = 2; @@ -492,6 +493,7 @@ interface SupervisorStatusCounts { interface CodexQueuesOptions { full: boolean; + commander: boolean; limit: number; offset: number; page: number; @@ -2663,7 +2665,7 @@ function parseUnreadOptions(args: string[]): CodexUnreadOptions { function parseQueuesOptions(args: string[]): CodexQueuesOptions { assertKnownOptions(args, { - flags: ["--full", "--all"], + flags: ["--full", "--all", "--commander"], valueOptions: ["--limit", "--offset", "--page"], }, "codex queues"); const limit = positiveIntegerOption(args, ["--limit"], defaultQueuesLimit, maxTasksLimit); @@ -2672,6 +2674,7 @@ function parseQueuesOptions(args: string[]): CodexQueuesOptions { const offset = offsetExplicit ? nonNegativeIntegerOption(args, ["--offset"], 0) : (page - 1) * limit; return { full: hasFlag(args, "--full") || hasFlag(args, "--all"), + commander: hasFlag(args, "--commander") || !hasFlag(args, "--full") && !hasFlag(args, "--all"), limit, offset, page: Math.floor(offset / limit) + 1, @@ -4667,6 +4670,233 @@ function compactQueueRow(value: unknown): Record { }; } +function taskNameFromRecord(task: Record | null): string | null { + if (task === null) return null; + for (const key of ["name", "title", "summary", "displayPrompt", "basePrompt", "prompt"]) { + const value = asString(task[key]).replace(/\s+/gu, " ").trim(); + if (value.length > 0) return value.slice(0, 96); + } + return null; +} + +function buildQueueLookup(queues: Record[]): Map> { + const lookup = new Map>(); + for (const queue of queues) { + const id = asString(queue.id); + if (id.length > 0) lookup.set(id, queue); + } + return lookup; +} + +function buildTaskLookupFromQueuesBody(body: Record): Map> { + const lookup = new Map>(); + for (const source of [body.tasks, asRecord(body.overview)?.tasks, asRecord(body.taskPage)?.tasks]) { + for (const item of asArray(source)) { + const task = asRecord(item); + const id = asString(task?.id); + if (task !== null && id.length > 0 && !lookup.has(id)) lookup.set(id, task); + } + } + return lookup; +} + +function queueIdsForTaskId(taskId: string, queues: Record[]): string[] { + const ids: string[] = []; + for (const queue of queues) { + const queueId = asString(queue.id); + if (queueId.length === 0) continue; + if (asString(queue.activeTaskId) === taskId || asString(queue.runnableTaskId) === taskId) ids.push(queueId); + } + return ids; +} + +function compactCommanderTaskList( + ids: string[], + queues: Record[], + queueLookup: Map>, + taskLookup: Map>, + limit: number, + source: string, + totalCount?: number, +): Record { + const all = orderedUniqueStringList(ids); + const count = Math.max(all.length, totalCount ?? all.length); + const items = all.slice(0, limit).map((id) => { + const task = taskLookup.get(id) ?? null; + const taskQueueId = asString(task?.queueId) || asString(task?.queue) || null; + const inferredQueueIds = taskQueueId === null ? queueIdsForTaskId(id, queues) : [taskQueueId]; + const queueRecords = inferredQueueIds.map((queueId) => queueLookup.get(queueId) ?? null).filter((queue): queue is Record => queue !== null); + const queueNames = queueRecords.map((queue) => asString(queue.name)).filter((name) => name.length > 0); + const name = taskNameFromRecord(task); + return { + id, + name, + nameSource: name === null ? "not-returned-by-api-queues" : "task-row", + queueIds: inferredQueueIds, + queueNames, + queueSource: taskQueueId !== null + ? "task-row" + : inferredQueueIds.length > 0 + ? "queue-row-active-or-runnable-task-id" + : "not-returned-by-api-queues", + commands: { + show: `bun scripts/cli.ts codex task ${id}`, + detail: `bun scripts/cli.ts codex task ${id} --detail`, + trace: `bun scripts/cli.ts codex task ${id} --trace --tail --limit ${defaultTraceLimit}`, + }, + }; + }); + return { + count, + returned: items.length, + omitted: Math.max(0, count - items.length), + truncated: count > items.length || all.length > items.length, + idsUnavailable: count > 0 && items.length === 0, + source, + items, + }; +} + +function sortedQueueRowsForTaskStatus(queues: Record[], statuses: string[]): Record[] { + return queues + .filter((queue) => statuses.some((status) => countForStatus(asRecord(queue.counts) ?? {}, status) > 0)) + .sort((left, right) => { + const leftUpdated = timestampMs(left.updatedAt) ?? 0; + const rightUpdated = timestampMs(right.updatedAt) ?? 0; + return rightUpdated - leftUpdated || asString(left.id).localeCompare(asString(right.id)); + }); +} + +function compactCommanderQueueRows(queues: Record[], limit: number, statuses: string[], source: string): Record { + const selected = sortedQueueRowsForTaskStatus(queues, statuses); + const items = selected.slice(0, limit).map((queue) => ({ + id: queue.id ?? null, + name: queue.name ?? null, + counts: queue.counts ?? {}, + activeTaskId: queue.activeTaskId ?? null, + runnableTaskId: queue.runnableTaskId ?? null, + unreadTerminal: queue.unreadTerminal ?? 0, + updatedAt: queue.updatedAt ?? null, + commands: queue.commands ?? {}, + })); + return { + count: selected.length, + returned: items.length, + omitted: Math.max(0, selected.length - items.length), + truncated: selected.length > items.length, + source, + items, + }; +} + +function compactCommanderHeartbeatList(ids: string[], totalCount: number, limit: number, source: string, command: string): Record { + const knownIds = orderedUniqueStringList(ids); + const preview = compactIdPreview(knownIds, Math.max(totalCount, knownIds.length), limit, source); + return { + ...preview, + command, + }; +} + +function heartbeatIdsFromDiagnostics(rawDiagnostics: Record, diagnostics: Record | null, keys: string[]): string[] { + const ids: string[] = []; + for (const key of keys) { + ids.push(...idPreviewInputItems(rawDiagnostics[key])); + ids.push(...idPreviewInputItems(diagnostics?.[key])); + } + return orderedUniqueStringList(ids); +} + +function buildQueuesCommanderSummary(input: { + queue: Record; + queues: Record[]; + diagnostics: Record | null; + activity: Record; + commanderConcurrency: Record; + activeTaskIds: { items: string[]; count: number; omitted: number; truncated: boolean }; + queuedTaskIds: { items: string[]; count: number; omitted: number; truncated: boolean }; + options: CodexQueuesOptions; + body: Record; +}): Record { + const rawDiagnostics = asRecord(input.queue.executionDiagnostics) ?? {}; + const liveness = asRecord(input.diagnostics?.liveness) ?? {}; + const recovery = asRecord(input.activity.recovery) ?? asRecord(input.diagnostics?.recovery) ?? {}; + const activeRunnerCount = asNumber(input.commanderConcurrency.activeRunnerCount, asNumber(input.activity.effectiveActiveTaskCount, 0)); + const target = commanderConcurrencyTarget; + const slotDeficit = Math.max(0, target - activeRunnerCount); + const slotDeficitState = activeRunnerCount >= target ? "at-or-above-target" : "below-target"; + const queuedCount = countForStatus(asRecord(input.queue.counts) ?? {}, "queued") + countForStatus(asRecord(input.queue.counts) ?? {}, "retry_wait"); + const queueLookup = buildQueueLookup(input.queues); + const taskLookup = buildTaskLookupFromQueuesBody(input.body); + const activeIds = orderedUniqueStringList([ + ...input.activeTaskIds.items, + ...idPreviewInputItems(rawDiagnostics.databaseActiveTaskIds), + ...idPreviewInputItems(rawDiagnostics.heartbeatFreshTaskIds), + ...idPreviewInputItems(input.diagnostics?.heartbeatFreshTaskIds), + ]); + const queuedIds = orderedUniqueStringList(input.queuedTaskIds.items); + const heartbeatFreshIds = heartbeatIdsFromDiagnostics(rawDiagnostics, input.diagnostics, ["heartbeatFreshTaskIds", "activeHeartbeatTaskIds"]); + const heartbeatRiskIds = heartbeatIdsFromDiagnostics(rawDiagnostics, input.diagnostics, ["heartbeatRiskTaskIds", "heartbeatExpiredTaskIds", "heartbeatMissingTaskIds"]); + const staleCandidateIds = heartbeatIdsFromDiagnostics(rawDiagnostics, input.diagnostics, ["staleRecoveryCandidateTaskIds"]); + const taskPreviewLimit = Math.min(input.options.limit, commanderSectionReturnedLimit); + return { + view: "commander", + fieldOrder: [ + "activeRunnerCount", + "source", + "target", + "slotDeficit", + "runningTasks", + "heartbeat", + "queuedCount", + "queueRows", + "historyRows", + ], + activeRunnerCount, + source: input.commanderConcurrency.activeRunnerCountSource ?? input.activity.effectiveActiveSource ?? "activity.effectiveActiveTaskCount", + activeRunnerCountField: input.commanderConcurrency.activeRunnerCountField ?? "activity.effectiveActiveTaskCount", + target, + slotDeficit, + slotDeficitState, + queuedCount, + queuedSource: "queue.counts.queued + queue.counts.retry_wait", + counts: input.queue.counts ?? {}, + runningTasks: compactCommanderTaskList(activeIds, input.queues, queueLookup, taskLookup, taskPreviewLimit, "activeTaskIds + executionDiagnostics.databaseActiveTaskIds + heartbeatFreshTaskIds", activeRunnerCount), + queuedTasks: compactCommanderTaskList(queuedIds, input.queues, queueLookup, taskLookup, taskPreviewLimit, "queue.queuedTaskIds", queuedCount), + activeQueues: compactCommanderQueueRows(input.queues, taskPreviewLimit, ["running", "judging"], "per-queue counts running/judging"), + runnableQueues: compactCommanderQueueRows(input.queues, taskPreviewLimit, ["queued", "retry_wait"], "per-queue counts queued/retry_wait"), + heartbeat: { + effectiveLiveness: input.diagnostics?.effectiveLiveness ?? input.activity.effectiveLiveness ?? null, + splitBrainLive: input.diagnostics?.splitBrainLive ?? input.activity.splitBrainLive ?? null, + lastSchedulerHeartbeatAt: input.diagnostics?.lastSchedulerHeartbeatAt ?? recovery.lastSchedulerHeartbeatAt ?? null, + lastObservedAgentEventAt: input.diagnostics?.lastObservedAgentEventAt ?? recovery.lastObservedAgentEventAt ?? null, + fresh: compactCommanderHeartbeatList(heartbeatFreshIds, asNumber(input.activity.heartbeatFreshActiveTaskCount, heartbeatFreshIds.length), taskPreviewLimit, "executionDiagnostics.heartbeatFreshTaskIds + activeHeartbeatTaskIds", "bun scripts/cli.ts codex tasks --view supervisor --status running,judging --limit 20"), + risk: compactCommanderHeartbeatList(heartbeatRiskIds, asNumber(input.activity.heartbeatRiskTaskCount, heartbeatRiskIds.length), taskPreviewLimit, "executionDiagnostics.heartbeatRiskTaskIds + heartbeatExpiredTaskIds + heartbeatMissingTaskIds", "bun scripts/cli.ts codex tasks --view supervisor --limit 20"), + staleRecoveryCandidates: compactCommanderHeartbeatList(staleCandidateIds, asNumber(input.activity.staleRecoveryCandidateTaskCount, staleCandidateIds.length), taskPreviewLimit, "executionDiagnostics.staleRecoveryCandidateTaskIds", "bun scripts/cli.ts microservice proxy code-queue '/api/scheduler/reconcile?dryRun=1' --raw"), + recoveryHint: input.commanderConcurrency.recoveryHint ?? recovery.hint ?? liveness.recoveryHint ?? null, + }, + attentionRequired: input.commanderConcurrency.attentionRequired ?? false, + interventionRequired: input.commanderConcurrency.interventionRequired ?? false, + splitBrainDisposition: input.commanderConcurrency.splitBrainDisposition ?? input.activity.splitBrainDisposition ?? null, + decisionRule: `remaining slots = ${target} - activeRunnerCount; slotDeficit is clamped at 0 when activeRunnerCount >= target.`, + historyRows: { + returned: Math.min(input.options.limit, input.queues.length), + secondary: true, + stablePath: "data.queues.items[]", + note: "Queue history rows are paged and secondary; use commander.runningTasks, heartbeat, and queuedCount first for concurrency decisions.", + }, + commands: { + refresh: queueListCommand({ full: input.options.full, commander: input.options.commander, limit: input.options.limit, offset: input.options.offset }), + supervisor: "bun scripts/cli.ts codex tasks --view supervisor --limit 20", + running: "bun scripts/cli.ts codex tasks --view supervisor --status running,judging --limit 20", + unread: "bun scripts/cli.ts codex unread --limit 20", + fullQueues: queueListCommand({ full: true, limit: input.options.limit, offset: 0 }), + rawQueues: "bun scripts/cli.ts microservice proxy code-queue /api/queues --raw --full", + rawOverview: rawCodeQueueOverviewCommand, + }, + }; +} + function queueListCommand(options: Partial = {}): string { const full = options.full === true; const limit = options.limit ?? defaultQueuesLimit; @@ -4674,6 +4904,7 @@ function queueListCommand(options: Partial = {}): string { return [ "bun scripts/cli.ts codex queues", full ? "--full" : "", + !full && options.commander === true ? "--commander" : "", limit === defaultQueuesLimit ? "" : `--limit ${limit}`, offset > 0 ? `--offset ${offset}` : "", ].filter(Boolean).join(" "); @@ -4698,16 +4929,30 @@ function compactQueuesResponse(body: Record, options: CodexQueu const previousOffset = Math.max(0, options.offset - options.limit); const hasMore = nextOffset < selected.length; const hasPrevious = options.offset > 0; + const commander = buildQueuesCommanderSummary({ + queue, + queues, + diagnostics, + activity, + commanderConcurrency, + activeTaskIds, + queuedTaskIds, + options, + body, + }); return { upstream, queues: { view: options.full ? "full" : "summary", + summaryMode: options.commander ? "commander-first" : "full-paged", bounded: true, outputPolicy: { - default: "paged-low-noise", + default: options.commander ? "commander-first-paged-low-noise" : "paged-low-noise", stableItemsPath: "data.queues.items[]", + commanderPath: "data.queues.commander", rawFullCommand: "bun scripts/cli.ts microservice proxy code-queue /api/queues --raw --full", }, + commander, count: selected.length, returned: visible.length, limit: options.limit, @@ -4757,10 +5002,10 @@ function compactQueuesResponse(body: Record, options: CodexQueu } : {}), commands: { - refresh: queueListCommand({ full: options.full, limit: options.limit, offset: options.offset }), - next: hasMore ? queueListCommand({ full: options.full, limit: options.limit, offset: nextOffset }) : null, - previous: hasPrevious ? queueListCommand({ full: options.full, limit: options.limit, offset: previousOffset }) : null, - first: queueListCommand({ full: options.full, limit: options.limit, offset: 0 }), + refresh: queueListCommand({ full: options.full, commander: options.commander, limit: options.limit, offset: options.offset }), + next: hasMore ? queueListCommand({ full: options.full, commander: options.commander, limit: options.limit, offset: nextOffset }) : null, + previous: hasPrevious ? queueListCommand({ full: options.full, commander: options.commander, limit: options.limit, offset: previousOffset }) : null, + first: queueListCommand({ full: options.full, commander: options.commander, limit: options.limit, offset: 0 }), full: queueListCommand({ full: true, limit: options.limit, offset: 0 }), tasks: `bun scripts/cli.ts codex tasks --view supervisor --limit ${Math.min(options.limit, defaultTasksLimit)}`, unread: `bun scripts/cli.ts codex unread --limit ${Math.min(options.limit, defaultTasksLimit)}`, diff --git a/scripts/src/help.ts b/scripts/src/help.ts index 0f6ddd37..fbecbf91 100644 --- a/scripts/src/help.ts +++ b/scripts/src/help.ts @@ -284,7 +284,7 @@ function codexHelp(): unknown { "bun scripts/cli.ts codex resume [prompt|--prompt-file path|--prompt-stdin] [--resume-id id] [--dry-run] [--full|--raw]", "bun scripts/cli.ts codex steer-confirm --steer-id [--raw]", "bun scripts/cli.ts codex interrupt|cancel ", - "bun scripts/cli.ts codex queues [--full|--all] [--limit N] [--page N|--offset N] | queue create | queue merge --into | move --queue ", + "bun scripts/cli.ts codex queues [--commander] [--full|--all] [--limit N] [--page N|--offset N] | queue create | queue merge --into | move --queue ", ], promptInput: { recommended: ["--prompt-stdin", "--prompt-file"], @@ -313,6 +313,13 @@ function codexHelp(): unknown { mutationGuard: "Batch mark-read is blocked unless the explicit mark-read subcommand is used with --confirm; use codex read for per-task review.", disclosure: "Raw prompt, final response, trace and output are omitted; use the returned task/detail/trace/output/read commands for drill-down.", }, + queuesSummary: { + default: "codex queues defaults to commander-first summary mode; --commander is an explicit alias, while --full keeps the same bounded queue rows with compatibility metadata.", + commanderPath: "data.queues.commander", + stableRowsPath: "data.queues.items[]", + fields: ["activeRunnerCount", "source", "target=15", "slotDeficit", "runningTasks", "heartbeat.fresh", "heartbeat.risk", "heartbeat.staleRecoveryCandidates", "queuedCount"], + rule: "Use data.queues.commander.activeRunnerCount and slotDeficit for quick capacity decisions; activeQueueIds are scheduler-local and can be empty during split-brain live.", + }, examples: { promptLint: "bun scripts/cli.ts codex prompt-lint --prompt-file /tmp/code-queue-prompt.md", stdin: [