import test from "node:test"; import assert from "node:assert/strict"; import os from "node:os"; import path from "node:path"; import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import { NextRequest } from "next/server"; let runtimeRoot = ""; let POST: (typeof import("../src/app/api/v1/projects/[projectId]/messages/route"))["POST"]; let saveAiAccount: (typeof import("../src/lib/boss-data"))["saveAiAccount"]; let updateProjectAgentControls: (typeof import("../src/lib/boss-data"))["updateProjectAgentControls"]; let readState: (typeof import("../src/lib/boss-data"))["readState"]; let createAuthSession: (typeof import("../src/lib/boss-data"))["createAuthSession"]; let AUTH_SESSION_COOKIE = ""; async function setup() { if (runtimeRoot) { return; } runtimeRoot = await mkdtemp(path.join(os.tmpdir(), "boss-master-agent-message-queue-")); process.env.BOSS_RUNTIME_ROOT = runtimeRoot; process.env.BOSS_STATE_FILE = path.join(runtimeRoot, "boss-state.json"); const [messageRoute, data, auth] = await Promise.all([ import("../src/app/api/v1/projects/[projectId]/messages/route.ts"), import("../src/lib/boss-data.ts"), import("../src/lib/boss-auth.ts"), ]); POST = messageRoute.POST; saveAiAccount = data.saveAiAccount; updateProjectAgentControls = data.updateProjectAgentControls; readState = data.readState; createAuthSession = data.createAuthSession; AUTH_SESSION_COOKIE = auth.AUTH_SESSION_COOKIE; } async function createAuthedRequest(projectId: string, body: unknown) { const session = await createAuthSession({ account: "17600003315", role: "highest_admin", displayName: "Boss 超级管理员", loginMethod: "password", }); return new NextRequest(`http://127.0.0.1:3000/api/v1/projects/${projectId}/messages`, { method: "POST", headers: { "content-type": "application/json", cookie: `${AUTH_SESSION_COOKIE}=${session.sessionToken}`, }, body: JSON.stringify(body), }); } async function waitFor(predicate: () => Promise, timeoutMs = 5_000) { const startedAt = Date.now(); while (Date.now() - startedAt < timeoutMs) { if (await predicate()) { return; } await new Promise((resolve) => setTimeout(resolve, 50)); } throw new Error("waitFor timed out"); } test.after(async () => { if (runtimeRoot) { await rm(runtimeRoot, { recursive: true, force: true }); } }); test.beforeEach(async () => { await setup(); await rm(runtimeRoot, { recursive: true, force: true }); await mkdir(runtimeRoot, { recursive: true }); }); test("POST /api/v1/projects/master-agent/messages 快速返回队列态并在异步实际回复时继承当前会话覆盖", async () => { await saveAiAccount({ accountId: "openai-master-agent-queue", label: "API 容灾", role: "api_fallback", provider: "openai_api", displayName: "OpenAI API 队列测试", model: "gpt-5.4", apiKey: "sk-test-openai-queue", enabled: true, setActive: true, loginStatusNote: "用于 master-agent 队列测试。", }); await updateProjectAgentControls("master-agent", { modelOverride: "gpt-4.1-mini", reasoningEffortOverride: "high", }); const fetchCalls: Array<{ url: string; body: unknown }> = []; const originalFetch = globalThis.fetch; globalThis.fetch = (async (input, init) => { const body = typeof init?.body === "string" ? JSON.parse(init.body) : init?.body ?? null; fetchCalls.push({ url: String(input), body }); return new Response(JSON.stringify({ output_text: "已切到异步队列回复。" }), { status: 200, headers: { "content-type": "application/json", "x-request-id": "req-master-agent-queue", }, }); }) as typeof fetch; try { const response = await POST( await createAuthedRequest("master-agent", { body: "请同步 master-agent 当前阻塞点", }), { params: Promise.resolve({ projectId: "master-agent" }) }, ); assert.equal(response.status, 200); const payload = (await response.json()) as { ok: boolean; task?: { taskId: string; taskType: string; status: string } | null; masterReplyState?: "queued" | "running" | "completed"; masterReply?: { accountId?: string } | null; }; assert.equal(payload.ok, true); assert.equal(payload.masterReply?.accountId, "openai-master-agent-queue"); assert.equal(payload.masterReplyState, "queued"); assert.ok(payload.task, "expected master-agent message to return a task envelope"); assert.equal(payload.task?.taskType, "conversation_reply"); assert.equal(payload.task?.status, "queued"); assert.ok(payload.task?.taskId, "expected a stable taskId in the response"); await waitFor(async () => { const state = await readState(); const task = state.masterAgentTasks.find((item) => item.taskId === payload.task?.taskId); return task?.status === "completed"; }); const nextState = await readState(); const task = nextState.masterAgentTasks.find((item) => item.taskId === payload.task?.taskId); assert.ok(task, "expected the queued task to remain in state"); assert.equal(task?.status, "completed"); assert.equal(task?.replyBody, "已切到异步队列回复。"); const masterProject = nextState.projects.find((project) => project.id === "master-agent"); const mirroredReply = masterProject?.messages.at(-1); assert.ok(mirroredReply, "expected the async reply to be written back to the master-agent ledger"); assert.match(mirroredReply?.body ?? "", /已切到异步队列回复/); assert.equal(fetchCalls.length, 1); assert.equal(fetchCalls[0]?.url, "https://api.openai.com/v1/responses"); const requestBody = fetchCalls[0]?.body as { model?: string; reasoning?: { effort?: string }; }; assert.equal(requestBody?.model, "gpt-4.1-mini"); assert.equal(requestBody?.reasoning?.effort, "high"); } finally { globalThis.fetch = originalFetch; } }); test("master-agent enqueue 在主节点离线时会自动切到 OpenAI 后台队列而不是挂到本机设备队列", async () => { await saveAiAccount({ accountId: "master-codex-primary-offline", label: "主 GPT", role: "primary", provider: "master_codex_node", displayName: "离线 Master Codex Node", nodeId: "offline-node", nodeLabel: "离线节点", model: "gpt-5.4", enabled: true, setActive: true, loginStatusNote: "离线主节点", }); await saveAiAccount({ accountId: "openai-backup-queue", label: "备用 GPT", role: "backup", provider: "openai_api", displayName: "OpenAI 备用账号", accountIdentifier: "sk-queue-demo", model: "gpt-5.4", apiKey: "sk-queue-demo", enabled: true, setActive: false, loginStatusNote: "备用 API 账号", }); const originalFetch = globalThis.fetch; globalThis.fetch = (async () => new Response(JSON.stringify({ output_text: "离线主节点已切到 API 后台队列。" }), { status: 200, headers: { "content-type": "application/json", "x-request-id": "req-master-agent-offline-fallback-queue", }, })) as typeof fetch; try { const response = await POST( await createAuthedRequest("master-agent", { body: "请走备用 API 队列", }), { params: Promise.resolve({ projectId: "master-agent" }) }, ); assert.equal(response.status, 200); const payload = (await response.json()) as { ok: boolean; task?: { taskId: string; taskType: string; status: string } | null; masterReplyState?: "queued" | "running" | "completed"; masterReply?: { accountId?: string } | null; }; assert.equal(payload.ok, true); assert.equal(payload.masterReply?.accountId, "openai-backup-queue"); assert.equal(payload.masterReplyState, "queued"); assert.equal(payload.task?.taskType, "conversation_reply"); await waitFor(async () => { const state = await readState(); const task = state.masterAgentTasks.find((item) => item.taskId === payload.task?.taskId); return task?.status === "completed"; }); const nextState = await readState(); const task = nextState.masterAgentTasks.find((item) => item.taskId === payload.task?.taskId); assert.equal(task?.deviceId, "master-agent-openai"); assert.equal(task?.status, "completed"); assert.equal(task?.accountId, "openai-backup-queue"); } finally { globalThis.fetch = originalFetch; } }); test("master-agent enqueue 在显式选择 claw-runtime 时会通过 Claw 异步回写回复", async () => { const clawDir = await mkdtemp(path.join(os.tmpdir(), "boss-claw-queue-")); const clawScriptPath = path.join(clawDir, "claw-runtime.mjs"); await writeFile( clawScriptPath, ` let stdin = ""; process.stdin.setEncoding("utf8"); for await (const chunk of process.stdin) { stdin += chunk; } const payload = JSON.parse(stdin); process.stdout.write(JSON.stringify({ status: "completed", output: "Claw 已接管当前主 Agent 会话:" + payload.body })); `, "utf8", ); const previousEnv = { BOSS_CLAW_ENABLED: process.env.BOSS_CLAW_ENABLED, BOSS_CLAW_COMMAND: process.env.BOSS_CLAW_COMMAND, BOSS_CLAW_ARGS: process.env.BOSS_CLAW_ARGS, BOSS_CLAW_TIMEOUT_MS: process.env.BOSS_CLAW_TIMEOUT_MS, }; process.env.BOSS_CLAW_ENABLED = "true"; process.env.BOSS_CLAW_COMMAND = process.execPath; process.env.BOSS_CLAW_ARGS = clawScriptPath; process.env.BOSS_CLAW_TIMEOUT_MS = "1000"; await saveAiAccount({ accountId: "master-codex-primary-claw", label: "主 GPT", role: "primary", provider: "master_codex_node", displayName: "Mac 上的 Master Codex Node", nodeId: "local-codex-node", nodeLabel: "本机 Codex", model: "gpt-5.4", enabled: true, setActive: true, loginStatusNote: "用于 Claw backend 队列测试。", }); await updateProjectAgentControls("master-agent", { backendOverride: "claw-runtime", }); try { const response = await POST( await createAuthedRequest("master-agent", { body: "请走 Claw runtime", }), { params: Promise.resolve({ projectId: "master-agent" }) }, ); assert.equal(response.status, 200); const payload = (await response.json()) as { ok: boolean; task?: { taskId: string; status: string } | null; masterReply?: { accountId?: string } | null; masterReplyState?: string | null; }; assert.equal(payload.ok, true); assert.equal(payload.masterReply?.accountId, "claw-runtime"); assert.equal(payload.masterReplyState, "queued"); assert.ok(payload.task?.taskId); await waitFor(async () => { const state = await readState(); const task = state.masterAgentTasks.find((item) => item.taskId === payload.task?.taskId); return task?.status === "completed"; }); const nextState = await readState(); const task = nextState.masterAgentTasks.find((item) => item.taskId === payload.task?.taskId); assert.equal(task?.status, "completed"); assert.equal(task?.replyBody, "Claw 已接管当前主 Agent 会话:请走 Claw runtime"); const masterProject = nextState.projects.find((project) => project.id === "master-agent"); const mirroredReply = masterProject?.messages.at(-1); assert.match(mirroredReply?.body ?? "", /Claw 已接管当前主 Agent 会话/); } finally { process.env.BOSS_CLAW_ENABLED = previousEnv.BOSS_CLAW_ENABLED; process.env.BOSS_CLAW_COMMAND = previousEnv.BOSS_CLAW_COMMAND; process.env.BOSS_CLAW_ARGS = previousEnv.BOSS_CLAW_ARGS; process.env.BOSS_CLAW_TIMEOUT_MS = previousEnv.BOSS_CLAW_TIMEOUT_MS; await rm(clawDir, { recursive: true, force: true }); } }); test("master-agent enqueue 在首选主节点离线时会回退到可用的备用主节点并返回实际账号", async () => { await saveAiAccount({ accountId: "master-codex-primary-offline", label: "主 GPT", role: "primary", provider: "master_codex_node", displayName: "离线 Master Codex Node", nodeId: "offline-node", nodeLabel: "离线节点", model: "gpt-5.4", enabled: true, setActive: true, loginStatusNote: "离线主节点", }); await saveAiAccount({ accountId: "master-codex-backup-online", label: "备用主节点", role: "backup", provider: "master_codex_node", displayName: "在线备用 Master Codex Node", nodeId: "mac-studio", nodeLabel: "Mac Studio", model: "gpt-5.4", enabled: true, setActive: false, loginStatusNote: "在线备用主节点", }); const response = await POST( await createAuthedRequest("master-agent", { body: "请走备用主节点队列", }), { params: Promise.resolve({ projectId: "master-agent" }) }, ); assert.equal(response.status, 200); const payload = (await response.json()) as { ok: boolean; task?: { taskId: string; taskType: string; status: string } | null; masterReplyState?: "queued" | "running" | "completed"; masterReply?: { accountId?: string } | null; }; assert.equal(payload.ok, true); assert.equal(payload.masterReply?.accountId, "master-codex-backup-online"); assert.equal(payload.masterReplyState, "queued"); assert.equal(payload.task?.taskType, "conversation_reply"); assert.equal(payload.task?.status, "queued"); const state = await readState(); const task = state.masterAgentTasks.find((item) => item.taskId === payload.task?.taskId); assert.ok(task, "expected queued master-agent task"); assert.equal(task?.accountId, "master-codex-backup-online"); assert.equal(task?.deviceId, "mac-studio"); }); test("master-agent enqueue 会在首个 API 候选失败后切到下一条备用链并重写任务账号", async () => { await saveAiAccount({ accountId: "openai-primary-queue", label: "OpenAI 主控", role: "primary", provider: "openai_api", displayName: "OpenAI 主账号", model: "gpt-5.4", apiKey: "sk-openai-primary-queue", enabled: true, setActive: true, loginStatusNote: "OpenAI 主控", }); await saveAiAccount({ accountId: "aliyun-qwen-backup-queue", label: "阿里备用", role: "backup", provider: "aliyun_qwen_api", displayName: "阿里百炼备用账号", model: "qwen3.5-plus", apiKey: "sk-aliyun-backup-queue", enabled: true, setActive: false, loginStatusNote: "阿里备用账号", }); const fetchCalls: string[] = []; const originalFetch = globalThis.fetch; globalThis.fetch = (async (input) => { fetchCalls.push(String(input)); if (typeof input === "string" && input === "https://api.openai.com/v1/responses") { return new Response(JSON.stringify({ error: { message: "openai queue failure" } }), { status: 500, headers: { "content-type": "application/json" }, }); } if (typeof input === "string" && input === "https://dashscope.aliyuncs.com/compatible-mode/v1/responses") { return new Response(JSON.stringify({ output_text: "后台队列已切到阿里备用。" }), { status: 200, headers: { "content-type": "application/json", "x-request-id": "req-master-agent-queue-fallback-chain", }, }); } throw new Error(`unexpected fetch: ${String(input)}`); }) as typeof fetch; try { const response = await POST( await createAuthedRequest("master-agent", { body: "请让后台队列自动切备用链", }), { params: Promise.resolve({ projectId: "master-agent" }) }, ); assert.equal(response.status, 200); const payload = (await response.json()) as { ok: boolean; task?: { taskId: string; taskType: string; status: string } | null; masterReplyState?: "queued" | "running" | "completed"; masterReply?: { accountId?: string } | null; }; assert.equal(payload.ok, true); assert.equal(payload.masterReply?.accountId, "openai-primary-queue"); assert.equal(payload.masterReplyState, "queued"); await waitFor(async () => { const state = await readState(); const task = state.masterAgentTasks.find((item) => item.taskId === payload.task?.taskId); return task?.status === "completed"; }); const state = await readState(); const task = state.masterAgentTasks.find((item) => item.taskId === payload.task?.taskId); assert.ok(task, "expected queued task to remain in state"); assert.equal(task?.status, "completed"); assert.equal(task?.accountId, "aliyun-qwen-backup-queue"); assert.equal(task?.deviceId, "master-agent-aliyun-qwen"); const aliyunAccount = state.aiAccounts.find((item) => item.accountId === "aliyun-qwen-backup-queue"); assert.equal(aliyunAccount?.isActive, true); assert.equal(fetchCalls[0], "https://api.openai.com/v1/responses"); assert.equal(fetchCalls[1], "https://dashscope.aliyuncs.com/compatible-mode/v1/responses"); } finally { globalThis.fetch = originalFetch; } });