fix(canvas): 本地 Codex 会话 - 修复新会话首次发送失败 (#24)

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"

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<string, unknown>;
 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<unknown> = 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";

canvas-agent/src/codex-thread.ts

@@ -0,0 +1,58 @@
+import path from "node:path";
+
+type Json = Record<string, unknown>;
+
+export type CodexThreadClient = {
+    startThread: (cwd?: string) => Promise<unknown>;
+    resumeThread: (threadId: string, cwd?: string) => Promise<unknown>;
+    readThread: (threadId: string, includeTurns?: boolean) => Promise<unknown>;
+};
+
+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;
+}

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) => {

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<CodexThreadClient> = {}) {
+    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 };
+}