diff --git a/canvas-agent/src/agent/codex-client.test.ts b/canvas-agent/src/agent/codex-client.test.ts index 32406d4..0bdd128 100644 --- a/canvas-agent/src/agent/codex-client.test.ts +++ b/canvas-agent/src/agent/codex-client.test.ts @@ -2,10 +2,12 @@ import assert from "node:assert/strict"; import test from "node:test"; import { CodexAppClient } from "./codex-client.js"; +import { assertDraftHasNoSensitiveValues, canvasSkillSource } from "./codex.js"; type TestClient = { currentThreadId: string; currentTurnId: string; + skillDraftActive: boolean; completedTurns: Map; plansByTurn: Map; lastUsage: unknown; @@ -55,6 +57,14 @@ test("中断请求只作用于当前运行线程", async () => { assert.ok(request); testClient.handle({ id: request.id, result: {} }); assert.equal(await interrupt, true); + + testClient.currentThreadId = "draft-thread"; + testClient.currentTurnId = "draft-turn"; + const draftInterrupt = client.interruptCurrentTurn(); + const draftRequest = writes.at(-1); + assert.deepEqual(draftRequest?.params, { threadId: "draft-thread", turnId: "draft-turn" }); + testClient.handle({ id: draftRequest?.id, result: {} }); + assert.equal(await draftInterrupt, true); }); test("Skill 列表与启用配置使用 app-server 原生协议", async () => { @@ -140,6 +150,412 @@ test("skills/changed 作为站点级事件单独广播", () => { assert.deepEqual(events, [{ type: "skills_changed", payload: {} }]); }); +test("静默 Skill 草稿 turn 只返回结构化结果，不广播也不写历史", async () => { + const writes: Array> = []; + const events: Array<{ type: string; payload: unknown }> = []; + const persistedItems: unknown[] = []; + const persistedTurns: unknown[] = []; + const child = { stdin: { write: (line: string) => (writes.push(JSON.parse(line)), true) } }; + const history = { + record: (entry: unknown) => (persistedItems.push(entry), Promise.resolve()), + recordTurn: (entry: unknown) => (persistedTurns.push(entry), Promise.resolve()), + }; + const client = Reflect.construct(CodexAppClient, [child, (type: string, payload: unknown) => events.push({ type, payload }), history]) as CodexAppClient; + const testClient = client as unknown as TestClient; + + const starting = client.startSkillDraftThread("D:\\site"); + const threadRequest = writes.find((item) => item.method === "thread/start"); + assert.ok(threadRequest); + testClient.handleNotification("thread/started", { thread: { id: "draft-thread", ephemeral: true } }); + testClient.handle({ id: threadRequest.id, result: { thread: { id: "draft-thread", ephemeral: true } } }); + await starting; + + const schema = { type: "object", properties: { name: { type: "string" } } }; + const output = JSON.stringify({ name: "product-image-flow" }); + const generating = client.generateSkillDraft("draft-thread", "提炼流程", schema); + const turnRequest = writes.find((item) => item.method === "turn/start"); + assert.deepEqual(turnRequest?.params, { + threadId: "draft-thread", + input: [{ type: "text", text: "提炼流程", text_elements: [] }], + approvalPolicy: "never", + sandboxPolicy: { type: "readOnly", networkAccess: false }, + outputSchema: schema, + }); + testClient.handle({ id: turnRequest?.id, result: { turn: { id: "draft-turn" } } }); + await new Promise((resolve) => setImmediate(resolve)); + testClient.handleNotification("turn/started", { threadId: "draft-thread", turn: { id: "draft-turn", status: "inProgress" } }); + testClient.handleNotification("item/reasoning/summaryTextDelta", { threadId: "draft-thread", turnId: "draft-turn", itemId: "reasoning-1", delta: "正在提炼" }); + testClient.handleNotification("item/completed", { threadId: "draft-thread", turnId: "draft-turn", item: { id: "reasoning-1", type: "reasoning", summary: ["正在提炼"] } }); + testClient.handleNotification("item/agentMessage/delta", { threadId: "draft-thread", turnId: "draft-turn", itemId: "assistant-1", delta: output }); + testClient.handleNotification("item/completed", { threadId: "draft-thread", turnId: "draft-turn", item: { id: "assistant-1", type: "agentMessage", text: "" } }); + testClient.handleNotification("turn/completed", { threadId: "draft-thread", turn: { id: "draft-turn", status: "completed" } }); + + assert.equal(await generating, output); + assert.deepEqual(events, []); + assert.deepEqual(persistedItems, []); + assert.deepEqual(persistedTurns, []); + + assert.equal(testClient.skillDraftActive, true); + const closing = client.closeSkillDraftThread("draft-thread"); + const unsubscribe = writes.find((item) => item.method === "thread/unsubscribe"); + assert.ok(unsubscribe); + testClient.handle({ id: unsubscribe.id, result: { status: "unsubscribed" } }); + await closing; + assert.equal(testClient.skillDraftActive, false); +}); + +test("静默线程创建期间只隐藏响应返回的线程", async () => { + const writes: Array> = []; + const events: Array<{ type: string; payload: unknown }> = []; + const child = { stdin: { write: (line: string) => (writes.push(JSON.parse(line)), true) } }; + const client = Reflect.construct(CodexAppClient, [child, (type: string, payload: unknown) => events.push({ type, payload }), emptyEventHistory]) as CodexAppClient; + const testClient = client as unknown as TestClient; + + const starting = client.startSkillDraftThread("D:\\site"); + const request = writes.find((item) => item.method === "thread/start"); + assert.ok(request); + + testClient.handleNotification("thread/started", { thread: { id: "normal-thread" } }); + testClient.handleNotification("thread/started", { thread: { id: "draft-thread", ephemeral: true } }); + assert.deepEqual(events, []); + + testClient.handle({ id: request.id, result: { thread: { id: "draft-thread", ephemeral: true } } }); + await starting; + + assert.deepEqual(events, [{ + type: "agent_event", + payload: { agent: "codex", type: "thread.started", thread_id: "normal-thread" }, + }]); + + testClient.handleNotification("thread/started", { thread: { id: "draft-thread", ephemeral: true } }); + assert.equal(events.length, 1); +}); + +test("静默 Skill 草稿不吞掉全局 Skill 变更通知", async () => { + const events: Array<{ type: string; payload: unknown }> = []; + const writes: Array> = []; + const child = { stdin: { write: (line: string) => (writes.push(JSON.parse(line)), true) } }; + const client = Reflect.construct(CodexAppClient, [child, (type: string, payload: unknown) => events.push({ type, payload }), emptyEventHistory]) as CodexAppClient; + const testClient = client as unknown as TestClient; + + const starting = client.startSkillDraftThread("D:\\site"); + const request = writes.find((item) => item.method === "thread/start"); + assert.ok(request); + testClient.handle({ id: request.id, result: { thread: { id: "draft-thread", ephemeral: true } } }); + await starting; + + testClient.handleNotification("thread/started", { thread: { id: "draft-thread", ephemeral: true } }); + testClient.handleNotification("skills/changed", {}); + + assert.deepEqual(events, [{ type: "skills_changed", payload: {} }]); +}); + +test("静默 Skill 草稿自动拒绝未携带线程 ID 的权限请求", async () => { + const events: Array<{ type: string; payload: unknown }> = []; + const writes: Array> = []; + const child = { stdin: { write: (line: string) => (writes.push(JSON.parse(line)), true) } }; + const client = Reflect.construct(CodexAppClient, [child, (type: string, payload: unknown) => events.push({ type, payload }), emptyEventHistory]) as CodexAppClient; + const testClient = client as unknown as TestClient; + + const starting = client.startSkillDraftThread("D:\\site"); + const request = writes.find((item) => item.method === "thread/start"); + assert.ok(request); + testClient.handle({ id: request.id, result: { thread: { id: "draft-thread", ephemeral: true } } }); + await starting; + testClient.handleNotification("turn/started", { threadId: "draft-thread", turn: { id: "draft-turn", status: "inProgress" } }); + + testClient.answerServerRequest({ id: 17, method: "item/permissions/requestApproval", params: { turnId: "draft-turn", permissions: { network: true } } }); + + assert.deepEqual(writes.at(-1), { id: 17, result: { permissions: {}, scope: "turn" } }); + assert.deepEqual(events, []); +}); + +test("app-server 退出时清理尚未确认的静默线程", async () => { + const events: Array<{ type: string; payload: unknown }> = []; + const writes: Array> = []; + const child = { stdin: { write: (line: string) => (writes.push(JSON.parse(line)), true) } }; + const client = Reflect.construct(CodexAppClient, [child, (type: string, payload: unknown) => events.push({ type, payload }), emptyEventHistory]) as CodexAppClient; + const testClient = client as unknown as TestClient; + + const starting = client.startSkillDraftThread("D:\\site"); + testClient.handleNotification("thread/started", { thread: { id: "draft-thread", ephemeral: true } }); + assert.equal(testClient.skillDraftActive, true); + + testClient.failAll("Codex app-server exited: 1"); + + assert.equal(testClient.skillDraftActive, false); + assert.deepEqual(events, []); + await assert.rejects(starting, /Codex app-server exited/); +}); + +test("app-server 在草稿运行中退出后不再等待线程取消订阅", async () => { + const writes: Array> = []; + const child = { stdin: { write: (line: string) => (writes.push(JSON.parse(line)), true) } }; + const client = Reflect.construct(CodexAppClient, [child, () => undefined, emptyEventHistory]) as CodexAppClient; + const testClient = client as unknown as TestClient; + + const starting = client.startSkillDraftThread("D:\\site"); + const threadRequest = writes.find((item) => item.method === "thread/start"); + assert.ok(threadRequest); + testClient.handle({ id: threadRequest.id, result: { thread: { id: "draft-thread", ephemeral: true } } }); + await starting; + + const generating = client.generateSkillDraft("draft-thread", "提炼流程", { type: "object" }); + const turnRequest = writes.find((item) => item.method === "turn/start"); + assert.ok(turnRequest); + testClient.handle({ id: turnRequest.id, result: { turn: { id: "draft-turn" } } }); + await new Promise((resolve) => setImmediate(resolve)); + testClient.failAll("Codex app-server exited: 1"); + + await assert.rejects(generating, /Codex app-server exited/); + const writeCount = writes.length; + await client.closeSkillDraftThread("draft-thread"); + assert.equal(writes.length, writeCount); + await assert.rejects(client.listModels(), /Codex app-server exited/); + assert.equal(writes.length, writeCount); +}); + +test("画布 Skill 草稿源保留流程信息并移除媒体、本地路径、凭据和临时任务字段", () => { + const source = canvasSkillSource({ + projectId: "private-project", + clientId: "private-client", + title: "商品图流程", + viewport: { x: 100, y: 200, k: 1.5 }, + nodes: [{ + id: "prompt-node", + type: "text", + title: "提示词", + position: { x: 20, y: 30 }, + width: 480, + height: 240, + metadata: { + content: "保留这段可复用流程；API Key=secret-api-key；Bearer abcdefghijklmnop", + sourceUrl: "https://example.com/reference.png", + extensionlessMediaUrl: "https://cdn.example.com/private-media/abc123", + sensitiveUrl: "https://example.com/reference.png?token=secret-value", + signedUrl: "https://example.com/reference.png?X-Amz-Signature=secret-value", + basicAuthUrl: "https://user:password@example.com/reference.png", + dataUrl: "data:image/png;base64,c2VjcmV0", + blobUrl: "blob:http://localhost/private-image", + reference: "D:\\private\\reference.png", + referenceNote: "本地文件 D:\\private\\reference.png 参考图 https://example.com/reference.png", + unixReference: "/var/lib/private/reference.png", + unixReferenceNote: "文件 /root/private.png 与 file:///opt/private.png", + batchRootId: "output-node", + batchChildIds: ["output-node", "missing-node"], + primaryImageId: "output-node", + groupId: "prompt-node", + storageKey: "canvas-private-key", + apiKey: "secret-api-key", + credentials: { token: "secret-token" }, + status: "running", + progress: 75, + errorDetails: "temporary failure", + taskId: "task-private", + createdAt: "2026-08-03", + nested: { keep: "保留", password: "secret-password", preview: "blob:http://localhost/nested" }, + }, + }, { + id: "output-node", + type: "image", + title: "结果图片", + position: { x: 600, y: 30 }, + width: 480, + height: 480, + metadata: { content: "data:image/png;base64,c2VjcmV0" }, + }], + connections: [{ id: "connection-1", fromNodeId: "prompt-node", toNodeId: "output-node" }], + selectedNodeIds: ["prompt-node"], + }); + + assert.deepEqual(source, { + title: "商品图流程", + nodes: [{ + ref: "node-1", + type: "text", + title: "提示词", + metadata: { + content: "保留这段可复用流程；[敏感凭证已移除]；[敏感凭证已移除]", + referenceNote: "本地文件 [本地路径已移除] 参考图 [外部地址已移除]", + unixReferenceNote: "文件 [本地路径已移除] 与 [本地路径已移除]", + batchRootRef: "node-2", + batchChildRefs: ["node-2"], + primaryImageRef: "node-2", + groupRef: "node-1", + nested: { keep: "保留" }, + }, + }, { + ref: "node-2", + type: "image", + title: "结果图片", + }], + connections: [{ from: "node-1", to: "node-2" }], + selectedNodeRefs: ["node-1"], + }); + assert.doesNotMatch(JSON.stringify(source), /prompt-node|output-node|connection-1|secret-value|secret-api-key|abcdefghijklmnop|D:\\\\private|\/var\/|\/root\/|\/opt\/|file:|https?:\/\//); +}); + +test("画布 Skill 草稿源保留普通中文斜杠文本和斜杠命令", () => { + const source = canvasSkillSource({ + nodes: [{ + id: "prompt-node", + type: "text", + title: "提示词", + position: { x: 0, y: 0 }, + width: 320, + height: 180, + metadata: { + content: "故事概念/剧本圣经 → 配音/音效/音乐", + prompt: "/imagine 商品图", + }, + }], + connections: [], + }); + + assert.deepEqual(source, { + nodes: [{ + ref: "node-1", + type: "text", + title: "提示词", + metadata: { + content: "故事概念/剧本圣经 → 配音/音效/音乐", + prompt: "/imagine 商品图", + }, + }], + connections: [], + }); +}); + +test("画布 Skill 草稿源匿名化任意字符串中的已知节点 ID", () => { + const imageNodeId = "8f36ab15-30f1-4c73-b169-f12c36584ddb"; + const source = canvasSkillSource({ + nodes: [{ + id: "prompt-node", + type: "text", + title: "提示词", + position: { x: 0, y: 0 }, + width: 320, + height: 180, + metadata: { composerContent: `参考 @[node:${imageNodeId}] 生成商品图` }, + }, { + id: imageNodeId, + type: "image", + title: "参考图", + position: { x: 400, y: 0 }, + width: 320, + height: 320, + metadata: {}, + }], + connections: [], + }); + + assert.equal( + ((source.nodes as Array>)[0]?.metadata as Record).composerContent, + "参考 @[node:node-2] 生成商品图", + ); + assert.doesNotMatch(JSON.stringify(source), new RegExp(imageNodeId)); +}); + +test("画布 Skill 草稿源匿名化 metadata 对象键中的已知节点 ID", () => { + const imageNodeId = "8f36ab15-30f1-4c73-b169-f12c36584ddb"; + const source = canvasSkillSource({ + nodes: [{ + id: "prompt-node", + type: "text", + title: "提示词", + position: { x: 0, y: 0 }, + width: 320, + height: 180, + metadata: { pluginState: { [`node:${imageNodeId}`]: { label: "参考图" } } }, + }, { + id: imageNodeId, + type: "image", + title: "参考图", + position: { x: 400, y: 0 }, + width: 320, + height: 320, + metadata: {}, + }], + connections: [], + }); + const metadata = (source.nodes as Array>)[0]?.metadata as Record; + + assert.deepEqual(metadata.pluginState, { "node:node-2": { label: "参考图" } }); + assert.doesNotMatch(JSON.stringify(source), new RegExp(imageNodeId)); +}); + +test("画布 Skill 草稿源为截断节点保留完整快照范围内的安全引用", () => { + const targetNodeId = "target-node-id"; + const nodes = Array.from({ length: 301 }, (_, index) => ({ + id: index === 300 ? targetNodeId : `filler-node-${index + 1}`, + type: "text" as const, + title: index === 0 ? "提示词" : `节点 ${index + 1}`, + position: { x: index * 10, y: 0 }, + width: 320, + height: 180, + metadata: index === 0 ? { composerContent: `参考 @[node:${targetNodeId}]` } : {}, + })); + const source = canvasSkillSource({ nodes, connections: [] }); + const firstMetadata = (source.nodes as Array>)[0]?.metadata as Record; + + assert.equal(firstMetadata.composerContent, "参考 @[node:node-301]"); + assert.equal(source.truncated, true); + assert.doesNotMatch(JSON.stringify(source), new RegExp(targetNodeId)); +}); + +test("画布 Skill 草稿源限制总长度并优先保留选中流程", () => { + const nodes = Array.from({ length: 320 }, (_, index) => ({ + id: `node-${index + 1}`, + type: "text" as const, + title: index === 318 ? "关联节点" : index === 319 ? "选中节点" : `普通节点 ${index + 1}`, + position: { x: index * 10, y: 0 }, + width: 320, + height: 180, + metadata: { content: "流程说明".repeat(2_000) }, + })); + const source = canvasSkillSource({ + nodes, + connections: [{ id: "private-connection", fromNodeId: "node-319", toNodeId: "node-320" }], + selectedNodeIds: ["node-320"], + }); + const sourceNodes = source.nodes as Array>; + + assert.ok(JSON.stringify(source).length <= 120_000); + assert.equal(sourceNodes[0]?.title, "选中节点"); + assert.equal(sourceNodes[1]?.title, "关联节点"); + assert.deepEqual(source.connections, [{ from: "node-2", to: "node-1" }]); + assert.deepEqual(source.selectedNodeRefs, ["node-1"]); + assert.equal(source.truncated, true); +}); + +test("Skill 草稿拒绝普通外部 URL，避免对话内容泄露媒体地址", () => { + const draft = { + name: "product-image-flow", + displayName: "Product image flow", + description: "Reusable image generation workflow", + instructions: "参考图地址：https://example.com/reference.png", + shortDescription: "Generate product images from a reusable workflow", + defaultPrompt: "$product-image-flow", + }; + + assert.throws(() => assertDraftHasNoSensitiveValues(draft, []), /外部地址/); +}); + +test("Skill 草稿允许普通动作名称，但拒绝任意本地绝对路径", () => { + const draft = { + name: "request-review", + displayName: "Review request", + description: "Use node-workflow-builder to review a reusable flow", + instructions: "Run generation-status-check, then use job-application-helper.", + shortDescription: "Review a reusable workflow", + defaultPrompt: "$request-review", + }; + + assert.doesNotThrow(() => assertDraftHasNoSensitiveValues(draft, [])); + assert.throws(() => assertDraftHasNoSensitiveValues({ ...draft, instructions: "读取 /var/lib/private.png" }, []), /本地路径/); + assert.throws(() => assertDraftHasNoSensitiveValues({ ...draft, instructions: "读取 file:///root/private.png" }, []), /本地路径/); +}); + test("turn/started 早于 turn/start 响应时保持完整事件归属", async () => { const writes: Array> = []; const events: Array<{ type: string; payload: unknown }> = []; diff --git a/canvas-agent/src/agent/codex-client.ts b/canvas-agent/src/agent/codex-client.ts index 9b95819..2b9e1af 100644 --- a/canvas-agent/src/agent/codex-client.ts +++ b/canvas-agent/src/agent/codex-client.ts @@ -12,7 +12,7 @@ import type { CodexNotificationParams, CodexPlanUpdate, CodexReasoningEffort, Co import type { AgentEmit, AgentPermissionMode } from "./types.js"; type AgentEvent = JsonRecord & { type: string; usage?: unknown }; -type PendingRequest = { resolve: (value: unknown) => void; reject: (error: Error) => void }; +type PendingRequest = { resolve: (value: unknown) => void; reject: (error: Error) => void; silent?: boolean }; type ActiveTurn = PendingRequest & { threadId: string; turnId: string; prompt: string; messageText?: string }; type ItemDeltaParams = { threadId: string; turnId: string; itemId: string; delta: string; summaryIndex?: number }; type PendingDelta = { delta: string; itemType: string; params: ItemDeltaParams; timer: ReturnType }; @@ -23,6 +23,7 @@ const canvasAgentMcp = canvasAgentMcpCommand(); const require = createRequire(import.meta.url); const STREAM_UPDATE_INTERVAL_MS = 40; const supplementalItemTypes = new Set(["agent_message", "reasoning", "plan", "mcp_tool_call", "command_execution", "file_change", "dynamic_tool_call", "collab_tool_call", "web_search", "image_view", "image_generation", "context_compaction"]); +const SKILL_DRAFT_INSTRUCTIONS = "你只负责根据已提供的对话或画布快照生成可编辑的 Codex Skill 草稿。不要调用任何工具，不要执行命令，不要读取文件，不要访问网络，不要修改任何状态。严格按 outputSchema 返回结果，并排除凭证、密钥、Token、本地路径、临时错误、调试日志和一次性结果。"; /** 表示错误已经通过 app-server 终态或进程事件通知过网页。 */ export class CodexReportedError extends Error { @@ -43,6 +44,7 @@ export class CodexAppClient { private pending = new Map(); private activeTurns = new Map(); private completedTurns = new Map(); + private completedTurnResults = new Map(); private pendingDeltas = new Map(); private startedItems = new Map(); private itemSequences = new Map(); @@ -51,7 +53,12 @@ export class CodexAppClient { private approvalRequests = new Map(); private finalizingTurns = new Map>(); private skillReloads = new Map>>(); + private silentThreadIds = new Set(); + private structuredOutputByTurn = new Map(); + private pendingSilentThreadStarts = new Set(); + private pendingThreadStartedNotifications: JsonRecord[] = []; private failing = false; + private failureMessage = ""; /** 保存 app-server 子进程和事件出口。 */ private constructor(private child: ChildProcess, private emit: AgentEmit, private eventHistory: Pick = codexEventHistory) {} @@ -69,6 +76,7 @@ export class CodexAppClient { }; child.stdout?.on("data", (chunk) => client.read(chunk.toString())); child.stderr?.on("data", (chunk) => { + if (client.skillDraftActive) return; const text = stripAnsi(chunk.toString()).replace(/^\d{4}-\d{2}-\d{2}T[\d:.]+Z\s+/, ""); logger.warn("Codex app-server stderr", { text }); emit("agent_log", { text }); @@ -97,6 +105,16 @@ export class CodexAppClient { return thread; } + /** 创建不会持久化或向网页广播的草稿线程。 */ + async startSkillDraftThread(cwd: string) { + return await this.startSilentThread("thread/start", { ...skillDraftThreadSettings(cwd), threadSource: "user" }); + } + + /** 从指定对话派生不会持久化或向网页广播的草稿线程。 */ + async forkSkillDraftThread(threadId: string, cwd: string) { + return await this.startSilentThread("thread/fork", { ...skillDraftThreadSettings(cwd), threadId, threadSource: "user" }); + } + /** 恢复已有 Codex 线程。 */ async resumeThread(threadId: string, cwd?: string, permissionMode: AgentPermissionMode = "request") { const { thread } = await this.request("thread/resume", { threadId, ...threadSettings(permissionMode), ...(cwd ? { cwd } : {}) }); @@ -119,6 +137,17 @@ export class CodexAppClient { return this.request("thread/archive", { threadId }); } + /** 释放临时草稿线程的 App Server 订阅和进程内缓存。 */ + async closeSkillDraftThread(threadId: string) { + try { + if (!this.failing) await this.request("thread/unsubscribe", { threadId }, true); + } finally { + this.silentThreadIds.delete(threadId); + const prefix = `${threadId}\0`; + [...this.structuredOutputByTurn.keys()].filter((key) => key.startsWith(prefix)).forEach((key) => this.structuredOutputByTurn.delete(key)); + } + } + /** 查询当前账号可用的 Codex 模型。 */ listModels() { return this.request("model/list", { limit: 100, includeHidden: false }); @@ -157,14 +186,14 @@ export class CodexAppClient { } /** 启动一个 Codex turn 并等待完成通知。 */ - async startTurn(threadId: string, prompt: string, images: string[], permissionMode: AgentPermissionMode, model?: string, effort?: CodexReasoningEffort, onTurn?: (turnId: string) => void, skill?: CodexSkillSelector, messageText?: string) { + async startTurn(threadId: string, prompt: string, images: string[], permissionMode: AgentPermissionMode, model?: string, effort?: CodexReasoningEffort, onTurn?: (turnId: string) => void, skill?: CodexSkillSelector, messageText?: string, outputSchema?: JsonRecord) { this.currentThreadId = threadId; this.currentTurnId = ""; this.lastUsage = null; const pendingStart: PendingTurnStart = { threadId, prompt, messageText, onTurn }; this.pendingTurnStart = pendingStart; try { - const { turn } = await this.request("turn/start", { threadId, input: codexInput(prompt, images, skill), ...turnSettings(permissionMode), ...(model ? { model } : {}), ...(effort ? { effort } : {}) }); + const { turn } = await this.request("turn/start", { threadId, input: codexInput(prompt, images, skill), ...(outputSchema ? skillDraftTurnSettings() : turnSettings(permissionMode)), ...(model ? { model } : {}), ...(effort ? { effort } : {}), ...(outputSchema ? { outputSchema } : {}) }, Boolean(outputSchema)); const turnId = turn.id; if (!turnId) throw new Error("Codex app-server 没有返回 turn id"); pendingStart.turnId = turnId; @@ -174,12 +203,14 @@ export class CodexAppClient { const completed = this.completedTurns.get(turnKey); if (this.completedTurns.has(turnKey)) { this.completedTurns.delete(turnKey); + const result = this.completedTurnResults.get(turnKey); + this.completedTurnResults.delete(turnKey); this.currentThreadId = ""; this.currentTurnId = ""; if (completed) throw completed; - return; + return result; } - await new Promise((resolve, reject) => this.activeTurns.set(turnKey, { resolve, reject, threadId, turnId, prompt, messageText })); + return await new Promise((resolve, reject) => this.activeTurns.set(turnKey, { resolve, reject, threadId, turnId, prompt, messageText })); } catch (error) { if (!this.currentTurnId) this.currentThreadId = ""; throw error; @@ -189,6 +220,15 @@ export class CodexAppClient { } } + /** 在静默线程中生成结构化输出。 */ + async generateSkillDraft(threadId: string, prompt: string, outputSchema: JsonRecord, model?: string, effort?: CodexReasoningEffort) { + this.silentThreadIds.add(threadId); + const result = await this.startTurn(threadId, prompt, [], "request", model, effort, undefined, undefined, undefined, outputSchema); + const output = String(field(result, "output") || "").trim(); + if (!output) throw new Error("Codex 没有返回 Skill 草稿"); + return output; + } + /** 中断当前正在运行且属于指定线程的 Codex turn。 */ async interruptCurrentTurn(requestedThreadId?: string) { const threadId = this.currentThreadId; @@ -219,11 +259,28 @@ export class CodexAppClient { return true; } + /** 标记下一次临时线程创建，使早于请求响应到达的通知也不会外泄。 */ + private async startSilentThread(method: Method, params: CodexRequestParams) { + const pendingStart = Symbol(); + this.pendingSilentThreadStarts.add(pendingStart); + try { + const result = await this.request(method, params, true); + const thread = result.thread; + if (!thread.id) throw new Error("Codex app-server 没有返回 thread id"); + this.silentThreadIds.add(thread.id); + return thread; + } finally { + this.pendingSilentThreadStarts.delete(pendingStart); + if (!this.pendingSilentThreadStarts.size) this.flushPendingThreadStartedNotifications(); + } + } + /** 发送 JSON-RPC 请求并保存待处理 Promise。 */ - private request(method: Method, params: CodexRequestParams) { + private request(method: Method, params: CodexRequestParams, silent = false) { + if (this.failing) return Promise.reject(new CodexReportedError(this.failureMessage || "Codex app-server 已停止")) as Promise>; const id = this.nextId++; - this.write({ id, method, params }); - return new Promise>((resolve, reject) => this.pending.set(id, { resolve: (result) => resolve(result as CodexRequestResult), reject })); + this.write({ id, method, params }, silent); + return new Promise>((resolve, reject) => this.pending.set(id, { resolve: (result) => resolve(result as CodexRequestResult), reject, silent })); } /** 发送无需响应的 JSON-RPC 通知。 */ @@ -232,10 +289,10 @@ export class CodexAppClient { } /** 将 JSON-RPC 消息写入 app-server 标准输入。 */ - private write(value: unknown) { + private write(value: unknown, silent = false) { const method = String(field(value, "method") || ""); const params = field(value, "params"); - if (method) logger.debug(`Codex ${method}`, { id: field(value, "id"), threadId: field(params, "threadId") }); + if (method && !silent) logger.debug(`Codex ${method}`, { id: field(value, "id"), threadId: field(params, "threadId") }); this.child.stdin?.write(`${JSON.stringify(value)}\n`); } @@ -248,8 +305,10 @@ export class CodexAppClient { try { this.handle(JSON.parse(line) as JsonRecord); } catch (error) { - logger.warn("Invalid Codex app-server output", { error, line }); - this.emit("agent_log", { text: line }); + if (!this.skillDraftActive) { + logger.warn("Invalid Codex app-server output", { error, line }); + this.emit("agent_log", { text: line }); + } } }); } @@ -259,8 +318,10 @@ export class CodexAppClient { const id = Number(message.id); if (message.error && this.pending.has(id)) { const error = String(field(message.error, "message") || "Codex request failed"); - if (/not materialized yet.*includeTurns/i.test(error)) logger.debug("Codex thread has no messages yet", { id }); - else logger.warn("Codex request failed", { id, error }); + if (!this.pending.get(id)?.silent) { + if (/not materialized yet.*includeTurns/i.test(error)) logger.debug("Codex thread has no messages yet", { id }); + else logger.warn("Codex request failed", { id, error }); + } return this.reject(id, error); } if (this.pending.has(id)) return this.resolve(id, message.result); @@ -270,6 +331,7 @@ export class CodexAppClient { /** 转换并广播 app-server 通知。 */ private handleNotification(method: string, params: JsonRecord) { + if (this.handleSilentNotification(method, params)) return; if (method === "skills/changed") { this.emit("skills_changed", {}); return; @@ -393,6 +455,81 @@ export class CodexAppClient { this.emit("agent_event", { agent: "codex", ...event }); } + /** 草稿线程存续期间隐藏 app-server 自身输出，避免混入网页诊断日志。 */ + private get skillDraftActive() { + return this.pendingSilentThreadStarts.size > 0 || this.silentThreadIds.size > 0; + } + + /** 消化临时草稿线程事件，不广播、不落补充历史。 */ + private handleSilentNotification(method: string, params: JsonRecord) { + if (method === "thread/started") { + const thread = field(params, "thread"); + const threadId = String(field(thread, "id") || ""); + if (!threadId) return false; + if (this.silentThreadIds.has(threadId)) return true; + if (!this.pendingSilentThreadStarts.size) return false; + this.pendingThreadStartedNotifications.push(params); + return true; + } + if (!(method === "error" || method.startsWith("turn/") || method.startsWith("item/") || method === "thread/tokenUsage/updated")) return false; + const threadId = String(field(params, "threadId") || this.currentThreadId); + if (!threadId || !this.silentThreadIds.has(threadId)) return false; + const turnId = String(field(params, "turnId") || field(field(params, "turn"), "id") || this.currentTurnId); + if (method === "turn/started") { + this.currentThreadId = threadId; + this.currentTurnId = turnId; + this.notifyTurnStarted(threadId, turnId); + return true; + } + if (method === "item/agentMessage/delta") { + const itemId = String(field(params, "itemId") || ""); + if (itemId && turnId) { + const key = itemCacheKey({ threadId, turnId, itemId }); + this.textByItem.set(key, `${this.textByItem.get(key) || ""}${String(field(params, "delta") || "")}`); + } + return true; + } + if (method === "item/completed") { + const item = normalizeItem(field(params, "item")); + const itemId = String(field(item, "id") || ""); + const key = itemCacheKey({ threadId, turnId, itemId }); + if (item.type === "agent_message" && turnId) { + const text = String(item.text || this.textByItem.get(key) || "").trim(); + if (text) this.structuredOutputByTurn.set(turnCacheKey(threadId, turnId), text); + } + if (itemId) this.textByItem.delete(key); + return true; + } + if (method !== "turn/completed") return true; + const turn = field(params, "turn") as CodexNotificationParams<"turn/completed">["turn"]; + const completedTurnId = String(field(turn, "id") || turnId); + const turnKey = turnCacheKey(threadId, completedTurnId); + const output = this.structuredOutputByTurn.get(turnKey) || ""; + this.structuredOutputByTurn.delete(turnKey); + this.finishTurnDeltas(threadId, completedTurnId); + const failure = turn.error ? new CodexReportedError(turn.error.message || "Codex turn failed") : null; + const pending = this.activeTurns.get(turnKey); + const result = { output }; + if (pending) { + this.activeTurns.delete(turnKey); + failure ? pending.reject(failure) : pending.resolve(result); + } else if (completedTurnId) { + this.completedTurns.set(turnKey, failure); + this.completedTurnResults.set(turnKey, result); + } + if (completedTurnId === this.currentTurnId) { + this.currentThreadId = ""; + this.currentTurnId = ""; + } + return true; + } + + /** 请求响应确定静默线程 ID 后，重新分流早到的启动通知。 */ + private flushPendingThreadStartedNotifications() { + const notifications = this.pendingThreadStartedNotifications.splice(0); + notifications.forEach((params) => this.handleNotification("thread/started", params)); + } + /** 补充历史落盘后再广播 turn 终态，确保界面完成状态可跨 Agent 重启恢复。 */ private completeTurn(event: AgentEvent, params: JsonRecord, eventScope: ReturnType) { const turn = (params as unknown as CodexNotificationParams<"turn/completed">).turn; @@ -492,6 +629,12 @@ export class CodexAppClient { private answerServerRequest(message: JsonRecord) { const method = String(message.method); const params = (field(message, "params") as JsonRecord) || {}; + const threadId = String(field(params, "threadId") || this.currentThreadId); + if (this.silentThreadIds.has(threadId)) { + const result = method === "item/permissions/requestApproval" ? { permissions: {}, scope: "turn" } : { decision: "decline" }; + this.write({ id: message.id, result }); + return; + } if (["item/commandExecution/requestApproval", "item/fileChange/requestApproval", "item/permissions/requestApproval"].includes(method)) { const requestId = String(message.id); this.approvalRequests.set(requestId, { id: Number(message.id), method, params }); @@ -519,13 +662,15 @@ export class CodexAppClient { private failAll(message: string, reported = false) { if (this.failing) return; this.failing = true; + this.failureMessage = message; this.approvalRequests.forEach((request, requestId) => this.emit("codex_approval_resolved", { ...request.params, requestId, decision: request.decision || "cancel" })); const failedTurns = new Map(); this.activeTurns.forEach(({ threadId, turnId, prompt, messageText }, key) => { + if (this.silentThreadIds.has(threadId)) return; if (!this.finalizingTurns.has(key)) failedTurns.set(key, { threadId, turnId, prompt, messageText }); }); const pendingStart = this.pendingTurnStart; - if (pendingStart?.turnId) { + if (pendingStart?.turnId && !this.silentThreadIds.has(pendingStart.threadId)) { const key = turnCacheKey(pendingStart.threadId, pendingStart.turnId); if (!this.finalizingTurns.has(key) && !failedTurns.has(key)) failedTurns.set(key, { threadId: pendingStart.threadId, turnId: pendingStart.turnId, prompt: pendingStart.prompt, messageText: pendingStart.messageText }); } @@ -544,7 +689,12 @@ export class CodexAppClient { this.nextItemSequences.clear(); this.plansByTurn.clear(); this.completedTurns.clear(); + this.completedTurnResults.clear(); this.approvalRequests.clear(); + this.silentThreadIds.clear(); + this.pendingSilentThreadStarts.clear(); + this.pendingThreadStartedNotifications.length = 0; + this.structuredOutputByTurn.clear(); this.pendingTurnStart = undefined; this.startedTurnKeys.clear(); this.lastUsage = null; @@ -606,6 +756,17 @@ function threadSettings(permissionMode: AgentPermissionMode) { return { approvalPolicy: permissionMode === "full" ? "never" as const : "on-request" as const, sandbox: permissionMode === "full" ? "danger-full-access" as const : "workspace-write" as const, config: codexConfig(permissionMode) }; } +function skillDraftThreadSettings(cwd: string) { + return { + approvalPolicy: "never" as const, + sandbox: "read-only" as const, + config: { model_reasoning_summary: "auto", mcp_servers: {} }, + cwd, + developerInstructions: SKILL_DRAFT_INSTRUCTIONS, + ephemeral: true, + }; +} + function turnSettings(permissionMode: AgentPermissionMode) { return { approvalPolicy: permissionMode === "full" ? "never" as const : "on-request" as const, @@ -613,6 +774,10 @@ function turnSettings(permissionMode: AgentPermissionMode) { }; } +function skillDraftTurnSettings() { + return { approvalPolicy: "never" as const, sandboxPolicy: { type: "readOnly" as const, networkAccess: false } }; +} + /** 将文本、本地图片和显式 Skill 转换为 Codex turn 输入。 */ function codexInput(prompt: string, images: string[], skill?: CodexSkillSelector): CodexTurnInput[] { const text = skill && !mentionsSkill(prompt, skill.name) ? `$${skill.name} ${prompt}` : prompt; diff --git a/canvas-agent/src/agent/codex-protocol.ts b/canvas-agent/src/agent/codex-protocol.ts index 4bb8019..7c977f9 100644 --- a/canvas-agent/src/agent/codex-protocol.ts +++ b/canvas-agent/src/agent/codex-protocol.ts @@ -49,9 +49,11 @@ export type CodexTurnInput = type ThreadOptions = { approvalPolicy: "never" | "on-request"; - sandbox: "workspace-write" | "danger-full-access"; + sandbox: "read-only" | "workspace-write" | "danger-full-access"; config: JsonRecord; cwd?: string; + developerInstructions?: string; + ephemeral?: boolean; }; type CodexRequestSpec = { @@ -70,6 +72,10 @@ type CodexRequestSpec = { params: ThreadOptions & { threadId: string }; result: { thread: CodexThread }; }; + "thread/fork": { + params: ThreadOptions & { threadId: string; threadSource: "user" }; + result: { thread: CodexThread }; + }; "thread/list": { params: { limit: number; @@ -89,6 +95,10 @@ type CodexRequestSpec = { params: { threadId: string }; result: Record; }; + "thread/unsubscribe": { + params: { threadId: string }; + result: { status: "notLoaded" | "notSubscribed" | "unsubscribed" }; + }; "model/list": { params: { limit: number; includeHidden: boolean }; result: { data: CodexModel[]; nextCursor: string | null }; @@ -102,7 +112,7 @@ type CodexRequestSpec = { result: { effectiveEnabled: boolean }; }; "turn/start": { - params: { threadId: string; input: CodexTurnInput[]; approvalPolicy: "never" | "on-request"; sandboxPolicy: { type: "workspaceWrite"; networkAccess: boolean } | { type: "dangerFullAccess" }; model?: string; effort?: CodexReasoningEffort }; + params: { threadId: string; input: CodexTurnInput[]; approvalPolicy: "never" | "on-request"; sandboxPolicy: { type: "readOnly"; networkAccess: boolean } | { type: "workspaceWrite"; networkAccess: boolean } | { type: "dangerFullAccess" }; model?: string; effort?: CodexReasoningEffort; outputSchema?: JsonRecord }; result: { turn: CodexTurn }; }; "turn/interrupt": { diff --git a/canvas-agent/src/agent/codex.ts b/canvas-agent/src/agent/codex.ts index 0e49425..85f820f 100644 --- a/canvas-agent/src/agent/codex.ts +++ b/canvas-agent/src/agent/codex.ts @@ -1,9 +1,11 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; +import { z } from "zod"; +import type { CanvasSnapshot } from "../canvas/types.js"; import { logger } from "../utils/logger.js"; -import { errorMessage, field } from "../utils/value.js"; +import { errorMessage, field, type JsonRecord } from "../utils/value.js"; import { CodexAppClient, CodexReportedError } from "./codex-client.js"; import { codexEventHistory } from "./codex-event-history.js"; import { settledTurnIds, summarizeCodexThread, threadMessages } from "./codex-history.js"; @@ -11,6 +13,35 @@ import type { CodexReasoningEffort, CodexSkillMetadata, CodexSkillSelector, Code import type { AgentAttachment, AgentEmit, AgentPermissionMode } from "./types.js"; type CodexRunOptions = { threadId?: string; cwd?: string; permissionMode?: AgentPermissionMode; model?: string; effort?: CodexReasoningEffort; skill?: CodexSkillSelector; messageText?: string; appEmit?: AgentEmit; onStart?: () => void; onThread?: (threadId: string) => void; onTurn?: (turnId: string) => void; onFinish?: () => void }; +type CodexSkillDraftInput = { model?: string; effort?: CodexReasoningEffort } & ({ source: "conversation"; threadId: string } | { source: "canvas"; snapshot: CanvasSnapshot }); + +const skillNamePattern = /^[a-z0-9]+(?:-[a-z0-9]+)*$/; +const skillDraftSchema = z.object({ + name: z.string().trim().min(1).max(64).regex(skillNamePattern), + displayName: z.string().trim().max(64), + description: z.string().trim().min(1).max(1024).refine((value) => !/[<>]/.test(value)), + instructions: z.string().trim().min(1).max(20000), + shortDescription: z.string().trim().max(64).refine((value) => !value || value.length >= 25), + defaultPrompt: z.string().trim().max(1024), +}).strict().superRefine((draft, context) => { + if (draft.defaultPrompt && !mentionsSkill(draft.defaultPrompt, draft.name)) context.addIssue({ code: "custom", path: ["defaultPrompt"], message: `默认提示词必须包含 $${draft.name}` }); +}); + +const SKILL_DRAFT_OUTPUT_SCHEMA: JsonRecord = { + type: "object", + additionalProperties: false, + required: ["name", "displayName", "description", "instructions", "shortDescription", "defaultPrompt"], + properties: { + name: { type: "string", pattern: "^[a-z0-9]+(?:-[a-z0-9]+)*$", minLength: 1, maxLength: 64 }, + displayName: { type: "string", maxLength: 64 }, + description: { type: "string", pattern: "^[^<>]+$", minLength: 1, maxLength: 1024 }, + instructions: { type: "string", minLength: 1, maxLength: 20000 }, + shortDescription: { anyOf: [{ type: "string", maxLength: 0 }, { type: "string", minLength: 25, maxLength: 64 }] }, + defaultPrompt: { type: "string", maxLength: 1024 }, + }, +}; + +export type AgentSkillDraft = z.infer; export class CodexSkillLookupError extends Error { override name = "CodexSkillLookupError"; @@ -34,6 +65,13 @@ export async function runCodexTurn(prompt: string, lifecycleEmit: AgentEmit, att await codexQueue; } +/** 从当前对话或指定网页画布生成可编辑草稿，不写入 Skill 文件。 */ +export async function generateCodexSkillDraft(emit: AgentEmit, cwd: string, input: CodexSkillDraftInput): Promise { + const queued = codexQueue.catch(() => undefined).then(() => generateCodexSkillDraftNow(emit, cwd, input)); + codexQueue = queued; + return await queued; +} + /** 中断当前线程正在执行的 Codex turn。 */ export async function interruptCodexTurn(threadId?: string) { if (!codexApp) return false; @@ -192,6 +230,190 @@ async function loadCodexThread(emit: AgentEmit, threadId: string, cwd: string | return thread; } +async function generateCodexSkillDraftNow(emit: AgentEmit, cwd: string, input: CodexSkillDraftInput) { + const app = await getCodexApp(emit); + let threadId = ""; + try { + const thread = input.source === "conversation" ? await app.forkSkillDraftThread(input.threadId, cwd) : await app.startSkillDraftThread(cwd); + threadId = String(field(thread, "id") || ""); + const raw = await app.generateSkillDraft(threadId, skillDraftPrompt(input), SKILL_DRAFT_OUTPUT_SCHEMA, input.model, input.effort); + let value: unknown; + try { + value = JSON.parse(raw); + } catch { + throw new Error("Codex 返回的 Skill 草稿不是有效 JSON"); + } + const parsed = skillDraftSchema.safeParse(value); + if (!parsed.success) throw new Error("Codex 返回的 Skill 草稿格式不正确"); + assertDraftHasNoSensitiveValues(parsed.data, input.source === "canvas" ? canvasPrivateValues(input.snapshot) : []); + return parsed.data; + } finally { + if (threadId) await app.closeSkillDraftThread(threadId).catch((error) => logger.warn("Failed to release Skill draft thread", { threadId, error })); + } +} + +function skillDraftPrompt(input: CodexSkillDraftInput) { + const source = input.source === "conversation" + ? "从这个临时分支继承的完整对话中，识别已经实际完成且值得复用的稳定流程。不要总结本条提炼请求，也不要保留一次性的结论、错误排查过程或工具日志。" + : `从下面经过清理的画布快照中，识别节点、连线和生成步骤所表达的可复用流程。不要把画布节点 ID 写进执行说明。\n\n画布快照：\n${JSON.stringify(canvasSkillSource(input.snapshot))}`; + return [ + "请生成一个可编辑的 Codex Skill 草稿。", + source, + "要求：", + "- name 使用不超过 64 个字符的小写字母、数字和连字符，优先使用简短的动词短语。", + "- description 同时说明能力和触发场景；所有何时使用的信息都写在这里。", + "- instructions 只写另一个 Codex 真正需要的、可复用的命令式步骤、约束和输出要求，不写 YAML frontmatter。", + "- shortDescription 写 25–64 个字符的人类可读短说明；没有合适内容时返回空字符串。", + "- defaultPrompt 必须包含与 name 完全一致的 $skill-name 调用标记；没有合适内容时返回空字符串。", + "- displayName 使用简洁的人类可读名称。", + "- 不得输出 Token、API Key、密码、凭证、本地路径、媒体 URL、敏感 URL、临时错误、调试日志或一次性结果。", + "只按 outputSchema 返回对象。", + ].join("\n"); +} + +const MAX_CANVAS_SKILL_NODES = 300; +const MAX_CANVAS_SKILL_CONNECTIONS = 600; +const MAX_CANVAS_SKILL_NODE_CHARS = 100_000; +const MAX_CANVAS_SKILL_SOURCE_CHARS = 120_000; + +/** 只保留理解画布流程所需的信息，避免把媒体、外部地址、坐标和本地凭证送入草稿线程。 */ +export function canvasSkillSource(snapshot: CanvasSnapshot): JsonRecord { + const allNodes = prioritizedCanvasNodes(snapshot); + const nodes = allNodes.slice(0, MAX_CANVAS_SKILL_NODES); + const nodeRefs = new Map(allNodes.map((node, index) => [node.id, `node-${index + 1}`])); + const title = cleanCanvasString(snapshot.title, nodeRefs); + const source: JsonRecord & { nodes: JsonRecord[]; connections: Array<{ from: string; to: string }> } = { + ...(title ? { title } : {}), + nodes: [], + connections: [], + }; + let truncated = nodes.length < allNodes.length; + nodes.forEach((node, index) => { + const metadata = { ...(node.metadata || {}) }; + if (node.type !== "text") delete metadata.content; + const cleanMetadata = sanitizeCanvasValue(metadata, nodeRefs); + const nodeTitle = cleanCanvasString(node.title, nodeRefs); + const summary = { ref: `node-${index + 1}`, type: node.type, ...(nodeTitle ? { title: nodeTitle } : {}) }; + const candidate = { ...summary, ...(cleanMetadata && Object.keys(cleanMetadata as JsonRecord).length ? { metadata: cleanMetadata } : {}) }; + if (canvasSourceFits({ ...source, nodes: [...source.nodes, candidate] }, MAX_CANVAS_SKILL_NODE_CHARS)) source.nodes.push(candidate); + else if (canvasSourceFits({ ...source, nodes: [...source.nodes, summary] }, MAX_CANVAS_SKILL_NODE_CHARS)) (source.nodes.push(summary), truncated = true); + else truncated = true; + }); + const includedRefs = new Set(source.nodes.map((node) => String(node.ref || ""))); + const selectedNodeRefs = (snapshot.selectedNodeIds || []).flatMap((id) => nodeRefs.get(id) || []).filter((ref) => includedRefs.has(ref)); + if (selectedNodeRefs.length) source.selectedNodeRefs = selectedNodeRefs; + const connections = (snapshot.connections || []).flatMap(({ fromNodeId, toNodeId }) => { + const from = nodeRefs.get(fromNodeId); + const to = nodeRefs.get(toNodeId); + return from && to && includedRefs.has(from) && includedRefs.has(to) ? [{ from, to }] : []; + }); + if (connections.length > MAX_CANVAS_SKILL_CONNECTIONS) truncated = true; + connections.slice(0, MAX_CANVAS_SKILL_CONNECTIONS).forEach((connection) => { + if (canvasSourceFits({ ...source, connections: [...source.connections, connection] }, MAX_CANVAS_SKILL_SOURCE_CHARS - 32)) source.connections.push(connection); + else truncated = true; + }); + if (truncated) source.truncated = true; + return source; +} + +const sensitiveCanvasKey = /api.?key|token|secret|password|authorization|credential|storage.?key|(?:local|file).?path/i; +const transientCanvasKey = /^(?:status|progress|errorDetails|taskId|createdAt|updatedAt|startedAt|completedAt)$/i; +const canvasNodeReferenceKey = /^.*(?:Node|Group|Parent|Child|Root|Source|Target|PrimaryImage)Ids?$/i; +const directLocalPath = /^(?:file:(?:\/\/)?|[a-z]:[\\/]|\\\\|\/(?!\/)(?=[^\s`'"“”<>]+\/))/i; +const fileUrl = /\bfile:(?:\/\/)?[^\s`'"“”<>]+/gi; +const inlineLocalPath = /(?]+|(?]+\/)[^\s`'"“”<>]+/giu; +const credentialAssignment = /(?:api[_ -]?key|access[_ -]?(?:key|token)|connect[_ -]?token|token|secret|password|authorization|credential)\s*(?:[:=：]|为|是)\s*(?:bearer\s+)?[`'"“]?[A-Za-z0-9_./+\-=]{8,}/gi; +const bearerToken = /\bbearer\s+[A-Za-z0-9._~+/=\-]{8,}/gi; +const jwtToken = /\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b/g; +const knownApiToken = /\b(?:sk-[A-Za-z0-9_-]{12,}|(?:gh[pousr]|github_pat)_[A-Za-z0-9_]{20,}|AKIA[A-Z0-9]{16})\b/g; +const transientIdentifier = /\b(?:task|job|request|generation|node)[_-](?:\d{4,}|[A-Fa-f0-9]{8,}|(?=[A-Za-z0-9_-]{12,}\b)(?=[A-Za-z0-9_-]*\d)[A-Za-z0-9_-]+)\b/gi; +const webUrl = /\bhttps?:\/\/[^\s<>{}\[\]`'"“”]+/gi; + +function sanitizeCanvasValue(value: unknown, nodeRefs: Map, depth = 0): unknown { + if (value === null || typeof value === "boolean" || typeof value === "number") return value; + if (typeof value === "string") return cleanCanvasString(value, nodeRefs); + if (depth >= 6) return undefined; + if (Array.isArray(value)) return value.slice(0, 300).map((item) => sanitizeCanvasValue(item, nodeRefs, depth + 1)).filter((item) => item !== undefined); + if (!value || typeof value !== "object") return undefined; + const result: JsonRecord = {}; + Object.entries(value as JsonRecord).forEach(([key, item]) => { + if (sensitiveCanvasKey.test(key) || transientCanvasKey.test(key)) return; + if (canvasNodeReferenceKey.test(key)) { + const refs = (Array.isArray(item) ? item : [item]).flatMap((id) => typeof id === "string" ? nodeRefs.get(id) || [] : []); + if (refs.length) result[key.replace(/Ids$/i, "Refs").replace(/Id$/i, "Ref")] = Array.isArray(item) ? [...new Set(refs)] : refs[0]; + return; + } + const clean = sanitizeCanvasValue(item, nodeRefs, depth + 1); + if (clean !== undefined) result[replaceCanvasNodeRefs(key, nodeRefs)] = clean; + }); + return result; +} + +function cleanCanvasString(value: unknown, nodeRefs: Map) { + const valueText = typeof value === "string" ? value.trim() : ""; + if (!valueText || /^(?:data:|blob:|https?:\/\/|file:)/i.test(valueText) || directLocalPath.test(valueText)) return undefined; + const text = replaceCanvasNodeRefs(valueText, nodeRefs) + .replace(/\b(?:data:|blob:)[^\s`'"“”<>]+/gi, "[媒体地址已移除]") + .replace(fileUrl, "[本地路径已移除]") + .replace(webUrl, "[外部地址已移除]") + .replace(inlineLocalPath, "[本地路径已移除]") + .replace(credentialAssignment, "[敏感凭证已移除]") + .replace(bearerToken, "[敏感凭证已移除]") + .replace(jwtToken, "[敏感凭证已移除]") + .replace(knownApiToken, "[敏感凭证已移除]") + .trim(); + return text.length > 12000 ? `${text.slice(0, 12000)}\n[内容已截断]` : text || undefined; +} + +function replaceCanvasNodeRefs(value: string, nodeRefs: Map) { + const nodeIds = [...nodeRefs.keys()].filter(Boolean).sort((left, right) => right.length - left.length); + const nodeIdPattern = nodeIds.length ? new RegExp(nodeIds.map((id) => id.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")).join("|"), "g") : undefined; + return nodeIdPattern ? value.replace(nodeIdPattern, (id) => nodeRefs.get(id) || id) : value; +} + +function prioritizedCanvasNodes(snapshot: CanvasSnapshot) { + const nodes = snapshot.nodes || []; + const selectedIds = new Set(snapshot.selectedNodeIds || []); + const relatedIds = new Set(); + (snapshot.connections || []).forEach(({ fromNodeId, toNodeId }) => { + if (selectedIds.has(fromNodeId)) relatedIds.add(toNodeId); + if (selectedIds.has(toNodeId)) relatedIds.add(fromNodeId); + }); + return [ + ...nodes.filter((node) => selectedIds.has(node.id)), + ...nodes.filter((node) => !selectedIds.has(node.id) && relatedIds.has(node.id)), + ...nodes.filter((node) => !selectedIds.has(node.id) && !relatedIds.has(node.id)), + ]; +} + +function canvasSourceFits(source: JsonRecord, limit: number) { + return JSON.stringify(source).length <= limit; +} + +function canvasPrivateValues(snapshot: CanvasSnapshot) { + return [snapshot.projectId, snapshot.clientId, ...(snapshot.nodes || []).map((node) => node.id), ...(snapshot.connections || []).map((connection) => connection.id)] + .filter((value): value is string => typeof value === "string" && value.length >= 6); +} + +function patternMatches(pattern: RegExp, text: string) { + pattern.lastIndex = 0; + return pattern.test(text); +} + +function mentionsSkill(prompt: string, name: string) { + return new RegExp(`\\$${name}(?![A-Za-z0-9_-]|:[A-Za-z0-9_-])`).test(prompt); +} + +export function assertDraftHasNoSensitiveValues(draft: AgentSkillDraft, privateValues: string[]) { + const text = Object.values(draft).join("\n"); + const localPath = /(?:^|[\s`'"“”（(\[{,:;：])(?:file:(?:\/\/)?|[a-z]:[\\/]|\\\\|\/(?!\/))/im; + const externalUrl = (text.match(webUrl) || []).length > 0; + const hasPrivateValue = privateValues.some((value) => text.includes(value)); + if (localPath.test(text) || /\b(?:data:|blob:)/i.test(text) || patternMatches(credentialAssignment, text) || patternMatches(bearerToken, text) || patternMatches(jwtToken, text) || patternMatches(knownApiToken, text) || patternMatches(transientIdentifier, text) || externalUrl || hasPrivateValue) { + throw new Error("生成的 Skill 草稿包含外部地址、本地路径、敏感凭证或一次性标识，已拒绝返回"); + } +} + /** 读取线程历史，并显式标记 Codex 是否已经物化 turns。 */ async function loadCodexHistory(emit: AgentEmit, threadId: string, cwd?: string) { try { diff --git a/canvas-agent/src/canvas/session.test.ts b/canvas-agent/src/canvas/session.test.ts index b77c85a..e4a4a97 100644 --- a/canvas-agent/src/canvas/session.test.ts +++ b/canvas-agent/src/canvas/session.test.ts @@ -23,6 +23,25 @@ test("MCP 读取当前激活网页的画布", async (t) => { assert.equal(field(await session.callTool("canvas_get_state", {}), "projectId"), "canvas-second"); }); +test("按精确 clientId 读取画布快照，不受当前焦点影响", (t) => { + const session = new CanvasSession(); + const first = connect(session, "first"); + const second = connect(session, "second"); + t.after(() => { + first.close(); + second.close(); + }); + session.updateState(snapshot("canvas-first"), "first"); + session.updateState(snapshot("canvas-second"), "second"); + session.activateClient("second"); + + assert.equal(field(session.canvasStateForClient("first"), "projectId"), "canvas-first"); + assert.equal(field(session.canvasStateForClient("second"), "projectId"), "canvas-second"); + assert.equal(session.canvasStateForClient("missing"), null); + first.close(); + assert.equal(session.canvasStateForClient("first"), null); +}); + test("画布写操作只发送给当前激活网页", async (t) => { const session = new CanvasSession(); const first = connect(session, "first"); @@ -202,7 +221,7 @@ test("new clients receive the current Codex state and later updates", (t) => { t.after(() => client.close()); const hello = client.event("hello"); - assert.equal(field(hello, "protocolVersion"), 4); + assert.equal(field(hello, "protocolVersion"), 5); assert.deepEqual(field(hello, "workspace"), { activeThreadId: "thread-2" }); assert.deepEqual(field(hello, "codex"), { busy: true, threadId: "thread-2", turnId: "turn-1" }); assert.deepEqual(field(hello, "pendingApprovals"), [{ requestId: "approval-1", threadId: "thread-2" }]); @@ -228,6 +247,46 @@ test("Codex 写操作在多窗口之间互斥且不能与运行 turn 并发", () assert.equal(session.beginCodexMutation(), false); }); +test("Skill draft generation broadcasts shared busy state and restores the previous thread", (t) => { + const session = new CanvasSession(); + const first = connect(session, "first"); + const second = connect(session, "second"); + t.after(() => { + first.close(); + second.close(); + }); + session.setCodexState({ threadId: "thread-1", turnId: "turn-previous" }); + const previous = session.codexStateSnapshot; + + assert.equal(session.beginCodexMutation(), true); + session.setCodexState({ busy: true, threadId: previous.threadId, turnId: "" }, { preserveReplay: true }); + assert.deepEqual(first.events("codex_state").at(-1), { busy: true, threadId: "thread-1", turnId: "" }); + assert.deepEqual(second.events("codex_state").at(-1), { busy: true, threadId: "thread-1", turnId: "" }); + assert.equal(session.beginCodexMutation(), false); + + session.setCodexState(previous, { preserveReplay: true }); + session.endCodexMutation(); + assert.deepEqual(first.events("codex_state").at(-1), previous); + assert.deepEqual(second.events("codex_state").at(-1), previous); + assert.equal(session.beginCodexMutation(), true); + session.endCodexMutation(); +}); + +test("Skill draft busy state preserves the previous turn replay until history acknowledges it", (t) => { + const session = new CanvasSession(); + session.setCodexState({ busy: true, threadId: "thread-1", turnId: "turn-1" }); + session.emitThread("agent_event", "thread-1", { turnId: "turn-1", type: "item.updated", item: { id: "assistant-1", type: "agent_message", text: "回答" } }); + session.setCodexState({ busy: false }); + const previous = session.codexStateSnapshot; + + session.setCodexState({ busy: true, threadId: "thread-1", turnId: "" }, { preserveReplay: true }); + session.setCodexState(previous, { preserveReplay: true }); + + const client = connect(session, "first", "thread-1"); + t.after(() => client.close()); + assert.equal(client.events("agent_event").length, 1); +}); + test("a bound client remains the tool target while focus changes", async (t) => { const session = new CanvasSession(); const first = connect(session, "first"); diff --git a/canvas-agent/src/canvas/session.ts b/canvas-agent/src/canvas/session.ts index 45d8d98..eebda5f 100644 --- a/canvas-agent/src/canvas/session.ts +++ b/canvas-agent/src/canvas/session.ts @@ -12,7 +12,7 @@ type PendingRequest = { clientId: string; resolve: (value: unknown) => void; rej type TurnAttachment = { clientId: string; id: string; name: string; type: string; size: number; width: number; height: number; dataUrl: string }; type ReplayEvent = { type: string; payload: Record }; export type CodexState = { busy: boolean; threadId: string; turnId: string }; -export const AGENT_PROTOCOL_VERSION = 4; +export const AGENT_PROTOCOL_VERSION = 5; const SITE_TOOLS = new Set([ "site_navigate", @@ -67,11 +67,21 @@ export class CanvasSession { return this.codexState.threadId; } + /** Return a copy that callers can restore after a temporary Codex operation. */ + get codexStateSnapshot(): CodexState { + return { ...this.codexState }; + } + /** 判断网页客户端是否仍连接到当前 Agent。 */ hasClient(clientId: string) { return this.clients.has(clientId); } + /** 读取指定网页上报的画布，避免提炼时受最近焦点或其他标签页影响。 */ + canvasStateForClient(clientId: string) { + return this.clients.has(clientId) ? this.canvasStates.get(clientId) || null : null; + } + /** 原子取得 Codex 写操作权限，避免多个网页并发切换或修改会话。 */ beginCodexMutation() { if (this.codexState.busy || this.codexMutationBusy) return false; @@ -106,13 +116,13 @@ export class CanvasSession { if (type === "agent_error") this.pendingApprovals.clear(); } - /** 更新并广播 Codex 运行状态。 */ - setCodexState(patch: Partial) { + /** 更新并广播 Codex 运行状态；静默后台活动可保留上一 turn 的断线重放。 */ + setCodexState(patch: Partial, options: { preserveReplay?: boolean } = {}) { const next = { ...this.codexState, ...patch }; const threadChanged = next.threadId !== this.codexState.threadId; const turnChanged = Boolean(this.codexState.turnId && next.turnId && next.turnId !== this.codexState.turnId); const nextTurnStarted = !this.codexState.busy && next.busy; - if (threadChanged || turnChanged || nextTurnStarted) { + if (!options.preserveReplay && (threadChanged || turnChanged || nextTurnStarted)) { this.codexReplayEvents.clear(); this.codexReplayActiveItems.clear(); } diff --git a/canvas-agent/src/server/http.ts b/canvas-agent/src/server/http.ts index 60aa3aa..91e03db 100644 --- a/canvas-agent/src/server/http.ts +++ b/canvas-agent/src/server/http.ts @@ -4,7 +4,7 @@ import path from "node:path"; import express, { type NextFunction, type Request, type Response } from "express"; import { runClaudeTurn } from "../agent/claude.js"; -import { archiveCodexThread, CodexSkillLookupError, configureCodexSkill, interruptCodexTurn, listCodexModels, listCodexSkills, listCodexThreads, readCodexThread, resolveCodexApproval, resolveCodexSkill, resumeCodexThread, runCodexTurn, startCodexThread, summarizeCodexThread } from "../agent/codex.js"; +import { archiveCodexThread, CodexSkillLookupError, configureCodexSkill, generateCodexSkillDraft, interruptCodexTurn, listCodexModels, listCodexSkills, listCodexThreads, readCodexThread, resolveCodexApproval, resolveCodexSkill, resumeCodexThread, runCodexTurn, startCodexThread, summarizeCodexThread } from "../agent/codex.js"; import type { CodexReasoningEffort, CodexSkillSelector } from "../agent/codex-protocol.js"; import type { AgentAttachment, AgentPermissionMode } from "../agent/types.js"; import { AGENT_PROTOCOL_VERSION, CanvasSession } from "../canvas/session.js"; @@ -50,6 +50,7 @@ export function startHttpServer() { return workspace; }; let draftThreadStart: ReturnType | null = null; + let skillDraftRunning = false; const prepareDraftThread = (clientId: string, permission: AgentPermissionMode) => { if (draftThreadStart) return draftThreadStart; const workspace = ensureSiteWorkspace(config); @@ -140,15 +141,48 @@ export function startHttpServer() { const result = await listCodexSkills(emit, workspace.workspacePath, String(req.query.forceReload || "") === "1"); res.json({ ok: true, data: result.skills.map((skill) => ({ ...skill, managed: skillStore.isManagedPath(skill.path) })), errors: result.errors }); })); + app.post("/agent/codex/skills/draft", codexMutation(async (req, res) => { + const workspace = ensureSiteWorkspace(config); + const source = String(req.body?.source || ""); + if (source !== "conversation" && source !== "canvas") return res.status(400).json({ ok: false, error: "Skill 草稿来源无效" }); + const clientId = String(req.body?.clientId || ""); + if (!clientId || !session.hasClient(clientId)) return res.status(409).json({ ok: false, error: "发起提炼的网页已断开，请重新连接后再试" }); + const model = String(req.body?.model || "") || undefined; + const effort = reasoningEffort(req.body?.effort); + const previousCodexState = session.codexStateSnapshot; + skillDraftRunning = true; + try { + if (source === "conversation") { + const threadId = String(req.body?.threadId || ""); + if (!threadId) return res.status(409).json({ ok: false, error: "当前没有可提炼的对话" }); + if (threadId !== (workspace.activeThreadId || "")) return res.status(409).json({ ok: false, error: "当前对话已在其他页面切换，请同步后重试" }); + const history = await readCodexThread(emit, threadId, workspace.workspacePath); + if (!history.messages.some((message) => message.role === "user" && message.turnId)) return res.status(409).json({ ok: false, error: "当前对话还没有可提炼的已完成内容" }); + session.setCodexState({ busy: true, threadId, turnId: "" }, { preserveReplay: true }); + const data = await generateCodexSkillDraft(emit, workspace.workspacePath, { source, threadId, model, effort }); + if (!session.hasClient(clientId)) return res.status(409).json({ ok: false, error: "发起提炼的网页已断开，请重新连接后再试" }); + return res.json({ ok: true, data }); + } + const snapshot = session.canvasStateForClient(clientId); + if (!snapshot || (snapshot as Record).hasCanvas === false) return res.status(409).json({ ok: false, error: "当前页面没有可提炼的画布" }); + session.setCodexState({ busy: true, threadId: workspace.activeThreadId || "", turnId: "" }, { preserveReplay: true }); + const data = await generateCodexSkillDraft(emit, workspace.workspacePath, { source, snapshot, model, effort }); + if (!session.hasClient(clientId)) return res.status(409).json({ ok: false, error: "发起提炼的网页已断开，请重新连接后再试" }); + return res.json({ ok: true, data }); + } finally { + skillDraftRunning = false; + session.setCodexState(previousCodexState, { preserveReplay: true }); + } + })); app.get("/agent/codex/skills/:name", route(async (req, res) => { res.json({ ok: true, data: await skillStore.get(routeParam(req.params.name)) }); })); - app.post("/agent/codex/skills", route(async (req, res) => { + app.post("/agent/codex/skills", codexMutation(async (req, res) => { const data = await skillStore.create(req.body); session.emitAll("skills_changed", { forceReload: true }); res.status(201).json({ ok: true, data }); })); - app.post("/agent/codex/skills/:name/enabled", route(async (req, res) => { + app.post("/agent/codex/skills/:name/enabled", codexMutation(async (req, res) => { if (typeof req.body?.enabled !== "boolean") return res.status(400).json({ ok: false, error: "Skill 启用状态无效" }); const workspace = ensureSiteWorkspace(config); const selector = skillSelector(req.body); @@ -157,12 +191,12 @@ export function startHttpServer() { session.emitAll("skills_changed", { forceReload: true }); res.json({ ok: true, data }); })); - app.post("/agent/codex/skills/:name/delete", route(async (req, res) => { + app.post("/agent/codex/skills/:name/delete", codexMutation(async (req, res) => { await skillStore.delete(routeParam(req.params.name), String(req.body?.expectedRevision || "")); session.emitAll("skills_changed", { forceReload: true }); res.json({ ok: true }); })); - app.post("/agent/codex/skills/:name", route(async (req, res) => { + app.post("/agent/codex/skills/:name", codexMutation(async (req, res) => { const data = await skillStore.update(routeParam(req.params.name), req.body); session.emitAll("skills_changed", { forceReload: true }); res.json({ ok: true, data }); @@ -326,7 +360,10 @@ export function startHttpServer() { const ok = await resolveCodexApproval(String(req.body?.requestId || ""), decision); res.status(ok ? 200 : 409).json({ ok, ...(ok ? {} : { error: "审批请求已失效" }) }); })); - app.post("/agent/codex/interrupt", route(async (req, res) => res.json({ ok: await interruptCodexTurn(String(req.body?.threadId || "")) }))); + app.post("/agent/codex/interrupt", route(async (req, res) => { + const ok = await interruptCodexTurn(skillDraftRunning ? undefined : String(req.body?.threadId || "")); + res.status(ok ? 200 : 409).json({ ok, ...(ok ? {} : { error: "当前没有可停止的任务" }) }); + })); app.post("/agent/claude/turn", (req, res) => { runClaudeTurn(String(req.body?.prompt || ""), emit); res.json({ ok: true });