fix(web): 同步 Codex 异步提问与答复

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 提问卡，支持选项及自由文本。Web 的“跳过”只撤下当前 Host 代次的问题投影，不发送原生输入或取消请求；Host 换代后，原生历史中的未答问题仍可恢复。答案按 Desktop 的 `<send_user_message_question_reply>` 用户消息格式提交：活动 Turn 用带 `expectedTurnId` 的 `turn/steer`，空闲时用 `turn/start`；不伪造工具响应。原生用户消息中已确认的答案会撤下问题卡，发送前再读取当前历史，多个 Web 响应在发送前认领同一问题。结果未知时不重发；明确的原生参数／状态拒绝保留问题供重试。该投影只保存在内存中，Host 换代后从当前快照窗口恢复，不扫描整份历史；原生协议没有跨 Desktop 与 Web 的原子异步答题接口，因此同时提交仍受原生用户输入语义约束。
+
 - `shared-contracts/client-channel` 定义浏览器安全的版本、游标、快照和交互契约。
 - `host-runtime` 提供本地认证 HTTP 通道，复用原请求分派、投影器、配置观察器和交互响应路径。
 - `desktop-control` 的公共 `HostClientChannel` 负责 Node 端发现、通知恢复和只读快照访问；命令不自动重试。

packages/host-runtime/src/native-async-questions.ts

@@ -0,0 +1,219 @@
+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<string, Pending>();
+  readonly #settled = new Map<string, Set<string>>();
+  readonly #unconfirmed = new Set<string>();
+
+  #settle(threadId: string, questionId: string): void {
+    let settled = this.#settled.get(threadId);
+    if (!settled) this.#settled.set(threadId, (settled = new Set()));
+    settled.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*<send_user_message_question_reply>\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.#settled.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.#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)) {
+      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<NativeClientReply>,
+  ): Promise<boolean> {
+    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 ||
+      (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", {
+      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 = `<send_user_message_question_reply>\n${JSON.stringify([
+      {
+        questionItemId: pending.question.id,
+        question: pending.question.prompt,
+        answer: answer.answers[0],
+      },
+    ])}\n</send_user_message_question_reply>\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.#settled.clear();
+    this.#unconfirmed.clear();
+  }
+}

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<string>();
   readonly #configuration = new Map<string, ReturnType<typeof nativeThreadConfiguration>>();
   readonly #items = new NativeClientItems();
+  readonly #asyncQuestions = new NativeAsyncQuestions();
   readonly #pending = new Map<string, PendingNativeInteraction>();
   readonly #pendingNativeIds = new Map<string, string>();
   readonly #unconfirmedResponses = new Set<string>();
@@ -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<ClientInteraction>((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<ClientThreadSnapshot> {
@@ -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<boolean> {
+    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");
   }