From aa86cf9af42b37693f16368b6cf004f87507cacc Mon Sep 17 00:00:00 2001 From: biground Date: Tue, 10 Nov 2026 15:16:24 +0800 Subject: [PATCH 1/3] =?UTF-8?q?fix(web):=20=E5=90=8C=E6=AD=A5=20Codex=20?= =?UTF-8?q?=E5=BC=82=E6=AD=A5=E6=8F=90=E9=97=AE=E4=B8=8E=E7=AD=94=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/architecture/host-client-channel.md | 2 + .../src/native-async-questions.ts | 212 +++++++++++++++ .../host-runtime/src/native-client-channel.ts | 33 ++- .../test/native-async-questions.test.ts | 246 ++++++++++++++++++ .../web-native-async-questions.test.ts | 160 ++++++++++++ 5 files changed, 651 insertions(+), 2 deletions(-) create mode 100644 packages/host-runtime/src/native-async-questions.ts create mode 100644 packages/host-runtime/test/native-async-questions.test.ts create mode 100644 tests/integration/web-native-async-questions.test.ts diff --git a/docs/architecture/host-client-channel.md b/docs/architecture/host-client-channel.md index f06169faa..a7cc1d7b2 100644 --- a/docs/architecture/host-client-channel.md +++ b/docs/architecture/host-client-channel.md @@ -10,6 +10,8 @@ GUI、Web 和其他 App 的共享外部 Threads 由一个独立后台中的 `App 原生命令／文件审批和提问可由 Web 回答。MCP 的空字段表单以确认审批呈现，保留 `message`、风险提示与工具参数，不投影为空问题列表；带 `codex_approval_kind: "mcp_tool_call"` 的请求只提供原生 `_meta.persist` 声明的会话／永久授权范围，普通空表单只提供单次允许与拒绝。审批响应在认领请求前校验 action、空 content 与授权范围，不自动批准，也不接受未声明的持久化权限。审批严格保留原生可选 decision；Web 使用代次限定的不透明请求 ID，服务端再映射到官方 ID，旧页面不能回答重启后复用编号的新请求。首次有效回应后删除待处理项，不自动重放结果未知的响应。未支持的原生权限授予请求不转换成通用允许／拒绝；Web 提问卡片没有密码输入契约，因此含 `isSecret` 的提问也不降级成明文字段。对 Web 不能回答的官方请求，有 Desktop 连接时交由 Desktop；没有 Desktop 连接时立即返回错误并投影原因，不让 Turn 悬挂。 +`request_user_input_async` 通过原生 `agentMessage {delivery: "async", questions: [{title, options?}]}` 发布，不是等待工具结果的 server request。客户端通道从实时 Item 与最近 5 轮快照恢复问题，每个问题保留原生 `questionItemId` 和索引，复用 Web 提问卡，支持选项及自由文本。答案按 Desktop 的 `` 用户消息格式提交：活动 Turn 用带 `expectedTurnId` 的 `turn/steer`，空闲时用 `turn/start`；不伪造工具响应。原生用户消息中已确认的答案会撤下问题卡，发送前再读取当前历史，多个 Web 响应在发送前认领同一问题。结果未知时不重发；明确的原生参数／状态拒绝保留问题供重试。该投影只保存在内存中，Host 换代后从当前快照窗口恢复，不扫描整份历史；原生协议没有跨 Desktop 与 Web 的原子异步答题接口，因此同时提交仍受原生用户输入语义约束。 + - `shared-contracts/client-channel` 定义浏览器安全的版本、游标、快照和交互契约。 - `host-runtime` 提供本地认证 HTTP 通道，复用原请求分派、投影器、配置观察器和交互响应路径。 - `desktop-control` 的公共 `HostClientChannel` 负责 Node 端发现、通知恢复和只读快照访问；命令不自动重试。 diff --git a/packages/host-runtime/src/native-async-questions.ts b/packages/host-runtime/src/native-async-questions.ts new file mode 100644 index 000000000..1a9574665 --- /dev/null +++ b/packages/host-runtime/src/native-async-questions.ts @@ -0,0 +1,212 @@ +import { randomUUID } from "node:crypto"; +import type { JsonObject, JsonValue } from "@codexhost/protocol-core"; +import type { ClientInteraction } from "@codexhost/shared-contracts"; +import { ClientChannelRequestError } from "./client-channel-error.js"; +import type { NativeClientReply } from "./native-client-channel.js"; + +type Question = Extract< + ClientInteraction["interaction"], + { type: "question" } +>["questions"][number]; +type Pending = { threadId: string; question: Question }; +const object = (value: unknown): value is JsonObject => + typeof value === "object" && value !== null && !Array.isArray(value); +const keyOf = (threadId: string, questionId: string) => JSON.stringify([threadId, questionId]); +const publicId = (prefix: string, pending: Pending) => + `${prefix}async:${keyOf(pending.threadId, pending.question.id)}`; + +/** Async questions are native messages, not server requests waiting for a tool result. + * Keep only a volatile interaction projection; native history remains authoritative. */ +export class NativeAsyncQuestions { + readonly #pending = new Map(); + readonly #answered = new Map>(); + readonly #unconfirmed = new Set(); + + #settle(threadId: string, questionId: string): void { + let answered = this.#answered.get(threadId); + if (!answered) this.#answered.set(threadId, (answered = new Set())); + answered.add(questionId); + this.#pending.delete(keyOf(threadId, questionId)); + } + + #observeItem(threadId: string, item: JsonObject): void { + if (item.type === "userMessage" && Array.isArray(item.content)) { + for (const part of item.content) { + if (!object(part) || part.type !== "text" || typeof part.text !== "string") continue; + const match = + /^\s*\s*([\s\S]*?)\s*<\/send_user_message_question_reply>\s*$/u.exec( + part.text, + ); + const body = match?.[1]; + if (body === undefined) continue; + try { + const replies: unknown = JSON.parse(body); + if (!Array.isArray(replies)) continue; + for (const reply of replies) + if ( + object(reply) && + typeof reply.questionItemId === "string" && + typeof reply.answer === "string" + ) + this.#settle(threadId, reply.questionItemId); + } catch { + // Ordinary or malformed user text cannot resolve a structured question. + } + } + } + if ( + item.type !== "agentMessage" || + item.delivery !== "async" || + typeof item.id !== "string" || + !Array.isArray(item.questions) + ) + return; + item.questions.forEach((value, index) => { + if (!object(value) || typeof value.title !== "string" || !value.title.trim()) return; + if ( + value.options !== undefined && + (!Array.isArray(value.options) || + value.options.some((option) => typeof option !== "string" || !option.trim())) + ) + return; + const id = JSON.stringify(["request_user_input_async", item.id, index]); + if (this.#answered.get(threadId)?.has(id)) return; + const options = (value.options ?? []) as string[]; + const question: Question = options.length + ? { + id, + prompt: value.title, + type: "choice", + multiple: false, + allowOther: true, + options: options.map((label) => ({ value: label, label })), + } + : { id, prompt: value.title, type: "text" }; + this.#pending.set(keyOf(threadId, id), { threadId, question }); + }); + } + + observe(method: string, params: JsonObject): boolean { + if (typeof params.threadId !== "string") return false; + const before = this.#pending.size; + if (method === "thread/closed") { + this.#answered.delete(params.threadId); + for (const [key, pending] of this.#pending) + if (pending.threadId === params.threadId) this.#pending.delete(key); + } else if ((method === "item/started" || method === "item/completed") && object(params.item)) { + this.#observeItem(params.threadId, params.item); + } + return this.#pending.size !== before; + } + + observeTurns(threadId: string, turns: JsonValue[]): void { + for (const turn of turns) + if (object(turn) && Array.isArray(turn.items)) + for (const item of turn.items) if (object(item)) this.#observeItem(threadId, item); + } + + interactions(threadId: string, prefix: string): ClientInteraction[] { + return [...this.#pending.values()] + .filter( + (pending) => + pending.threadId === threadId && !this.#unconfirmed.has(publicId(prefix, pending)), + ) + .map((pending) => ({ + requestId: publicId(prefix, pending), + threadId, + kind: "question", + interaction: { type: "question", questions: [pending.question] }, + request: { + method: "request_user_input_async", + params: { questionItemId: pending.question.id }, + }, + })); + } + + async respond( + requestId: string, + threadId: string, + result: JsonValue, + prefix: string, + request: (method: string, params: JsonObject) => Promise, + ): Promise { + if (this.#unconfirmed.has(requestId)) + throw new Error("Native interaction delivery outcome is unknown; do not resend the response"); + const pending = [...this.#pending.values()].find( + (value) => publicId(prefix, value) === requestId, + ); + if (!pending) return false; + if (pending.threadId !== threadId) + throw new Error("Interaction does not belong to this Thread"); + const answers = object(result) && object(result.answers) ? result.answers : undefined; + const answer = answers?.[pending.question.id]; + if ( + !answers || + Object.keys(answers).length !== 1 || + !object(answer) || + !Array.isArray(answer.answers) || + answer.answers.length !== 1 || + typeof answer.answers[0] !== "string" || + !answer.answers[0].trim() + ) + throw new Error("Provide one answer to the native async question"); + + // Desktop may have answered since the card was rendered. Read before claiming; + // concurrent Web responders then compete in one synchronous claim below. + const page = await request("thread/turns/list", { + threadId, + limit: 5, + sortDirection: "desc", + itemsView: "full", + }); + if ("error" in page) throw new ClientChannelRequestError(page.error.code, page.error.message); + if (!object(page.result) || !Array.isArray(page.result.data)) + throw new Error("Codex returned an invalid Turn page"); + this.observeTurns(threadId, page.result.data); + if ( + !this.#pending.has(keyOf(threadId, pending.question.id)) || + this.#unconfirmed.has(requestId) + ) + return false; + this.#unconfirmed.add(requestId); + const active = page.result.data.find( + (turn) => object(turn) && turn.status === "inProgress" && typeof turn.id === "string", + ); + const text = `\n${JSON.stringify([ + { + questionItemId: pending.question.id, + question: pending.question.prompt, + answer: answer.answers[0], + }, + ])}\n\n`; + const sent = await request(object(active) ? "turn/steer" : "turn/start", { + threadId, + ...(object(active) ? { expectedTurnId: active.id } : {}), + input: [{ type: "text", text, text_elements: [] }], + clientUserMessageId: randomUUID(), + }); + if ("error" in sent) { + // Native validation refusals did not accept input; a refreshed card can be answered again. + // Transport and generation failures retain the claim because delivery is uncertain. + if (sent.error.code === -32600 || sent.error.code === -32602) + this.#unconfirmed.delete(requestId); + throw new ClientChannelRequestError(sent.error.code, sent.error.message); + } + if ( + !object(sent.result) || + (object(active) + ? sent.result.turnId !== active.id + : !object(sent.result.turn) || typeof sent.result.turn.id !== "string") + ) + throw new Error("Codex did not confirm the question reply; delivery outcome is unknown"); + this.#settle(threadId, pending.question.id); + this.#unconfirmed.delete(requestId); + return true; + } + + clear(): void { + this.#pending.clear(); + this.#answered.clear(); + this.#unconfirmed.clear(); + } +} diff --git a/packages/host-runtime/src/native-client-channel.ts b/packages/host-runtime/src/native-client-channel.ts index 5ea352fdf..906fcd081 100644 --- a/packages/host-runtime/src/native-client-channel.ts +++ b/packages/host-runtime/src/native-client-channel.ts @@ -15,6 +15,7 @@ import { nativeThreadConfiguration } from "./codex-runtime/native-thread-control import { sendNativeQueuedMessage } from "./native-queued-steering.js"; import { submitNativeQueuedInput } from "./native-queued-input.js"; import { NativeClientItems } from "./native-client-items.js"; +import { NativeAsyncQuestions } from "./native-async-questions.js"; import { nativeElicitationApproval, nativeElicitationApprovalResponse, @@ -390,6 +391,7 @@ export class NativeClientChannel { readonly #resumedThreads = new Set(); readonly #configuration = new Map>(); readonly #items = new NativeClientItems(); + readonly #asyncQuestions = new NativeAsyncQuestions(); readonly #pending = new Map(); readonly #pendingNativeIds = new Map(); readonly #unconfirmedResponses = new Set(); @@ -408,6 +410,7 @@ export class NativeClientChannel { if (!isRecord(value) || typeof value.method !== "string") return; const params = isRecord(value.params) ? (value.params as JsonObject) : {}; this.#items.observe(value.method, params); + const asyncQuestionsChanged = this.#asyncQuestions.observe(value.method, params); const threadId = threadIdFromParams(params); if (threadId) { this.noteThread(threadId); @@ -416,6 +419,8 @@ export class NativeClientChannel { if (value.method === "thread/closed") this.#resumedThreads.delete(threadId); if (value.method === "turn/started") this.#interactionErrors.delete(threadId); this.#host.events.changed(threadId, value.method); + if (asyncQuestionsChanged) + this.#host.events.changed(threadId, "codexhost/thread/questions/updated"); } if (value.method === "serverRequest/resolved") { const requestId = params.requestId; @@ -576,15 +581,17 @@ export class NativeClientChannel { } interactions(threadId: string): ClientInteraction[] { + const prefix = `${NATIVE_REQUEST_PREFIX}${this.#generationToken}:`; return [...this.#pending.values()] .filter((pending) => pending.threadId === threadId) - .map((pending) => ({ + .map((pending) => ({ requestId: pending.requestId, threadId, kind: pending.kind, interaction: pending.interaction, request: { method: pending.method, params: pending.params }, - })); + })) + .concat(this.#asyncQuestions.interactions(threadId, prefix)); } async snapshot(threadId: string, options: ClientSnapshotOptions): Promise { @@ -629,6 +636,7 @@ export class NativeClientChannel { throw new ClientChannelRequestError(turnsReply.error.code, turnsReply.error.message); const turnsResult = resultObject(turnsReply) ?? {}; const data = Array.isArray(turnsResult.data) ? turnsResult.data : []; + this.#asyncQuestions.observeTurns(threadId, data); const turnsPage = { data, nextCursor: typeof turnsResult.nextCursor === "string" ? turnsResult.nextCursor : null, @@ -660,6 +668,26 @@ export class NativeClientChannel { } async respond(requestId: string | number, threadId: string, result: JsonValue): Promise { + const generation = this.#generationToken; + const prefix = `${NATIVE_REQUEST_PREFIX}${generation}:`; + if (typeof requestId === "string" && requestId.startsWith(`${prefix}async:`)) { + try { + return await this.#asyncQuestions.respond( + requestId, + threadId, + result, + prefix, + (method, params) => + generation === this.#generationToken + ? this.request(method, params) + : Promise.resolve({ + error: { code: -32094, message: "Host generation changed; refresh the Thread" }, + }), + ); + } finally { + this.#host.events.changed(threadId, "codexhost/thread/questions/updated"); + } + } if (typeof requestId === "string" && this.#unconfirmedResponses.has(requestId)) throw new Error("Native interaction delivery outcome is unknown; do not resend the response"); const pending = typeof requestId === "string" ? this.#pending.get(requestId) : undefined; @@ -688,6 +716,7 @@ export class NativeClientChannel { this.#interactionErrors.clear(); this.#configuration.clear(); this.#items.clear(); + this.#asyncQuestions.clear(); for (const threadId of threads) this.#host.events.changed(threadId, "thread/status/changed"); } diff --git a/packages/host-runtime/test/native-async-questions.test.ts b/packages/host-runtime/test/native-async-questions.test.ts new file mode 100644 index 000000000..dbdd73a0e --- /dev/null +++ b/packages/host-runtime/test/native-async-questions.test.ts @@ -0,0 +1,246 @@ +import { expect, it, vi } from "vitest"; +import type { JsonObject } from "@codexhost/protocol-core"; +import { ClientChannelEvents } from "../src/client-channel-events.js"; +import { NativeClientChannel, type NativeClientReply } from "../src/native-client-channel.js"; + +const threadId = "native-thread"; +function defined(value: T | undefined): T { + if (value === undefined) throw new Error("Missing native test fixture value"); + return value; +} +const questionId = (index = 0) => + JSON.stringify(["request_user_input_async", "call-question", index]); +const question: JsonObject = { + id: "call-question", + type: "agentMessage", + delivery: "async", + text: "Choose a direction\n- Keep\n- Replace", + questions: [{ title: "Choose a direction", options: ["Keep", "Replace"] }], +}; + +function reply(index = 0): JsonObject { + return { + id: `desktop-answer-${index}`, + type: "userMessage", + content: [ + { + type: "text", + text: `\n${JSON.stringify([ + { questionItemId: questionId(index), question: "Choose a direction", answer: "Keep" }, + ])}\n\n`, + }, + ], + }; +} + +function fixture(status = "inProgress") { + const events = new ClientChannelEvents(); + const turns: JsonObject[] = [{ id: "turn-1", status, items: [structuredClone(question)] }]; + const request = vi.fn(async (method: string, params: JsonObject): Promise => { + if (method === "thread/resume" || method === "thread/read") + return { + result: { + thread: { + id: threadId, + cwd: "/synthetic", + modelProvider: "openai", + createdAt: 1, + updatedAt: 1, + status: { type: status === "inProgress" ? "active" : "idle" }, + }, + }, + }; + if (method === "thread/turns/list") + return { result: { data: structuredClone(turns), nextCursor: null } }; + if (method === "turn/steer") return { result: { turnId: params.expectedTurnId ?? null } }; + if (method === "turn/start") + return { result: { turn: { id: "turn-2", status: "inProgress" } } }; + throw new Error(`Unexpected method ${method}`); + }); + const respond = vi.fn(async () => undefined); + const channel = new NativeClientChannel({ events, request, respond }); + const observe = (item: JsonObject, target = threadId) => + channel.observeMessage({ + method: "item/completed", + params: { threadId: target, turnId: "turn-1", item }, + }); + const answer = { answers: { [questionId()]: { answers: ["My own direction"] } } }; + return { channel, request, respond, turns, observe, answer, events }; +} + +it("restores structured async questions from native history as answerable cards", async () => { + const { channel } = fixture(); + const snapshot = await channel.snapshot(threadId, {}); + expect(snapshot.interactions).toHaveLength(1); + expect(snapshot.interactions[0]).toMatchObject({ + threadId, + kind: "question", + interaction: { + type: "question", + questions: [ + { + id: questionId(), + prompt: "Choose a direction", + type: "choice", + multiple: false, + allowOther: true, + options: [ + { value: "Keep", label: "Keep" }, + { value: "Replace", label: "Replace" }, + ], + }, + ], + }, + }); + expect((await channel.snapshot(threadId, {})).interactions[0]?.requestId).toBe( + snapshot.interactions[0]?.requestId, + ); +}); + +it("observes async questions live without changing the native message", () => { + const { channel, observe, events } = fixture(); + const changed = vi.spyOn(events, "changed"); + const item = structuredClone(question); + observe(item); + observe(item); + expect(channel.interactions(threadId)).toHaveLength(1); + expect(item).toEqual(question); + expect( + changed.mock.calls.filter(([, method]) => method === "codexhost/thread/questions/updated"), + ).toHaveLength(1); +}); + +it.each(["inProgress", "completed"])( + "delivers an async answer while the Turn is %s", + async (status) => { + const { channel, request, respond, answer } = fixture(status); + const snapshot = await channel.snapshot(threadId, {}); + const pending = defined(snapshot.interactions[0]); + expect(pending).toBeDefined(); + await channel.respond(pending.requestId, threadId, answer); + const method = status === "inProgress" ? "turn/steer" : "turn/start"; + const sent = request.mock.calls.find(([name]) => name === method)?.[1]; + expect(sent).toMatchObject({ threadId, input: [{ type: "text" }] }); + if (status === "inProgress") expect(sent?.expectedTurnId).toBe("turn-1"); + const text = (sent?.input as JsonObject[])[0]?.text as string; + expect(JSON.parse(text.replace(/<\/?send_user_message_question_reply>/gu, "").trim())).toEqual([ + { questionItemId: questionId(), question: "Choose a direction", answer: "My own direction" }, + ]); + expect(respond).not.toHaveBeenCalled(); + expect(channel.interactions(threadId)).toEqual([]); + await channel.snapshot(threadId, {}); + await channel.respond(pending.requestId, threadId, answer); + expect(request.mock.calls.filter(([name]) => name === method)).toHaveLength(1); + }, +); + +it("withdraws Desktop answers and does not resurrect them from a late history page", async () => { + const { channel, observe } = fixture(); + const pending = defined((await channel.snapshot(threadId, {})).interactions[0]); + expect(pending).toBeDefined(); + observe(reply()); + expect(channel.interactions(threadId)).toEqual([]); + expect((await channel.snapshot(threadId, {})).interactions).toEqual([]); +}); + +it("checks native history before delivering an answer that Desktop already supplied", async () => { + const { channel, request, turns, answer } = fixture(); + const pending = defined((await channel.snapshot(threadId, {})).interactions[0]); + expect(pending).toBeDefined(); + (defined(turns[0]).items as JsonObject[]).push(reply()); + await channel.respond(pending.requestId, threadId, answer); + expect(request.mock.calls.some(([method]) => method === "turn/steer")).toBe(false); + expect(channel.interactions(threadId)).toEqual([]); +}); + +it("retains unanswered questions and their native indexes across descending history", async () => { + const { channel, turns } = fixture(); + (defined(turns[0]).items as JsonObject[])[0] = { + ...question, + questions: [...(question.questions as JsonObject[]), { title: "Anything else?" }], + }; + turns.unshift({ id: "turn-2", status: "completed", items: [reply()] }); + const snapshot = await channel.snapshot(threadId, {}); + expect(snapshot.interactions).toHaveLength(1); + expect(snapshot.interactions[0]?.interaction).toMatchObject({ + questions: [{ id: questionId(1), type: "text", prompt: "Anything else?" }], + }); +}); + +it("rejects wrong-Thread and invalid answers without consuming the question", async () => { + const { channel, request, answer } = fixture(); + const pending = defined((await channel.snapshot(threadId, {})).interactions[0]); + expect(pending).toBeDefined(); + await expect(channel.respond(pending.requestId, "another-thread", answer)).rejects.toThrow( + "does not belong", + ); + await expect(channel.respond(pending.requestId, threadId, { answers: {} })).rejects.toThrow(); + expect(channel.interactions(threadId)).toHaveLength(1); + expect(request.mock.calls.some(([method]) => method === "turn/steer")).toBe(false); +}); + +it("claims concurrent answers once and never resends an uncertain delivery", async () => { + const { channel, request, answer } = fixture(); + const pending = defined((await channel.snapshot(threadId, {})).interactions[0]); + expect(pending).toBeDefined(); + const original = defined(request.getMockImplementation()); + request.mockImplementation(async (method, params) => { + if (method === "turn/steer") throw new Error("Synthetic transport failure"); + return original(method, params); + }); + const results = await Promise.allSettled([ + channel.respond(pending.requestId, threadId, answer), + channel.respond(pending.requestId, threadId, answer), + ]); + expect(results.some((result) => result.status === "rejected")).toBe(true); + expect(request.mock.calls.filter(([method]) => method === "turn/steer")).toHaveLength(1); + await channel.snapshot(threadId, {}); + await expect(channel.respond(pending.requestId, threadId, answer)).rejects.toThrow("unknown"); +}); + +it("keeps a question answerable after an explicit native validation refusal", async () => { + const { channel, request, answer } = fixture(); + const pending = defined((await channel.snapshot(threadId, {})).interactions[0]); + const original = defined(request.getMockImplementation()); + request.mockImplementation(async (method, params) => + method === "turn/steer" + ? { error: { code: -32600, message: "active Turn changed" } } + : original(method, params), + ); + await expect(channel.respond(pending.requestId, threadId, answer)).rejects.toThrow( + "active Turn changed", + ); + expect(channel.interactions(threadId)).toHaveLength(1); + request.mockImplementation(original); + await expect(channel.respond(pending.requestId, threadId, answer)).resolves.toBe(true); +}); + +it("does not deliver an answer after the Host resets during the native refresh", async () => { + const { channel, request, answer } = fixture(); + const pending = defined((await channel.snapshot(threadId, {})).interactions[0]); + const original = defined(request.getMockImplementation()); + request.mockImplementation(async (method, params) => { + if (method === "thread/turns/list") channel.resetGeneration(); + return original(method, params); + }); + await expect(channel.respond(pending.requestId, threadId, answer)).rejects.toThrow( + "generation changed", + ); + expect(request.mock.calls.some(([method]) => method === "turn/steer")).toBe(false); +}); + +it("ignores ordinary or malformed messages and invalidates IDs after a Host reset", async () => { + const { channel, observe, answer } = fixture(); + observe({ ...question, delivery: "normal" }); + observe({ ...question, questions: [{ title: "Invalid", options: [1] }] }); + expect(channel.interactions(threadId)).toEqual([]); + observe(question); + const pending = defined(channel.interactions(threadId)[0]); + expect(pending).toBeDefined(); + channel.resetGeneration(); + expect(channel.interactions(threadId)).toEqual([]); + expect(channel.isStaleRequestId(pending.requestId)).toBe(true); + expect(await channel.respond(pending.requestId, threadId, answer)).toBe(false); + observe(question); + expect(channel.interactions(threadId)[0]?.requestId).not.toBe(pending.requestId); +}); diff --git a/tests/integration/web-native-async-questions.test.ts b/tests/integration/web-native-async-questions.test.ts new file mode 100644 index 000000000..2601854fc --- /dev/null +++ b/tests/integration/web-native-async-questions.test.ts @@ -0,0 +1,160 @@ +import { expect, it, vi } from "vitest"; +import type { JsonObject } from "@codexhost/protocol-core"; +import { jsonValueSchema } from "@codexhost/shared-contracts"; +import { ClientChannelEvents } from "../../packages/host-runtime/src/client-channel-events.js"; +import { + NativeClientChannel, + type NativeClientReply, +} from "../../packages/host-runtime/src/native-client-channel.js"; +import { ChInteractions } from "../../packages/web-server/src/ch-interactions.ts"; +import { EventHub } from "../../packages/web-server/src/transport.ts"; + +const threadId = "native-async"; +const questionId = JSON.stringify(["request_user_input_async", "call-native-question", 0]); + +function fixture() { + const items: JsonObject[] = [ + { + id: "call-native-question", + type: "agentMessage", + delivery: "async", + text: "Which direction?\n- Keep\n- Replace", + questions: [{ title: "Which direction?", options: ["Keep", "Replace"] }], + }, + ]; + const request = vi.fn(async (method: string, params: JsonObject): Promise => { + if (method === "thread/resume" || method === "thread/read") + return { + result: { + thread: { + id: threadId, + cwd: "/synthetic", + modelProvider: "openai", + createdAt: 1, + updatedAt: 1, + status: { type: "active" }, + }, + }, + }; + if (method === "thread/turns/list") + return { + result: { + data: [{ id: "active-turn", status: "inProgress", items: structuredClone(items) }], + nextCursor: null, + }, + }; + if (method === "turn/steer") return { result: { turnId: params.expectedTurnId ?? null } }; + throw new Error(`Unexpected native request ${method}`); + }); + const native = new NativeClientChannel({ + events: new ClientChannelEvents(), + request, + respond: async () => { + throw new Error("Async answers are user messages"); + }, + }); + const events = new EventHub("/synthetic"); + const invoke = vi.spyOn(events, "invoke"); + const error = vi.fn(); + const web = new ChInteractions( + { + subscribe: () => () => {}, + request: async (method: string, params: Record) => { + const reply = await native.request(method, jsonValueSchema.parse(params) as JsonObject); + if ("error" in reply) throw new Error(reply.error.message); + return reply.result as T; + }, + snapshot: (id, options = {}) => native.snapshot(id, options), + respond: async (input) => ({ + resolved: await native.respond( + input.requestId, + input.threadId, + jsonValueSchema.parse(input.result), + ), + }), + }, + events, + error, + ); + return { items, request, native, events, invoke, error, web }; +} + +it.each(["Keep", "Use the existing behavior"])( + "delivers the Web answer '%s' through the native async-question protocol", + async (answer) => { + const { native, events, invoke, request, web, error } = fixture(); + try { + web.update(await native.snapshot(threadId, {})); + expect(invoke).toHaveBeenCalledWith("user-questions/request", threadId, { + questions: [ + { + id: questionId, + question: "Which direction?", + options: [ + { label: "Keep", description: "" }, + { label: "Replace", description: "" }, + ], + multiSelect: false, + allowOther: true, + }, + ], + }); + const invocation = invoke.mock.results[0]?.value as + ReturnType | undefined; + if (!invocation) throw new Error("Web did not present the native question"); + events.settle({ + eventId: invocation.eventId, + outcome: { + kind: "resolved", + value: { + answers: [ + { + id: questionId, + ...(answer === "Keep" ? { selected: [answer] } : { custom: answer }), + }, + ], + }, + }, + }); + await vi.waitFor(() => + expect(request.mock.calls.filter(([method]) => method === "turn/steer")).toHaveLength(1), + ); + const sent = request.mock.calls.find(([method]) => method === "turn/steer")?.[1]; + expect(sent).toMatchObject({ threadId, expectedTurnId: "active-turn" }); + const text = (sent?.input as JsonObject[])[0]?.text; + expect(text).toBe( + `\n${JSON.stringify([{ questionItemId: questionId, question: "Which direction?", answer }])}\n\n`, + ); + web.update(await native.snapshot(threadId, {})); + expect(invoke).toHaveBeenCalledOnce(); + expect(error.mock.calls.every(([, message]) => message === null)).toBe(true); + } finally { + web.clear(); + } + }, +); + +it("withdraws a Web card when the same async question is answered in Desktop", async () => { + const { native, items, invoke, web } = fixture(); + try { + web.update(await native.snapshot(threadId, {})); + const invocation = invoke.mock.results[0]?.value as ReturnType | undefined; + if (!invocation) throw new Error("Web did not present the native question"); + const cancelled = expect(invocation.result).rejects.toThrow("cancelled"); + items.push({ + id: "desktop-answer", + type: "userMessage", + content: [ + { + type: "text", + text: `\n${JSON.stringify([{ questionItemId: questionId, question: "Which direction?", answer: "Keep" }])}\n\n`, + }, + ], + }); + web.update(await native.snapshot(threadId, {})); + await cancelled; + expect(invoke).toHaveBeenCalledOnce(); + } finally { + web.clear(); + } +}); From c9380330ab8a545e582d9dcbcdd8794d69d688b4 Mon Sep 17 00:00:00 2001 From: biground Date: Tue, 10 Nov 2026 15:23:04 +0800 Subject: [PATCH 2/3] =?UTF-8?q?test(web):=20=E5=9B=BA=E5=AE=9A=E5=AE=9E?= =?UTF-8?q?=E6=97=B6=E5=90=8C=E6=AD=A5=E5=9B=9E=E5=BD=92=E7=9A=84=E6=B5=8F?= =?UTF-8?q?=E8=A7=88=E5=99=A8=E8=AF=AD=E8=A8=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/web-server/test/browser/ch-realtime.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/web-server/test/browser/ch-realtime.test.ts b/packages/web-server/test/browser/ch-realtime.test.ts index 7eb6e5f9b..1b3085735 100644 --- a/packages/web-server/test/browser/ch-realtime.test.ts +++ b/packages/web-server/test/browser/ch-realtime.test.ts @@ -59,7 +59,7 @@ it( ); const browser = await chromium.launch(); t.after(() => browser.close()); - const page = await browser.newPage({ viewport: { width: 1280, height: 900 } }); + const page = await browser.newPage({ viewport: { width: 1280, height: 900 }, locale: "en-US" }); const errors: string[] = []; page.on("pageerror", (error) => errors.push(error.message)); await page.goto(url); @@ -123,11 +123,12 @@ it( ); const browser = await chromium.launch(); t.after(() => browser.close()); - const a = await browser.newPage({ viewport: { width: 1280, height: 900 } }), + const a = await browser.newPage({ viewport: { width: 1280, height: 900 }, locale: "en-US" }), b = await browser.newPage({ viewport: { width: 390, height: 844 }, isMobile: true, hasTouch: true, + locale: "en-US", }); const errors: string[] = []; for (const page of [a, b]) page.on("pageerror", (error) => errors.push(error.message)); From fe7def4eaa552980f14e91517f2c4b2688d98af1 Mon Sep 17 00:00:00 2001 From: BytePioneer-AI <28417416@qq.com> Date: Tue, 10 Nov 2026 17:51:58 +0800 Subject: [PATCH 3/3] fix(web): skip native async questions without sending input --- docs/architecture/host-client-channel.md | 2 +- .../src/native-async-questions.ts | 27 ++++++++++++------- .../test/native-async-questions.test.ts | 26 ++++++++++++++++++ .../web-native-async-questions.test.ts | 25 +++++++++++++++++ 4 files changed, 69 insertions(+), 11 deletions(-) diff --git a/docs/architecture/host-client-channel.md b/docs/architecture/host-client-channel.md index a7cc1d7b2..f2d1bb7d3 100644 --- a/docs/architecture/host-client-channel.md +++ b/docs/architecture/host-client-channel.md @@ -10,7 +10,7 @@ GUI、Web 和其他 App 的共享外部 Threads 由一个独立后台中的 `App 原生命令／文件审批和提问可由 Web 回答。MCP 的空字段表单以确认审批呈现，保留 `message`、风险提示与工具参数，不投影为空问题列表；带 `codex_approval_kind: "mcp_tool_call"` 的请求只提供原生 `_meta.persist` 声明的会话／永久授权范围，普通空表单只提供单次允许与拒绝。审批响应在认领请求前校验 action、空 content 与授权范围，不自动批准，也不接受未声明的持久化权限。审批严格保留原生可选 decision；Web 使用代次限定的不透明请求 ID，服务端再映射到官方 ID，旧页面不能回答重启后复用编号的新请求。首次有效回应后删除待处理项，不自动重放结果未知的响应。未支持的原生权限授予请求不转换成通用允许／拒绝；Web 提问卡片没有密码输入契约，因此含 `isSecret` 的提问也不降级成明文字段。对 Web 不能回答的官方请求，有 Desktop 连接时交由 Desktop；没有 Desktop 连接时立即返回错误并投影原因，不让 Turn 悬挂。 -`request_user_input_async` 通过原生 `agentMessage {delivery: "async", questions: [{title, options?}]}` 发布，不是等待工具结果的 server request。客户端通道从实时 Item 与最近 5 轮快照恢复问题，每个问题保留原生 `questionItemId` 和索引，复用 Web 提问卡，支持选项及自由文本。答案按 Desktop 的 `` 用户消息格式提交：活动 Turn 用带 `expectedTurnId` 的 `turn/steer`，空闲时用 `turn/start`；不伪造工具响应。原生用户消息中已确认的答案会撤下问题卡，发送前再读取当前历史，多个 Web 响应在发送前认领同一问题。结果未知时不重发；明确的原生参数／状态拒绝保留问题供重试。该投影只保存在内存中，Host 换代后从当前快照窗口恢复，不扫描整份历史；原生协议没有跨 Desktop 与 Web 的原子异步答题接口，因此同时提交仍受原生用户输入语义约束。 +`request_user_input_async` 通过原生 `agentMessage {delivery: "async", questions: [{title, options?}]}` 发布，不是等待工具结果的 server request。客户端通道从实时 Item 与最近 5 轮快照恢复问题，每个问题保留原生 `questionItemId` 和索引，复用 Web 提问卡，支持选项及自由文本。Web 的“跳过”只撤下当前 Host 代次的问题投影，不发送原生输入或取消请求；Host 换代后，原生历史中的未答问题仍可恢复。答案按 Desktop 的 `` 用户消息格式提交：活动 Turn 用带 `expectedTurnId` 的 `turn/steer`，空闲时用 `turn/start`；不伪造工具响应。原生用户消息中已确认的答案会撤下问题卡，发送前再读取当前历史，多个 Web 响应在发送前认领同一问题。结果未知时不重发；明确的原生参数／状态拒绝保留问题供重试。该投影只保存在内存中，Host 换代后从当前快照窗口恢复，不扫描整份历史；原生协议没有跨 Desktop 与 Web 的原子异步答题接口，因此同时提交仍受原生用户输入语义约束。 - `shared-contracts/client-channel` 定义浏览器安全的版本、游标、快照和交互契约。 - `host-runtime` 提供本地认证 HTTP 通道，复用原请求分派、投影器、配置观察器和交互响应路径。 diff --git a/packages/host-runtime/src/native-async-questions.ts b/packages/host-runtime/src/native-async-questions.ts index 1a9574665..5cdfbe512 100644 --- a/packages/host-runtime/src/native-async-questions.ts +++ b/packages/host-runtime/src/native-async-questions.ts @@ -19,13 +19,13 @@ const publicId = (prefix: string, pending: Pending) => * Keep only a volatile interaction projection; native history remains authoritative. */ export class NativeAsyncQuestions { readonly #pending = new Map(); - readonly #answered = new Map>(); + readonly #settled = new Map>(); readonly #unconfirmed = new Set(); #settle(threadId: string, questionId: string): void { - let answered = this.#answered.get(threadId); - if (!answered) this.#answered.set(threadId, (answered = new Set())); - answered.add(questionId); + let settled = this.#settled.get(threadId); + if (!settled) this.#settled.set(threadId, (settled = new Set())); + settled.add(questionId); this.#pending.delete(keyOf(threadId, questionId)); } @@ -70,7 +70,7 @@ export class NativeAsyncQuestions { ) return; const id = JSON.stringify(["request_user_input_async", item.id, index]); - if (this.#answered.get(threadId)?.has(id)) return; + if (this.#settled.get(threadId)?.has(id)) return; const options = (value.options ?? []) as string[]; const question: Question = options.length ? { @@ -90,7 +90,7 @@ export class NativeAsyncQuestions { if (typeof params.threadId !== "string") return false; const before = this.#pending.size; if (method === "thread/closed") { - this.#answered.delete(params.threadId); + this.#settled.delete(params.threadId); for (const [key, pending] of this.#pending) if (pending.threadId === params.threadId) this.#pending.delete(key); } else if ((method === "item/started" || method === "item/completed") && object(params.item)) { @@ -145,12 +145,19 @@ export class NativeAsyncQuestions { Object.keys(answers).length !== 1 || !object(answer) || !Array.isArray(answer.answers) || - answer.answers.length !== 1 || - typeof answer.answers[0] !== "string" || - !answer.answers[0].trim() + answer.answers.length > 1 || + (answer.answers.length === 1 && + (typeof answer.answers[0] !== "string" || !answer.answers[0].trim())) ) throw new Error("Provide one answer to the native async question"); + // The Web card encodes Skip as an empty answer list. Dismiss only this + // volatile projection; no native answer or cancellation is submitted. + if (answer.answers.length === 0) { + this.#settle(threadId, pending.question.id); + return true; + } + // Desktop may have answered since the card was rendered. Read before claiming; // concurrent Web responders then compete in one synchronous claim below. const page = await request("thread/turns/list", { @@ -206,7 +213,7 @@ export class NativeAsyncQuestions { clear(): void { this.#pending.clear(); - this.#answered.clear(); + this.#settled.clear(); this.#unconfirmed.clear(); } } diff --git a/packages/host-runtime/test/native-async-questions.test.ts b/packages/host-runtime/test/native-async-questions.test.ts index dbdd73a0e..fbe1396ce 100644 --- a/packages/host-runtime/test/native-async-questions.test.ts +++ b/packages/host-runtime/test/native-async-questions.test.ts @@ -143,6 +143,27 @@ it("withdraws Desktop answers and does not resurrect them from a late history pa expect((await channel.snapshot(threadId, {})).interactions).toEqual([]); }); +it.each(["inProgress", "completed"])( + "skips a native async question locally while the Turn is %s", + async (status) => { + const { channel, request, respond, turns } = fixture(status); + const pending = defined((await channel.snapshot(threadId, {})).interactions[0]); + const original = structuredClone(turns); + request.mockClear(); + await expect( + channel.respond(pending.requestId, threadId, { + answers: { [questionId()]: { answers: [] } }, + }), + ).resolves.toBe(true); + expect(request).not.toHaveBeenCalled(); + expect(respond).not.toHaveBeenCalled(); + expect((await channel.snapshot(threadId, {})).interactions).toEqual([]); + expect(turns).toEqual(original); + channel.resetGeneration(); + expect((await channel.snapshot(threadId, {})).interactions).toHaveLength(1); + }, +); + it("checks native history before delivering an answer that Desktop already supplied", async () => { const { channel, request, turns, answer } = fixture(); const pending = defined((await channel.snapshot(threadId, {})).interactions[0]); @@ -175,6 +196,11 @@ it("rejects wrong-Thread and invalid answers without consuming the question", as "does not belong", ); await expect(channel.respond(pending.requestId, threadId, { answers: {} })).rejects.toThrow(); + await expect( + channel.respond(pending.requestId, threadId, { + answers: { [questionId()]: { answers: [" "] } }, + }), + ).rejects.toThrow(); expect(channel.interactions(threadId)).toHaveLength(1); expect(request.mock.calls.some(([method]) => method === "turn/steer")).toBe(false); }); diff --git a/tests/integration/web-native-async-questions.test.ts b/tests/integration/web-native-async-questions.test.ts index 2601854fc..c3e3618ba 100644 --- a/tests/integration/web-native-async-questions.test.ts +++ b/tests/integration/web-native-async-questions.test.ts @@ -134,6 +134,31 @@ it.each(["Keep", "Use the existing behavior"])( }, ); +it("skips a Web question without sending native input or reopening the card", async () => { + const { native, events, invoke, request, web, error } = fixture(); + try { + web.update(await native.snapshot(threadId, {})); + const invocation = invoke.mock.results[0]?.value as ReturnType | undefined; + if (!invocation) throw new Error("Web did not present the native question"); + events.settle({ + eventId: invocation.eventId, + outcome: { + kind: "resolved", + value: { answers: [{ id: questionId, selected: [] }] }, + }, + }); + await vi.waitFor(() => expect(native.interactions(threadId)).toEqual([])); + web.update(await native.snapshot(threadId, {})); + expect(invoke).toHaveBeenCalledOnce(); + expect( + request.mock.calls.some(([method]) => method === "turn/steer" || method === "turn/start"), + ).toBe(false); + expect(error.mock.calls.every(([, message]) => message === null)).toBe(true); + } finally { + web.clear(); + } +}); + it("withdraws a Web card when the same async question is answered in Desktop", async () => { const { native, items, invoke, web } = fixture(); try {