fix: unify codex queues json shape

2026-05-20 20:00:39 +00:00
parent 949540235c
commit c6a27e6c2b
4 changed files with 164 additions and 7 deletions
@@ -50,7 +50,7 @@ CLI 可以从 `master` 快速演进，但必须兼容 `deploy.json` 固定的 CI
 - `codex steer <taskId> [prompt|--prompt-file path|--prompt-stdin] [--dry-run]` 通过 Code Queue 私有代理向正在运行的 task 注入纠偏提示，正式替代底层 `microservice proxy code-queue /api/tasks/<taskId>/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/<taskId>/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 <taskId>` 通过 Code Queue 私有代理请求中断；running/judging 任务会请求 D601 当前 agent run 停止，queued/retry_wait 任务的取消也必须保持与 WebUI 相同代理路径，返回有界 task 摘要和后续查询命令。任何需要接触 active run 的动作仍属于 D601 执行面。
- Code Queue 多队列 lane 由 `codex` 命令命名空间管理：`queues [--full|--all]` 列表、`queue create <queueId>` 创建、`queue merge <sourceQueueId> --into <targetQueueId>` 合并、`move <taskId> --queue <queueId>` 迁移；这些队列管理入口默认由主 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 <queueId>` 创建、`queue merge <sourceQueueId> --into <targetQueueId>` 合并、`move <taskId> --queue <queueId>` 迁移；这些队列管理入口默认由主 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 <jobId|latest> [--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` 的正式验收步骤。
@@ -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 <taskId>`：默认只查看原始 prompt、最终 response、最后错误和 drill-down 命令，这是完成未读任务审阅的第一步。
@@ -0,0 +1,140 @@
+import { codexQueuesQueryForTest } from "./src/code-queue";
+
+type JsonRecord = Record<string, unknown>;
+
+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`);
+}
@@ -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<string, unknown>, 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<string, unknown>, 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<string, unknown>, 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 } }));
 }