From e8c6b5a2d977c96a539923df6e68f37c509b0392 Mon Sep 17 00:00:00 2001 From: Cyrus Auyeung Date: Thu, 23 Jul 2026 07:04:17 +0800 Subject: [PATCH] =?UTF-8?q?fix(canvas):=20=E6=9C=AC=E5=9C=B0=20Codex=20?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D=20-=20=E4=BF=AE=E5=A4=8D=E6=96=B0=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E9=A6=96=E6=AC=A1=E5=8F=91=E9=80=81=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=20(#24)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- canvas-agent/package.json | 1 + canvas-agent/src/agents.ts | 33 +++------- canvas-agent/src/codex-thread.ts | 58 +++++++++++++++++ canvas-agent/src/http-server.ts | 6 +- canvas-agent/test/codex-thread.test.ts | 86 ++++++++++++++++++++++++++ 5 files changed, 157 insertions(+), 27 deletions(-) create mode 100644 canvas-agent/src/codex-thread.ts create mode 100644 canvas-agent/test/codex-thread.test.ts diff --git a/canvas-agent/package.json b/canvas-agent/package.json index 450f6b874..513c462d7 100644 --- a/canvas-agent/package.json +++ b/canvas-agent/package.json @@ -13,6 +13,7 @@ ], "scripts": { "dev": "tsx src/index.ts", + "test": "tsx --test test/*.test.ts", "build": "tsc -p tsconfig.json", "start": "node dist/index.js", "prepack": "npm run build" diff --git a/canvas-agent/src/agents.ts b/canvas-agent/src/agents.ts index be0e18fc0..eba908112 100644 --- a/canvas-agent/src/agents.ts +++ b/canvas-agent/src/agents.ts @@ -6,12 +6,13 @@ import { createRequire } from "node:module"; import { fileURLToPath } from "node:url"; import { AGENT_PROMPT, VERSION } from "./config.js"; +import { assertCodexThreadWorkspace, codexThreadInWorkspace, resolveCodexThread } from "./codex-thread.js"; import type { AgentAttachment, AgentEmit } from "./types.js"; type Json = Record; type AgentEvent = Json & { type: string; usage?: unknown }; type PendingRequest = { resolve: (value: unknown) => void; reject: (error: Error) => void }; -type CodexRunOptions = { threadId?: string; cwd?: string }; +type CodexRunOptions = { threadId?: string; cwd?: string; onThreadId?: (threadId: string) => void }; type AgentHistoryMessage = { id: string; role: "user" | "assistant" | "tool" | "error"; title?: string; text: string; detail?: unknown; streamId?: string }; let codexQueue: Promise = Promise.resolve(); @@ -36,6 +37,7 @@ async function runCodexTurnNow(prompt: string, emit: AgentEmit, attachments: Age files = await writeAttachmentFiles(attachments); codexApp ||= await CodexAppClient.start(emit); const threadId = await ensureCodexThread(codexApp, options); + if (threadId !== options.threadId) options.onThreadId?.(threadId); await codexApp.startTurn(threadId, prompt, files); } catch (error) { emit("agent_error", { message: errorMessage(error) }); @@ -55,7 +57,7 @@ export async function resumeCodexThread(emit: AgentEmit, threadId: string, cwd?: codexApp ||= await CodexAppClient.start(emit); await loadCodexThread(emit, threadId, cwd, false); const thread = await codexApp.resumeThread(threadId, cwd); - assertThreadWorkspace(thread, cwd); + assertCodexThreadWorkspace(thread, cwd); codexThreadId = String(field(thread, "id") || threadId); return { thread, messages: threadMessages(thread) }; } @@ -70,7 +72,7 @@ export async function listCodexThreads(emit: AgentEmit, options: { cwd: string; cwd: options.cwd, ...(options.searchTerm ? { searchTerm: options.searchTerm } : {}), }); - const data = Array.isArray(field(result, "data")) ? (field(result, "data") as unknown[]).map(summarizeCodexThread).filter((thread) => threadInWorkspace(thread, options.cwd)) : []; + const data = Array.isArray(field(result, "data")) ? (field(result, "data") as unknown[]).map(summarizeCodexThread).filter((thread) => codexThreadInWorkspace(thread, options.cwd)) : []; return { data, nextCursor: field(result, "nextCursor") || null, backwardsCursor: field(result, "backwardsCursor") || null }; } @@ -97,18 +99,7 @@ export function runClaudeTurn(prompt: string, emit: AgentEmit) { } async function ensureCodexThread(app: CodexAppClient, options: CodexRunOptions) { - if (options.threadId) { - const result = await app.readThread(options.threadId, false); - assertThreadWorkspace(field(result, "thread") || {}, options.cwd); - const thread = await app.resumeThread(options.threadId, options.cwd); - assertThreadWorkspace(thread, options.cwd); - codexThreadId = String(field(thread, "id") || options.threadId); - return codexThreadId; - } - if (!codexThreadId) { - const thread = await app.startThread(options.cwd); - codexThreadId = String(field(thread, "id") || ""); - } + codexThreadId = await resolveCodexThread(app, options, codexThreadId); return codexThreadId; } @@ -301,20 +292,10 @@ async function loadCodexThread(emit: AgentEmit, threadId: string, cwd: string | codexApp ||= await CodexAppClient.start(emit); const result = await codexApp.readThread(threadId, includeTurns); const thread = field(result, "thread") || {}; - assertThreadWorkspace(thread, cwd); + assertCodexThreadWorkspace(thread, cwd); return thread; } -function assertThreadWorkspace(thread: unknown, cwd?: string) { - if (!cwd || threadInWorkspace(thread, cwd)) return; - throw new Error("该 Codex 会话不属于当前画布工作空间"); -} - -function threadInWorkspace(thread: unknown, cwd: string) { - const threadCwd = String(field(thread, "cwd") || ""); - return Boolean(threadCwd && path.resolve(threadCwd) === path.resolve(cwd)); -} - function normalizeItem(item: unknown) { const value = item && typeof item === "object" ? { ...(item as Json) } : {}; if (value.type === "agentMessage") value.type = "agent_message"; diff --git a/canvas-agent/src/codex-thread.ts b/canvas-agent/src/codex-thread.ts new file mode 100644 index 000000000..9753d8f14 --- /dev/null +++ b/canvas-agent/src/codex-thread.ts @@ -0,0 +1,58 @@ +import path from "node:path"; + +type Json = Record; + +export type CodexThreadClient = { + startThread: (cwd?: string) => Promise; + resumeThread: (threadId: string, cwd?: string) => Promise; + readThread: (threadId: string, includeTurns?: boolean) => Promise; +}; + +export type CodexThreadOptions = { threadId?: string; cwd?: string }; + +export async function resolveCodexThread(client: CodexThreadClient, options: CodexThreadOptions, currentThreadId = "") { + const requestedThreadId = String(options.threadId || ""); + if (requestedThreadId) { + // thread/start 会先返回尚未落盘的 ID；同一进程内应直接开始首轮，不能立即恢复它。 + if (requestedThreadId === currentThreadId) return requestedThreadId; + try { + const result = await client.readThread(requestedThreadId, false); + assertCodexThreadWorkspace(field(result, "thread") || {}, options.cwd); + const thread = await client.resumeThread(requestedThreadId, options.cwd); + assertCodexThreadWorkspace(thread, options.cwd); + return String(field(thread, "id") || requestedThreadId); + } catch (error) { + // Agent 可能在新会话首轮前重启；这种空会话没有用户内容，可以安全替换。 + if (!isUnmaterializedCodexThreadError(error)) throw error; + return await startCodexThread(client, options.cwd); + } + } + if (currentThreadId) return currentThreadId; + return await startCodexThread(client, options.cwd); +} + +export function assertCodexThreadWorkspace(thread: unknown, cwd?: string) { + if (!cwd || codexThreadInWorkspace(thread, cwd)) return; + throw new Error("该 Codex 会话不属于当前画布工作空间"); +} + +export function codexThreadInWorkspace(thread: unknown, cwd: string) { + const threadCwd = String(field(thread, "cwd") || ""); + return Boolean(threadCwd && path.resolve(threadCwd) === path.resolve(cwd)); +} + +async function startCodexThread(client: CodexThreadClient, cwd?: string) { + const thread = await client.startThread(cwd); + const threadId = String(field(thread, "id") || ""); + if (!threadId) throw new Error("Codex app-server 没有返回 thread id"); + return threadId; +} + +function isUnmaterializedCodexThreadError(error: unknown) { + const message = error instanceof Error ? error.message : String(error); + return /no rollout found for thread id|not materialized yet/i.test(message); +} + +function field(value: unknown, key: string) { + return value && typeof value === "object" ? (value as Json)[key] : undefined; +} diff --git a/canvas-agent/src/http-server.ts b/canvas-agent/src/http-server.ts index a173c3143..d0e6ca768 100644 --- a/canvas-agent/src/http-server.ts +++ b/canvas-agent/src/http-server.ts @@ -85,7 +85,11 @@ export function startHttpServer() { await verifyCodexThreadWorkspace(emit, threadId, workspace.workspacePath); updateCanvasWorkspace(config, workspace.canvasId, { activeThreadId: threadId }); } - void runCodexTurn(withAgentPrompt(String(req.body?.prompt || "")), emit, attachments, { threadId, cwd: workspace.workspacePath }); + void runCodexTurn(withAgentPrompt(String(req.body?.prompt || "")), emit, attachments, { + threadId, + cwd: workspace.workspacePath, + onThreadId: (nextThreadId) => updateCanvasWorkspace(config, workspace.canvasId, { activeThreadId: nextThreadId }), + }); res.json({ ok: true, threadId }); })); app.post("/agent/claude/turn", (req, res) => { diff --git a/canvas-agent/test/codex-thread.test.ts b/canvas-agent/test/codex-thread.test.ts new file mode 100644 index 000000000..bc2fb8d3b --- /dev/null +++ b/canvas-agent/test/codex-thread.test.ts @@ -0,0 +1,86 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { resolveCodexThread, type CodexThreadClient } from "../src/codex-thread.js"; + +const cwd = "C:\\canvas-workspace"; + +test("当前进程刚创建的会话直接开始首轮", async () => { + const client = fakeClient(); + + const threadId = await resolveCodexThread(client.value, { threadId: "thread-new", cwd }, "thread-new"); + + assert.equal(threadId, "thread-new"); + assert.deepEqual(client.calls, []); +}); + +test("已有会话仍校验工作空间并恢复", async () => { + const client = fakeClient({ + readThread: async () => ({ thread: { id: "thread-old", cwd } }), + resumeThread: async () => ({ id: "thread-old", cwd }), + }); + + const threadId = await resolveCodexThread(client.value, { threadId: "thread-old", cwd }); + + assert.equal(threadId, "thread-old"); + assert.deepEqual(client.calls, ["read:thread-old", "resume:thread-old"]); +}); + +test("未落盘的空会话自动替换", async () => { + const client = fakeClient({ + readThread: async () => ({ thread: { id: "thread-stale", cwd } }), + resumeThread: async () => { + throw new Error("no rollout found for thread id thread-stale"); + }, + startThread: async () => ({ id: "thread-replacement", cwd }), + }); + + const threadId = await resolveCodexThread(client.value, { threadId: "thread-stale", cwd }); + + assert.equal(threadId, "thread-replacement"); + assert.deepEqual(client.calls, ["read:thread-stale", "resume:thread-stale", "start"]); +}); + +test("读取阶段发现未 materialize 的空会话也会替换", async () => { + const client = fakeClient({ + readThread: async () => { + throw new Error("thread thread-stale is not materialized yet"); + }, + startThread: async () => ({ id: "thread-replacement", cwd }), + }); + + const threadId = await resolveCodexThread(client.value, { threadId: "thread-stale", cwd }); + + assert.equal(threadId, "thread-replacement"); + assert.deepEqual(client.calls, ["read:thread-stale", "start"]); +}); + +test("非空会话错误不会被静默替换", async () => { + const client = fakeClient({ + readThread: async () => { + throw new Error("permission denied"); + }, + }); + + await assert.rejects(() => resolveCodexThread(client.value, { threadId: "thread-old", cwd }), /permission denied/); + assert.deepEqual(client.calls, ["read:thread-old"]); +}); + +function fakeClient(overrides: Partial = {}) { + const calls: string[] = []; + const value: CodexThreadClient = { + readThread: async (threadId) => { + calls.push(`read:${threadId}`); + return await (overrides.readThread?.(threadId, false) || Promise.resolve({ thread: { id: threadId, cwd } })); + }, + resumeThread: async (threadId, workspace) => { + calls.push(`resume:${threadId}`); + return await (overrides.resumeThread?.(threadId, workspace) || Promise.resolve({ id: threadId, cwd: workspace })); + }, + startThread: async (workspace) => { + calls.push("start"); + return await (overrides.startThread?.(workspace) || Promise.resolve({ id: "thread-started", cwd: workspace })); + }, + }; + return { calls, value }; +}