From c6a27e6c2bc51f1bbbea5b50c9dfc460dbf0c0b6 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 20 May 2026 20:00:39 +0000 Subject: [PATCH] fix: unify codex queues json shape --- docs/reference/cli.md | 2 +- docs/reference/code-queue-supervision.md | 2 +- .../code-queue-queues-shape-contract-test.ts | 140 ++++++++++++++++++ scripts/src/code-queue.ts | 27 +++- 4 files changed, 164 insertions(+), 7 deletions(-) create mode 100644 scripts/code-queue-queues-shape-contract-test.ts diff --git a/docs/reference/cli.md b/docs/reference/cli.md index eddd538f..d26ddcc6 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -50,7 +50,7 @@ CLI 可以从 `master` 快速演进，但必须兼容 `deploy.json` 固定的 CI - `codex steer [prompt|--prompt-file path|--prompt-stdin] [--dry-run]` 通过 Code Queue 私有代理向正在运行的 task 注入纠偏提示，正式替代底层 `microservice proxy code-queue /api/tasks//steer` 调用。prompt 必须且只能来自位置参数、文件或 stdin 之一；`--dry-run` 只输出 `method`、`path`、`stableProxyPath`、prompt 字符数、截断预览和 raw proxy 等价命令，不触碰运行中 session，也不得泄露超长 prompt 全文。真实执行复用与 `codex task/tasks/read` 相同的 backend-core stable proxy helper，路径固定为 `/api/microservices/code-queue/proxy/api/tasks//steer`，只能作用于 D601 scheduler 上存在 active steerable turn 的 running task。 - `codex steer` 非 dry-run 失败仍输出 JSON 且退出非零；`.data.diagnostics.reason` 用于 runner 分流，当前包括 `backend-core-unreachable`、`code-queue-microservice-unregistered`、`proxy-unauthorized`、`proxy-404`、`steer-endpoint-404`、`upstream-runtime-rejected`、`stable-proxy-failed` 和 `invalid-proxy-response`。`scope` 区分 `backend-core`、`stable-proxy`、`code-queue-runtime` 或 `unknown`，并带 `status`、`exitCode`、`retryable`、有界 `upstreamBodyPreview` 和推荐交叉验证命令；若任务不在 running/active-turn 状态，通常归类为 `upstream-runtime-rejected`，不得静默成功。 - `codex interrupt|cancel ` 通过 Code Queue 私有代理请求中断；running/judging 任务会请求 D601 当前 agent run 停止，queued/retry_wait 任务的取消也必须保持与 WebUI 相同代理路径，返回有界 task 摘要和后续查询命令。任何需要接触 active run 的动作仍属于 D601 执行面。 -- Code Queue 多队列 lane 由 `codex` 命令命名空间管理：`queues [--full|--all]` 列表、`queue create ` 创建、`queue merge --into ` 合并、`move --queue ` 迁移；这些队列管理入口默认由主 server `code-queue-mgr` 直管 PostgreSQL，仍通过稳定 `code-queue` 用户服务代理路径访问。`codex queues` 默认只返回 active/nonempty/unread/runnable queue 摘要、全局 counts 和 execution diagnostics；完整队列数组必须显式 `--full` 或 `--all`。同一个 queue 内部串行执行，不同 queue 之间并行执行。迁移只允许尚未被 scheduler claim 的 `queued`/`retry_wait` 任务，必须满足 `startedAt=null`、`currentAttempt=0` 且没有 active thread/turn；已进入 `running`/`judging` 或已有 claim 标记的任务返回 409，不得被 move/merge 回写成 queued。合并会移动可迁移任务归属并自动删除源 queue 记录，只保留合并后的目标 queue；若 source 或 target queue 存在 active/claimed 任务，合并整体返回 409。合并后的目标 queue 按任务原 `queueEnteredAt`/`createdAt` 时间顺序串行，成功迁移 queued/retry_wait 任务后由 D601 scheduler 轮询推进。 +- Code Queue 多队列 lane 由 `codex` 命令命名空间管理：`queues [--full|--all]` 列表、`queue create ` 创建、`queue merge --into ` 合并、`move --queue ` 迁移；这些队列管理入口默认由主 server `code-queue-mgr` 直管 PostgreSQL，仍通过稳定 `code-queue` 用户服务代理路径访问。`codex queues` 默认只返回 active/nonempty/unread/runnable queue 摘要、全局 counts 和 execution diagnostics；完整队列数组必须显式 `--full` 或 `--all`。summary 和 full 的稳定机读路径都是 `.data.queues.items[]`，全局元数据固定在 `.data.queues.counts`、`.data.queues.executionDiagnostics`、`.data.queues.activeTaskIds` 和 `.data.queues.queuedTaskIds`；旧 full 顶层数组语义只允许作为 deprecated 兼容字段 `.data.queues.deprecatedFullArray[]` 出现，不再作为 `.data.queues` 主形态。同一个 queue 内部串行执行，不同 queue 之间并行执行。迁移只允许尚未被 scheduler claim 的 `queued`/`retry_wait` 任务，必须满足 `startedAt=null`、`currentAttempt=0` 且没有 active thread/turn；已进入 `running`/`judging` 或已有 claim 标记的任务返回 409，不得被 move/merge 回写成 queued。合并会移动可迁移任务归属并自动删除源 queue 记录，只保留合并后的目标 queue；若 source 或 target queue 存在 active/claimed 任务，合并整体返回 409。合并后的目标 queue 按任务原 `queueEnteredAt`/`createdAt` 时间顺序串行，成功迁移 queued/retry_wait 任务后由 D601 scheduler 轮询推进。 - 所有 `codex` 查询和管理命令必须走与 WebUI 相同的 backend-core 私有代理路径 `/api/microservices/code-queue/proxy/...`；CLI 不得为了提交、移动、中断、取消或队列管理直接调用 D601 内部 Service、数据库、pod curl 或 k3sctl scheduler 子服务。若该路径失败，应先修复 CLI/backend/provider tunnel 链路，而不是绕过控制面。 - `job list [--limit N] [--include-command]` 与 `job status [--tail-bytes N]` 查询 `.state/jobs/` 文件系统状态，是异步命令的可观测入口。`job list` 默认只返回最新 50 条摘要；`job status` 默认只返回 stdout/stderr 末尾 12000 字节，并带 `tailPolicy` 与完整日志路径。 - `debug health`、`debug dispatch` 与 `debug task` 走真实内部 core、WebSocket、数据库、provider、系统指标、Docker 状态和 Host SSH 维护桥流程，只用于开发调试，不写入 `TEST.md` 的正式验收步骤。 diff --git a/docs/reference/code-queue-supervision.md b/docs/reference/code-queue-supervision.md index 8e996c7c..e3c9ff33 100644 --- a/docs/reference/code-queue-supervision.md +++ b/docs/reference/code-queue-supervision.md @@ -129,7 +129,7 @@ bun scripts/code-queue-pr-preflight-example.ts --repo pikasTech/unidesk --base m 常用入口： - `bun scripts/cli.ts codex tasks --view supervisor --limit N`：查看默认有界监督视图，包括 running、完成未读、最近完成、queued/runnable、execution diagnostics 和下一步 drill-down 命令。 -- `bun scripts/cli.ts codex queues`：查看低噪声队列计数、active task id、完成未读队列、runnable 队列和控制面诊断；只有需要完整队列行时才加 `--full`。 +- `bun scripts/cli.ts codex queues`：查看低噪声队列计数、active task id、完成未读队列、runnable 队列和控制面诊断；只有需要完整队列行时才加 `--full`。summary 和 full 都使用稳定 JSON path `.data.queues.items[]` 读取队列行，并从 `.data.queues.counts` 与 `.data.queues.executionDiagnostics` 读取全局计数和执行诊断。 - `bun scripts/cli.ts codex tasks --unread --limit N`：查看完成未读审阅积压；`--unread` 与 `--unread-only` 等价，不能被静默忽略。 - `bun scripts/cli.ts codex tasks --status succeeded --unread --limit N`：按具体终态过滤监督结果；不支持的 status filter 必须显式失败，不能扩大为未过滤结果。 - `bun scripts/cli.ts codex task `：默认只查看原始 prompt、最终 response、最后错误和 drill-down 命令，这是完成未读任务审阅的第一步。 diff --git a/scripts/code-queue-queues-shape-contract-test.ts b/scripts/code-queue-queues-shape-contract-test.ts new file mode 100644 index 00000000..57dd839e --- /dev/null +++ b/scripts/code-queue-queues-shape-contract-test.ts @@ -0,0 +1,140 @@ +import { codexQueuesQueryForTest } from "./src/code-queue"; + +type JsonRecord = Record; + +function assertCondition(condition: unknown, message: string, detail: JsonRecord = {}): void { + if (!condition) throw new Error(`${message}: ${JSON.stringify(detail)}`); +} + +function asRecord(value: unknown): JsonRecord { + assertCondition(typeof value === "object" && value !== null && !Array.isArray(value), "expected JSON object", { value }); + return value as JsonRecord; +} + +function asArray(value: unknown): unknown[] { + assertCondition(Array.isArray(value), "expected JSON array", { value }); + return value as unknown[]; +} + +function fixtureResponse(): JsonRecord { + return { + ok: true, + status: 200, + body: { + ok: true, + queue: { + total: 4, + queueCount: 3, + activeQueueIds: ["alpha"], + activeTaskIds: ["task-running"], + queuedTaskIds: ["task-queued"], + counts: { running: 1, queued: 2, succeeded: 1 }, + unreadTerminal: 1, + executionDiagnostics: { + state: "split-brain", + splitBrain: true, + heartbeatFreshTaskIds: ["task-running"], + databaseActiveTaskCount: 1, + databaseActiveTaskIds: ["task-running"], + schedulerActiveRunSlotCount: 1, + schedulerActiveTaskIds: ["task-running"], + }, + }, + queues: [ + { + id: "alpha", + name: "Alpha", + total: 1, + counts: { running: 1, queued: 0 }, + unreadTerminal: 0, + activeTaskId: "task-running", + runnableTaskId: null, + updatedAt: "2026-05-20T00:00:00.000Z", + }, + { + id: "beta", + name: "Beta", + total: 2, + counts: { running: 0, queued: 2 }, + unreadTerminal: 0, + activeTaskId: null, + runnableTaskId: "task-queued", + updatedAt: "2026-05-20T00:01:00.000Z", + }, + { + id: "gamma", + name: "Gamma", + total: 1, + counts: { succeeded: 1 }, + unreadTerminal: 1, + activeTaskId: null, + runnableTaskId: null, + updatedAt: "2026-05-20T00:02:00.000Z", + }, + ], + }, + }; +} + +function assertQueuesShape(label: string, result: unknown, expectedView: string): void { + const data = asRecord(result); + const queues = asRecord(data.queues); + assertCondition(queues.view === expectedView, `${label} view mismatch`, queues); + const items = asArray(queues.items); + assertCondition(items.length === 3, `${label} must expose queue rows at data.queues.items[]`, queues); + const first = asRecord(items[0]); + assertCondition(first.id === "alpha", `${label} first item id mismatch`, first); + const firstCounts = asRecord(first.counts); + assertCondition(firstCounts.running === 1, `${label} item counts should be preserved`, first); + const counts = asRecord(queues.counts); + assertCondition(counts.running === 1 && counts.queued === 2, `${label} global counts should be preserved`, counts); + const diagnostics = asRecord(queues.executionDiagnostics); + assertCondition(diagnostics.state === "split-brain", `${label} executionDiagnostics should be preserved`, diagnostics); + assertCondition(diagnostics.effectiveLiveness === "live", `${label} diagnostics should retain derived liveness`, diagnostics); + assertCondition(Array.isArray(queues.activeTaskIds), `${label} activeTaskIds should be present`, queues); + assertCondition(Array.isArray(queues.queuedTaskIds), `${label} queuedTaskIds should be present`, queues); +} + +export function runCodeQueueQueuesShapeContract(): JsonRecord { + const fetcher = (path: string): JsonRecord => { + assertCondition(path === "/api/microservices/code-queue/proxy/api/queues", "codex queues should use stable proxy path", { path }); + return fixtureResponse(); + }; + + 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); + + const full = codexQueuesQueryForTest(["--full"], fetcher); + assertQueuesShape("full", full, "full"); + const fullQueues = asRecord(asRecord(full).queues); + assertCondition(!Array.isArray(fullQueues), "full queues payload must be an object, not the deprecated array shape", fullQueues); + assertCondition(fullQueues.bounded === false, "full without --limit should preserve complete queue listing semantics", fullQueues); + const deprecatedFullArray = asArray(fullQueues.deprecatedFullArray); + assertCondition(deprecatedFullArray.length === 3, "full should expose deprecated array only under a compatibility field", fullQueues); + const compatibility = asRecord(fullQueues.compatibility); + assertCondition(compatibility.stablePath === "data.queues.items[]", "compatibility metadata should document stable path", compatibility); + assertCondition(compatibility.deprecated === true, "compatibility metadata should mark old array path deprecated", compatibility); + + const limitedFull = codexQueuesQueryForTest(["--full", "--limit", "2"], fetcher); + const limitedFullQueues = asRecord(asRecord(limitedFull).queues); + assertCondition(limitedFullQueues.bounded === true, "full with explicit --limit should be bounded", limitedFullQueues); + assertCondition(asArray(limitedFullQueues.items).length === 2, "full with explicit --limit should limit data.queues.items[]", limitedFullQueues); + + return { + ok: true, + checks: [ + "summary data.queues.items[] shape", + "summary queue metadata", + "full data.queues.items[] shape", + "full queue metadata", + "deprecated full array compatibility field", + "full explicit limit remains bounded", + ], + }; +} + +if (import.meta.main) { + process.stdout.write(`${JSON.stringify(runCodeQueueQueuesShapeContract(), null, 2)}\n`); +} diff --git a/scripts/src/code-queue.ts b/scripts/src/code-queue.ts index f93fff73..a54a8caa 100644 --- a/scripts/src/code-queue.ts +++ b/scripts/src/code-queue.ts @@ -144,6 +144,7 @@ interface CodexTasksDegraded { interface CodexQueuesOptions { full: boolean; limit: number; + limitExplicit: boolean; } type CodexRequestInit = { method?: string; body?: unknown }; @@ -989,6 +990,7 @@ function parseQueuesOptions(args: string[]): CodexQueuesOptions { return { full: hasFlag(args, "--full") || hasFlag(args, "--all"), limit: positiveIntegerOption(args, ["--limit"], defaultTasksLimit, maxTasksLimit), + limitExplicit: args.includes("--limit"), }; } @@ -1675,13 +1677,14 @@ function compactQueuesResponse(body: Record, options: CodexQueu const runnableQueues = queues.filter((row) => row.runnableTaskId !== null && row.runnableTaskId !== undefined); const activeQueues = queues.filter((row) => typeof row.id === "string" && activeIds.includes(row.id)); const selected = options.full ? queues : Array.from(new Map([...activeQueues, ...unreadQueues, ...runnableQueues, ...nonemptyQueues].map((row) => [String(row.id), row])).values()); - const visible = selected.slice(0, options.limit); + const limitApplied = !options.full || options.limitExplicit; + const visible = limitApplied ? selected.slice(0, options.limit) : selected; const diagnostics = compactExecutionDiagnostics(queue.executionDiagnostics); return { upstream, queues: { view: options.full ? "full" : "summary", - bounded: true, + bounded: limitApplied, count: selected.length, returned: visible.length, hasMore: selected.length > visible.length, @@ -1700,6 +1703,17 @@ function compactQueuesResponse(body: Record, options: CodexQueu unreadTerminal: queue.unreadTerminal ?? 0, executionDiagnostics: diagnostics, items: visible, + ...(options.full + ? { + deprecatedFullArray: asArray(body.queues), + compatibility: { + deprecated: true, + deprecatedPath: "data.queues.deprecatedFullArray[]", + stablePath: "data.queues.items[]", + message: "Use data.queues.items[] for both codex queues and codex queues --full.", + }, + } + : {}), commands: { refresh: `bun scripts/cli.ts codex queues${options.limit === defaultTasksLimit ? "" : ` --limit ${options.limit}`}`, full: `bun scripts/cli.ts codex queues --full${options.limit === defaultTasksLimit ? "" : ` --limit ${options.limit}`}`, @@ -1711,13 +1725,16 @@ function compactQueuesResponse(body: Record, options: CodexQueu }; } -function codeQueues(optionArgs: string[] = []): unknown { +export function codexQueuesQueryForTest(optionArgs: string[], fetcher: CodexResponseFetcher): unknown { const options = parseQueuesOptions(optionArgs); - const response = unwrapCodexResponse(coreInternalFetch(codeQueueProxyPath("/api/queues"))); - if (options.full) return { upstream: response.upstream, queues: response.body.queues ?? [], queue: compactQueueMutationSummary(response.body.queue), commands: { summary: "bun scripts/cli.ts codex queues" } }; + const response = unwrapCodexResponse(fetcher(codeQueueProxyPath("/api/queues"))); return compactQueuesResponse(response.body, options, response.upstream); } +function codeQueues(optionArgs: string[] = []): unknown { + return codexQueuesQueryForTest(optionArgs, coreInternalFetch); +} + function codexCreateQueue(queueId: string): unknown { return unwrapCodexResponse(coreInternalFetch(codeQueueProxyPath("/api/queues"), { method: "POST", body: { queueId } })); }