fix(codex): support queued async user questions

2026-11-08 01:43:07 +00:00 · 2026-11-01 10:34:36 +08:00
parent 46bae9b0f1
commit 59dcddf32b
9 changed files with 542 additions and 17 deletions
@@ -0,0 +1,135 @@
+import { describe, expect, it, vi } from 'vitest';
+import type { ApiSessionClient } from '@/api/apiSession';
+import type { AgentState } from '@/api/types';
+import { INDETERMINATE_SYMBOL } from '../codexAppServerClient';
+import { asyncQuestionsForItem, SharedCodexAsyncQuestions } from './asyncQuestions';
+
+const item = { type: 'agentMessage', id: 'call_async', delivery: 'async', text: 'Which client?\n- Web\n- Terminal',
+    questions: [{ title: 'Which client?', options: ['Web', 'Terminal'] }, { title: 'Any details?', options: null }] };
+const questions = asyncQuestionsForItem('thread', item);
+function nativeReply(questionItemId: string, answer: string) {
+    return { type: 'userMessage', content: [{ type: 'text', text:
+        `<send_user_message_question_reply>\n${JSON.stringify([{ questionItemId, answer, question: 'Which client?' }])}\n</send_user_message_question_reply>` }] };
+}
+function fixture() {
+    let state: AgentState = {};
+    const send = vi.fn();
+    const session = { getMetadata: () => ({ codexSessionId: 'thread' }), sendAgentMessage: send,
+        updateAgentState: (fn: (state: AgentState) => AgentState) => { state = fn(state); }
+    } as unknown as ApiSessionClient;
+    const submit = vi.fn(async (_threadId: string, _text: string, _id: string) => {});
+    const manager = new SharedCodexAsyncQuestions(session, submit);
+    return { manager, send, submit, state: () => state };
+}
+
+describe('Codex queued follow-up questions', () => {
+    it('preserves per-question native IDs and options in the existing question contract', () => {
+        const f = fixture(); f.manager.observeItem('thread', item); f.manager.observeItem('thread', item);
+        expect(Object.values(f.state().requests!)).toHaveLength(2);
+        expect(questions[0].questionItemId).toBe('["request_user_input_async","call_async",0]');
+        expect(f.state().requests![questions[0].id]).toMatchObject({ tool: 'request_user_input', arguments: {
+            isBlocking: false, questions: [{ id: questions[0].questionItemId, question: 'Which client?', isOther: true,
+                options: [{ label: 'Web', description: '' }, { label: 'Terminal', description: '' }] }]
+        } });
+        expect(f.state().requests![questions[1].id].arguments).toMatchObject({ questions: [{ options: [] }] });
+    });
+
+    it.each([['Web'], ['None of the above', 'user_note: Mobile']])('submits %j as native user input and waits for consumption', async (...values) => {
+        const f = fixture(); f.manager.observeItem('thread', item);
+        const q = questions[0];
+        const reply = { id: q.id, approved: true, answers: { [q.questionItemId]: { answers: values } } };
+        await f.manager.answer(reply);
+        const answer = values.at(-1)!.replace(/^user_note: /, '');
+        expect(f.submit).toHaveBeenCalledWith('thread',
+            `<send_user_message_question_reply>\n${JSON.stringify([{ answer, question: q.title, questionItemId: q.questionItemId }])}\n</send_user_message_question_reply>`, `${q.id}:answer`);
+        expect(f.state().requests![q.id]).toBeDefined();
+        expect(f.state().completedRequests?.[q.id]).toBeUndefined();
+        await expect(f.manager.answer(reply)).rejects.toThrow('already submitted');
+        f.manager.observeItem('thread', nativeReply(q.questionItemId, answer));
+        expect(f.state().requests![q.id]).toBeUndefined();
+        expect(f.state().completedRequests![q.id]).toMatchObject({ status: 'resolved', answers: {
+            [q.questionItemId]: { answers: [values.at(-1)] }
+        } });
+        expect(f.state().requests![questions[1].id]).toBeDefined();
+    });
+
+    it('rejects missing/empty answers without submitting and allows a corrected answer', async () => {
+        const f = fixture(); f.manager.observeItem('thread', item); const q = questions[1];
+        for (const answers of [undefined, {}, { [q.questionItemId]: ['user_note:  '] }]) {
+            await expect(f.manager.answer({ id: q.id, approved: true, answers })).rejects.toThrow('answer is required');
+        }
+        expect(f.submit).not.toHaveBeenCalled();
+        await f.manager.answer({ id: q.id, approved: true, answers: { [q.questionItemId]: ['user_note: Details'] } });
+        expect(f.submit).toHaveBeenCalledOnce();
+    });
+
+    it('consumes terminal replies before replay and does not reopen answered questions', async () => {
+        const f = fixture(); const q = questions[0];
+        const reply = nativeReply(q.questionItemId, 'Terminal');
+        f.manager.history('thread', { turns: [{ items: [item, reply] }] });
+        f.manager.observeItem('thread', item);
+        expect(f.state().requests![q.id]).toBeUndefined();
+        expect(f.state().completedRequests![q.id].status).toBe('resolved');
+        expect(await f.manager.answer({ id: q.id, approved: true, answers: { [q.questionItemId]: ['Web'] } })).toBe(false);
+        expect(f.submit).not.toHaveBeenCalled();
+        f.manager.observeItem('thread', item);
+        expect(Object.keys(f.state().requests!)).toEqual([questions[1].id]);
+        expect(f.send.mock.lastCall![0]).toMatchObject({ type: 'tool-call-result', callId: q.id, output: {
+            answers: { [q.questionItemId]: { answers: ['Terminal'] } }
+        } });
+    });
+
+    it('scopes child replies and cards to their own thread', async () => {
+        const f = fixture(); f.manager.observeItem('child', item);
+        const q = asyncQuestionsForItem('child', item)[0];
+        f.manager.observeItem('thread', nativeReply(q.questionItemId, 'Web'));
+        expect(f.state().requests![q.id]).toBeDefined();
+        await f.manager.answer({ id: q.id, approved: true, answers: { [q.questionItemId]: ['Terminal'] } });
+        expect(f.submit.mock.lastCall![0]).toBe('child');
+        f.manager.observeItem('child', nativeReply(q.questionItemId, 'Terminal'));
+        expect(f.send.mock.lastCall![0]).toMatchObject({ type: 'agent-run-trace', agentId: 'child',
+            message: { type: 'tool-call-result', callId: q.id }, scope: { parentThreadId: 'thread' } });
+    });
+
+    it.each([false, true])('retries only definitively rejected submissions (indeterminate: %s)', async indeterminate => {
+        const f = fixture(); f.manager.observeItem('thread', item); const q = questions[0];
+        f.submit.mockRejectedValueOnce(Object.assign(new Error('Disconnected'), { [INDETERMINATE_SYMBOL]: indeterminate }));
+        const reply = { id: q.id, approved: true, answers: { [q.questionItemId]: ['Web'] } };
+        await expect(f.manager.answer(reply)).rejects.toThrow('Disconnected');
+        if (indeterminate) await expect(f.manager.answer(reply)).rejects.toThrow('already submitted');
+        else await f.manager.answer(reply);
+        expect(f.submit).toHaveBeenCalledTimes(indeterminate ? 1 : 2);
+        expect(f.state().completedRequests?.[q.id]).toBeUndefined();
+    });
+
+    it('dismisses an optional question without sending input or interrupting the turn', async () => {
+        const f = fixture(); f.manager.observeItem('thread', item); const q = questions[0];
+        await f.manager.answer({ id: q.id, approved: false });
+        expect(f.submit).not.toHaveBeenCalled();
+        expect(f.state().completedRequests![q.id].status).toBe('canceled');
+        f.manager.observeItem('thread', item);
+        expect(f.state().requests![q.id]).toBeUndefined();
+        f.manager.observeItem('thread', nativeReply(q.questionItemId, 'Terminal'));
+        expect(f.state().completedRequests![q.id].status).toBe('resolved');
+    });
+
+    it('parses literal reply tags inside an answer and ignores quoted or malformed replies', () => {
+        const f = fixture(); f.manager.observeItem('thread', item); const q = questions[1];
+        for (const text of ['<send_user_message_question_reply>bad json</send_user_message_question_reply>',
+            `Example: ${nativeReply(q.questionItemId, 'example').content[0].text}`]) {
+            f.manager.observeItem('thread', { type: 'userMessage', content: [{ type: 'text', text }] });
+        }
+        expect(f.state().requests![q.id]).toBeDefined();
+        const answer = 'Check </send_user_message_question_reply> escaping';
+        f.manager.observeItem('thread', nativeReply(q.questionItemId, answer));
+        expect(f.state().completedRequests![q.id].answers).toEqual({ [q.questionItemId]: { answers: [`user_note: ${answer}`] } });
+    });
+
+    it('withdraws controls on exit without persisting a false native result', () => {
+        const f = fixture(); f.manager.observeItem('thread', item); f.manager.close();
+        expect(f.state().requests).toEqual({});
+        expect(f.send).not.toHaveBeenCalled();
+        f.manager.observeItem('thread', item);
+        expect(f.state().requests).toEqual({});
+    });
+});
@@ -0,0 +1,178 @@
+import type { ApiSessionClient } from '@/api/apiSession';
+import { isIndeterminateError } from '../codexAppServerClient';
+import type { PermissionReply } from './permissions';
+import { record, string } from './gateway';
+
+type AsyncQuestion = {
+    id: string;
+    threadId: string;
+    questionItemId: string;
+    title: string;
+    options: string[];
+    input: {
+        isBlocking: false;
+        questions: Array<{ id: string; question: string; isOther: true; options: Array<{ label: string; description: string }> }>;
+    };
+};
+
+function requestId(threadId: string, questionItemId: string): string {
+    return `codex-async-question:${JSON.stringify([threadId, questionItemId])}`;
+}
+
+export function asyncQuestionReplies(text: string): Array<{ questionItemId: string; question?: string; answer: string }> {
+    const opening = '<send_user_message_question_reply>';
+    const closing = '</send_user_message_question_reply>';
+    text = text.trim();
+    if (!text.startsWith(opening) || !text.endsWith(closing)) return [];
+    let replies: unknown;
+    try { replies = JSON.parse(text.slice(opening.length, -closing.length)); } catch { return []; }
+    return (Array.isArray(replies) ? replies : [replies]).flatMap(reply => {
+        const parsed = record(reply);
+        const questionItemId = string(parsed.questionItemId);
+        return questionItemId && typeof parsed.answer === 'string'
+            ? [{ questionItemId, question: string(parsed.question), answer: parsed.answer }] : [];
+    });
+}
+
+/** Async questions are agentMessage items, not requestUserInput server requests. */
+export function asyncQuestionsForItem(threadId: string, value: unknown): AsyncQuestion[] {
+    const item = record(value);
+    const itemId = string(item.id);
+    if (item.type !== 'agentMessage' || item.delivery !== 'async' || !itemId || !Array.isArray(item.questions)) return [];
+    return item.questions.flatMap((value, index) => {
+        const question = record(value);
+        const title = string(question.title);
+        if (!title) return [];
+        const options = Array.isArray(question.options) ? question.options.filter((option): option is string => typeof option === 'string') : [];
+        // This is the native TUI's identity, including the index within a batch.
+        const questionItemId = JSON.stringify(['request_user_input_async', itemId, index]);
+        return [{ id: requestId(threadId, questionItemId), threadId, questionItemId, title, options, input: {
+            isBlocking: false as const,
+            questions: [{ id: questionItemId, question: title, isOther: true as const,
+                options: options.map(label => ({ label, description: '' })) }]
+        } }];
+    });
+}
+
+type Pending = { question: AsyncQuestion; submitted: boolean };
+type Resolution = { status: 'resolved'; answer: string } | { status: 'canceled' };
+
+/** Reuses the existing question UI; replies are native user input, never RPC responses. */
+export class SharedCodexAsyncQuestions {
+    private readonly pending = new Map<string, Pending>();
+    private readonly questions = new Map<string, AsyncQuestion>();
+    private readonly completed = new Map<string, Resolution>();
+    private closed = false;
+
+    constructor(private readonly session: ApiSessionClient,
+        private readonly submit: (threadId: string, text: string, clientUserMessageId: string) => Promise<void>) {}
+
+    observeItem(threadId: string, value: unknown): void {
+        if (this.closed) return;
+        const item = record(value);
+        if (item.type === 'userMessage' && Array.isArray(item.content)) {
+            for (const part of item.content) {
+                if (record(part).type !== 'text') continue;
+                for (const reply of asyncQuestionReplies(string(record(part).text) ?? '')) {
+                    const id = requestId(threadId, reply.questionItemId);
+                    if (this.completed.get(id)?.status === 'resolved') continue;
+                    const resolution = { status: 'resolved' as const, answer: reply.answer };
+                    this.completed.set(id, resolution);
+                    const question = this.questions.get(id);
+                    if (question) this.resolve(question, resolution);
+                }
+            }
+        }
+        for (const question of asyncQuestionsForItem(threadId, item)) {
+            this.questions.set(question.id, question);
+            const resolution = this.completed.get(question.id);
+            if (resolution) { this.resolve(question, resolution); continue; }
+            if (this.pending.has(question.id)) continue;
+            this.pending.set(question.id, { question, submitted: false });
+            this.session.updateAgentState(state => {
+                const completedRequests = { ...state.completedRequests };
+                delete completedRequests[question.id];
+                return { ...state, completedRequests, requests: { ...state.requests, [question.id]: {
+                    tool: 'request_user_input', toolCallId: question.id, arguments: question.input,
+                    createdAt: typeof item.completedAtMs === 'number' ? item.completedAtMs : Date.now()
+                } } };
+            });
+        }
+    }
+
+    /** Read answers first so replay never briefly resurrects answered questions. */
+    history(threadId: string, thread: unknown): void {
+        const turns = record(thread).turns;
+        if (!Array.isArray(turns)) return;
+        for (const turn of turns) {
+            const items = record(turn).items;
+            if (!Array.isArray(items)) continue;
+            for (const item of items) if (record(item).type === 'userMessage') this.observeItem(threadId, item);
+        }
+    }
+
+    async answer(reply: PermissionReply): Promise<boolean> {
+        const pending = this.pending.get(reply.id);
+        if (!pending) return false;
+        if (this.closed || pending.submitted) throw new Error('Question already submitted or unavailable');
+        const question = pending.question;
+        if (!reply.approved) {
+            // Dismissing an optional follow-up must not interrupt the running turn.
+            this.completed.set(question.id, { status: 'canceled' });
+            this.resolve(question, { status: 'canceled' });
+            return true;
+        }
+        const raw = reply.answers?.[question.questionItemId];
+        const values = Array.isArray(raw) ? raw : raw?.answers;
+        if (!values?.length) throw new Error('An answer is required');
+        const answer = values.filter(value => value !== 'None of the above' || question.options.includes(value))
+            .map(value => !question.options.includes(value) && value.startsWith('user_note: ') ? value.slice('user_note: '.length) : value)
+            .join('\n');
+        if (!answer.trim()) throw new Error('An answer is required');
+        const text = `<send_user_message_question_reply>\n${JSON.stringify([{
+            answer, question: question.title, questionItemId: question.questionItemId
+        }])}\n</send_user_message_question_reply>`;
+        pending.submitted = true;
+        try {
+            await this.submit(question.threadId, text, `${question.id}:answer`);
+        } catch (error) {
+            // A lost acknowledgement cannot authorize another copy of the answer.
+            if (!isIndeterminateError(error)) pending.submitted = false;
+            throw error;
+        }
+        // Native input/history, including an answer from another surface, settles
+        // the card. Acceptance alone does not prove the answer was consumed.
+        return true;
+    }
+
+    private resolve(question: AsyncQuestion, resolution: Resolution, emitResult = true): void {
+        this.pending.delete(question.id);
+        const answers = resolution.status === 'resolved' ? {
+            [question.questionItemId]: { answers: [question.options.includes(resolution.answer) ? resolution.answer : `user_note: ${resolution.answer}`] }
+        } : undefined;
+        const id = `codex:${question.threadId}:${question.id}:${resolution.status}`;
+        const message = { type: 'tool-call-result', callId: question.id, output: { status: resolution.status, ...(answers ? { answers } : {}) }, id };
+        const rootThreadId = this.session.getMetadata()?.codexSessionId;
+        if (emitResult) this.session.sendAgentMessage(rootThreadId && rootThreadId !== question.threadId ? {
+            type: 'agent-run-trace', agentId: question.threadId, cardId: `codex-agent:${question.threadId}`, message, id,
+            scope: { role: 'child', threadId: question.threadId, parentThreadId: rootThreadId }, scope_role: 'child'
+        } : message, id);
+        this.session.updateAgentState(state => {
+            const requests = { ...state.requests };
+            const prior = requests[question.id] ?? state.completedRequests?.[question.id];
+            delete requests[question.id];
+            return { ...state, requests, completedRequests: { ...state.completedRequests, [question.id]: {
+                tool: 'request_user_input', toolCallId: question.id, arguments: question.input,
+                createdAt: prior?.createdAt, completedAt: state.completedRequests?.[question.id]?.completedAt ?? Date.now(),
+                status: resolution.status, ...(answers ? { answers } : {})
+            } } };
+        });
+    }
+
+    close(): void {
+        this.closed = true;
+        // These questions still exist natively and can be restored on resume.
+        // Withdraw controls without persisting a false native completion.
+        for (const { question } of this.pending.values()) this.resolve(question, { status: 'canceled' }, false);
+    }
+}
@@ -11,7 +11,8 @@ const ReplySchema = z.object({
    reason: z.string().optional(),
    answers: z.union([z.record(z.string(), z.array(z.string())), z.record(z.string(), z.object({ answers: z.array(z.string()) }))]).optional()
 });
-type Reply = z.infer<typeof ReplySchema>;
+export type PermissionReply = z.infer<typeof ReplySchema>;
+type Reply = PermissionReply;
 type Pending = { nativeId: string | number; threadId: string; turnId?: string; toolCallId: string; userInput: boolean; input: unknown; submitted: boolean; answer(reply: Reply): void; cancel(error: Error): void };

 /** No winner is inferred from submitting a Web candidate: only native resolution closes it. */
@@ -30,9 +31,11 @@ export class SharedCodexPermissions {
            scope: { role: 'child', threadId: request.threadId, parentThreadId: rootThreadId }, scope_role: 'child'
        } : message, id);
    }
-    constructor(private readonly session: ApiSessionClient, private readonly client: CodexAppServerClient, private readonly generation: string) {
+    constructor(private readonly session: ApiSessionClient, private readonly client: CodexAppServerClient, private readonly generation: string,
+        answerAsyncQuestion?: (reply: Reply) => Promise<boolean>) {
        session.rpcHandlerManager.registerHandler(RPC_METHODS.Permission, async (raw: unknown) => {
            const reply = ReplySchema.parse(raw);
+            if (await answerAsyncQuestion?.(reply)) return;
            const request = this.pending.get(reply.id);
            if (!request || request.submitted) throw new Error('Request already resolved or submitted');
            const questions = record(request.input).questions;
@@ -6,8 +6,48 @@ import type { ApiSessionClient } from '@/api/apiSession';
 import { SharedCodexProjection, inputText } from './projection';
 import { codexPlanProposalId } from './plan';
 import { extractNotifySummary } from '@hapi/protocol/messages';
+import { asyncQuestionsForItem } from './asyncQuestions';

 describe('shared history projection', () => {
+    it('displays async answers as readable user messages while preserving native IDs', async () => {
+        const sendUserMessage = vi.fn(); const committed = vi.fn(async () => {});
+        const observe = vi.fn();
+        const session = { getMetadata: () => ({}), updateMetadata: vi.fn(), sendUserMessage } as unknown as ApiSessionClient;
+        const projection = new SharedCodexProjection(session, 'thread', committed, undefined, undefined, observe);
+        const text = `<send_user_message_question_reply>\n${JSON.stringify([{
+            questionItemId: '["request_user_input_async","call_async",0]', question: 'Which client?', answer: 'Web'
+        }])}\n</send_user_message_question_reply>`;
+        const item = { type: 'userMessage', id: 'native-answer', clientId: 'client-answer', content: [{ type: 'text', text }] };
+        await projection.notification('item/completed', { threadId: 'thread', turnId: 'turn', item });
+        expect(observe).toHaveBeenCalledWith('thread', item);
+        expect(committed).toHaveBeenCalledWith('client-answer');
+        expect(sendUserMessage).toHaveBeenCalledWith('Which client?\nWeb', undefined, 'client-answer');
+    });
+    it.each([undefined, 'root'])('projects async questions as actionable cards, including active snapshots (parent: %s)', async parentThreadId => {
+        const send = vi.fn();
+        const observe = vi.fn();
+        const session = { getMetadata: () => ({}), sendAgentMessage: send } as unknown as ApiSessionClient;
+        const projection = new SharedCodexProjection(session, 'thread', async () => {}, parentThreadId, undefined, observe);
+        const item = { type: 'agentMessage', id: 'call_async', delivery: 'async', phase: 'final_answer',
+            text: 'Which client?\n- Web\n- Terminal', questions: [{ title: 'Which client?', options: ['Web', 'Terminal'] }] };
+        await projection.notification('item/started', { threadId: 'thread', turnId: 'turn', item });
+        expect(send).not.toHaveBeenCalled();
+        await projection.notification('item/completed', { threadId: 'thread', turnId: 'turn', item });
+        await projection.notification('item/completed', { threadId: 'thread', turnId: 'turn', item });
+        expect(send).toHaveBeenCalledOnce();
+        const question = asyncQuestionsForItem('thread', item)[0];
+        const body = parentThreadId ? send.mock.calls[0][0].message : send.mock.calls[0][0];
+        expect(body).toMatchObject({ type: 'tool-call', name: 'request_user_input', callId: question.id, input: question.input });
+        expect(observe).toHaveBeenCalledWith('thread', item);
+        const live = send.mock.calls.slice();
+        projection.reset(); send.mockClear();
+        await projection.history({ turns: [{ id: 'turn', status: 'inProgress', items: [item] }] });
+        expect(send.mock.calls).toEqual(live);
+        // Async notifications with no questions still render their ordinary text.
+        await projection.notification('item/completed', { threadId: 'thread', turnId: 'turn',
+            item: { ...item, id: 'update', text: 'Still working', questions: null } });
+        expect(parentThreadId ? send.mock.lastCall![0].message : send.mock.lastCall![0]).toMatchObject({ type: 'message', message: 'Still working' });
+    });
    it.each([undefined, 'root'])('projects command outcomes for live output and replay (parent: %s)', async parentThreadId => {
        const send = vi.fn();
        const session = { getMetadata: () => ({}), sendAgentMessage: send } as unknown as ApiSessionClient;
@@ -7,6 +7,7 @@ import { AppServerEventConverter } from '../utils/appServerEventConverter';
 import { record, string } from './gateway';
 import { codexPlanProposalId } from './plan';
 import { CodexCompactionSummaryLookup } from '../utils/codexCompactionSummary';
+import { asyncQuestionReplies, asyncQuestionsForItem } from './asyncQuestions';

 export function inputText(input: unknown): string {
    if (!Array.isArray(input)) return '';
@@ -49,7 +50,8 @@ export class SharedCodexProjection {
    private summaryThread: unknown;
    constructor(private readonly session: ApiSessionClient, readonly threadId: string,
        private readonly committed: (id: string) => Promise<void>, private readonly parentThreadId?: string,
-        private readonly readSummaryThread?: () => Promise<unknown>) {
+        private readonly readSummaryThread?: () => Promise<unknown>,
+        private readonly observeItem?: (threadId: string, item: unknown) => void) {
        this.compactionSummaries = new CodexCompactionSummaryLookup(threadId);
        if (!parentThreadId) for (const [id, turn] of Object.entries(session.getMetadata()?.conversationHistoryTurns ?? {})) this.turns.set(id, turn);
    }
@@ -114,6 +116,17 @@ export class SharedCodexProjection {
        const send = (body: Record<string, unknown>, key: string) => this.send({
            ...body, ...(turnId ? { turn_id: turnId } : {})
        }, key);
+        const asyncQuestions = asyncQuestionsForItem(this.threadId, item);
+        if (asyncQuestions.length) {
+            if (method === 'item/completed') {
+                for (const question of asyncQuestions) {
+                    send({ type: 'tool-call', name: 'request_user_input', callId: question.id, input: question.input }, question.id);
+                }
+                this.observeItem?.(this.threadId, item);
+            }
+            return;
+        }
+        if (method === 'item/completed') this.observeItem?.(this.threadId, item);
        // Settings may already describe the next turn by the time queued
        // notifications are projected. Never attribute usage to that model.
        if (turnId && method === 'turn/started' && modelAtReceipt && !this.turnModels.has(turnId)) {
@@ -130,7 +143,9 @@ export class SharedCodexProjection {
                if (turnId) this.turns.set(id, turnId);
                await this.committed(id);
                const text = inputText(item.content);
-                if (text) this.session.sendUserMessage(text, undefined, id);
+                const replies = asyncQuestionReplies(text);
+                const displayText = replies.length ? replies.map(reply => [reply.question, reply.answer].filter(Boolean).join('\n')).join('\n\n') : text;
+                if (displayText) this.session.sendUserMessage(displayText, undefined, id);
                this.session.updateMetadata(metadata => ({ ...metadata, conversationHistoryTurns: Object.fromEntries(this.turns),
                    ...(turnId && (!firstInTurn || firstInTurn === id) ? { conversationHistoryPoints: { ...metadata.conversationHistoryPoints, [id]: true } } : {})
                }));
@@ -244,7 +259,8 @@ export class SharedCodexProjection {
                await this.project('item/started', params, undefined, true);
                // Active snapshots can contain partial assistant text. Do not
                // settle it under the final stable id and suppress completion.
-                if (turn.status !== 'inProgress' || record(item).status === 'completed' || record(item).type === 'userMessage') {
+                if (turn.status !== 'inProgress' || record(item).status === 'completed' || record(item).type === 'userMessage'
+                    || asyncQuestionsForItem(this.threadId, item).length > 0) {
                    if (!this.parentThreadId) {
                        const title = successfulTitle(record(item));
                        if (title && string(record(item).id)) {
@@ -6,6 +6,7 @@ import type { SessionBootstrapResult } from '@/agent/sessionFactory';
 import { SharedCodexRoot, type RootHost } from './root';
 import { codexPlanProposalId } from './plan';
 import { RPC_METHODS } from '@hapi/protocol/rpcMethods';
+import { asyncQuestionsForItem } from './asyncQuestions';

 type NativeTurn = { id: string; status: string; items: unknown[] };
 type NativeRequest = { id: string | number; method: string; params: unknown };
@@ -135,6 +136,54 @@ describe('native readiness after Hub reconnect', () => {
    })
 })

+describe('async question delivery', () => {
+    const item = { type: 'agentMessage', id: 'async_call', delivery: 'async', text: 'Which client?',
+        questions: [{ title: 'Which client?', options: ['Web', 'Terminal'] }] };
+
+    it.each([false, true])('routes question answers through native user input (active: %s)', async active => {
+        const f = await fixture(); await f.root.activate();
+        if (active) f.native.notify('turn/started', { threadId: 'thread', turn: { id: 'turn', status: 'inProgress' } });
+        f.native.notify('item/completed', { threadId: 'thread', turnId: 'turn', item });
+        await f.drain();
+        const q = asyncQuestionsForItem('thread', item)[0];
+        expect(f.state().requests![q.id].tool).toBe('request_user_input');
+        const method = active ? 'turn/steer' : 'turn/start';
+        const original = f.root.client.request.bind(f.root.client);
+        const request = vi.spyOn(f.root.client, 'request').mockImplementation(async (name, params) => {
+            if (name !== method) return original(name, params);
+            const p = params as Record<string, unknown>;
+            f.native.notify('item/completed', { threadId: 'thread', turnId: 'answer-turn', item: {
+                type: 'userMessage', id: 'native-answer', clientId: p.clientUserMessageId, content: p.input
+            } });
+            return {};
+        });
+        try {
+            await f.rpc.get(RPC_METHODS.Permission)!({ id: q.id, approved: true, answers: { [q.questionItemId]: ['Web'] } });
+            await f.drain();
+            expect(request).toHaveBeenCalledWith(method, expect.objectContaining({ threadId: 'thread',
+                ...(active ? { expectedTurnId: 'turn' } : {}), input: [expect.objectContaining({ text: expect.stringContaining(q.questionItemId.replaceAll('"', '\\"')) })] }));
+            expect(f.native.responses).toEqual([]);
+            expect(f.state().requests![q.id]).toBeUndefined();
+            expect(f.state().completedRequests![q.id]).toMatchObject({ status: 'resolved' });
+        } finally { request.mockRestore(); }
+    });
+
+    it('retains unanswered questions across turn completion and native reconnect', async () => {
+        const f = await fixture(); await f.root.activate();
+        f.native.thread.turns = [{ id: 'turn', status: 'completed', items: [item] }];
+        f.native.notify('item/completed', { threadId: 'thread', turnId: 'turn', item });
+        f.native.notify('turn/completed', { threadId: 'thread', turn: f.native.thread.turns[0] });
+        await f.drain();
+        const q = asyncQuestionsForItem('thread', item)[0];
+        expect(f.state().requests![q.id]).toBeDefined();
+        f.native.initialized = false; f.native.abandoned();
+        await vi.waitFor(() => expect(f.native.initialized).toBe(true));
+        await f.root.refresh();
+        expect(f.state().requests![q.id]).toBeDefined();
+        expect(f.state().completedRequests?.[q.id]).toBeUndefined();
+    });
+});
+
 async function startCompaction(f: Awaited<ReturnType<typeof fixture>>, id = 'input', turnId = 'compacting-turn') {
    f.native.queue = [{ id: `native-${id}`, clientUserMessageId: id, input: [{ type: 'text', text: id }] }];
    await f.root.refresh();
@@ -16,6 +16,7 @@ import { resolveCodexPermissionModeConfig } from '../utils/permissionModeConfig'
 import { resolveCodexSlashCommand } from '../utils/slashCommands';
 import { parseReasoningEffortValue } from '../utils/reasoningEffort';
 import { SharedCodexPermissions } from './permissions';
+import { SharedCodexAsyncQuestions } from './asyncQuestions';
 import { SharedCodexQueue } from './queue';
 import { SharedCodexProjection, inputText } from './projection';
 import { getCodexSystemPrompt } from '../utils/systemPrompt';
@@ -46,6 +47,7 @@ export class SharedCodexRoot {
    bridge!: HapiMcpBridge;
    threadId = '';
    private permissions!: SharedCodexPermissions;
+    private asyncQuestions!: SharedCodexAsyncQuestions;
    private queue!: SharedCodexQueue;
    private projection!: SharedCodexProjection;
    private readonly children = new Map<string, SharedCodexProjection>();
@@ -161,9 +163,26 @@ export class SharedCodexRoot {
        this.bridge = await buildHapiMcpBridge(this.session, { exportSessionEnv: false, emitTitleSummary: false,
            skillLookup: { workingDirectory: this.bootstrap.workingDirectory, flavor: 'codex' } });
        await initializeSharedClient(this.client);
-        this.permissions = new SharedCodexPermissions(this.session, this.client, this.host.generation);
+        this.asyncQuestions = new SharedCodexAsyncQuestions(this.session, (threadId, text, id) => this.submitQuestionAnswer(threadId, text, id));
+        this.permissions = new SharedCodexPermissions(this.session, this.client, this.host.generation, reply => this.asyncQuestions.answer(reply));
        this.client.setServerRequestHandler(request => { void this.receiveRequest(request); });
    }
+    private async submitQuestionAnswer(threadId: string, text: string, clientUserMessageId: string): Promise<void> {
+        if (this.closed || this.stopping) throw new Error('Codex execution is stopping');
+        const operation = (async () => {
+            const thread = threadId === this.threadId ? undefined : await this.readThread(threadId);
+            const turnId = threadId === this.threadId ? this.currentTurn
+                : string((Array.isArray(thread?.turns) ? thread.turns.map(record) : []).find(turn => turn.status === 'inProgress')?.id);
+            const params = { threadId, clientUserMessageId, input: buildUserInputFromMessage(text) };
+            await this.client.request(turnId ? 'turn/steer' : 'turn/start', { ...params, ...(turnId ? { expectedTurnId: turnId } : {}) });
+        })();
+        this.controls.add(operation);
+        try { await operation; } finally { this.controls.delete(operation); }
+    }
+    private async projectHistory(projection: SharedCodexProjection, thread: unknown): Promise<void> {
+        this.asyncQuestions.history(projection.threadId, thread);
+        await projection.history(thread);
+    }
    private async receiveRequest(request: { id: string | number; method: string; params: unknown }): Promise<void> {
        await this.bound;
        if (this.closed || this.stopping) return;
@@ -207,7 +226,9 @@ export class SharedCodexRoot {
            (id, input) => this.session.syncNativeQueuedMessage(id, input === null ? null : inputText(input)),
            ids => this.session.setSteerDeliveryState(ids, 'queued'));
        await this.queue.load();
-        this.projection = new SharedCodexProjection(this.session, threadId, id => this.queue.committed(id), undefined, async () => record(await this.client.request('thread/read', { threadId, includeTurns: false })).thread);
+        this.projection = new SharedCodexProjection(this.session, threadId, id => this.queue.committed(id), undefined,
+            async () => record(await this.client.request('thread/read', { threadId, includeTurns: false })).thread,
+            (threadId, item) => this.asyncQuestions.observeItem(threadId, item));
        this.session.updateMetadata(metadata => ({ ...metadata, codexSessionId: threadId, capabilities: {
            ...metadata.capabilities, concurrentClients: true, terminal: true,
            // In-place rewind needs a native + hub commit barrier. Do not
@@ -221,7 +242,7 @@ export class SharedCodexRoot {
        // 空会话的首次订阅保留完整响应，以兼容原生历史源文件的初始化。
        if (subscribe) response = record(await this.client.request('thread/resume', { threadId }));
        this.acceptSettings(response); this.acceptSettings(this.host.settingsFor(threadId) ?? {});
-        await this.projection.history(response.thread); await this.refresh(); await this.refreshChildren(true);
+        await this.projectHistory(this.projection, response.thread); await this.refresh(); await this.refreshChildren(true);
    }
    async activate(options: SharedLaunchOptions = {}): Promise<void> {
        // Restored input cannot run before the cold-resume settings are applied.
@@ -292,7 +313,9 @@ export class SharedCodexRoot {
            if (!await this.ownsThread(eventThread)) return;
            let projection = this.children.get(eventThread);
            if (!projection) {
-                projection = new SharedCodexProjection(this.session, eventThread, async () => {}, this.threadId, async () => record(await this.client.request('thread/read', { threadId: eventThread, includeTurns: false })).thread);
+                projection = new SharedCodexProjection(this.session, eventThread, async () => {}, this.threadId,
+                    async () => record(await this.client.request('thread/read', { threadId: eventThread, includeTurns: false })).thread,
+                    (threadId, item) => this.asyncQuestions.observeItem(threadId, item));
                this.children.set(eventThread, projection);
            }
            await projection.notification(method, params); return;
@@ -363,7 +386,7 @@ export class SharedCodexRoot {
            const id = string(last?.id);
            this.latestTurn = id && last ? { id, status: string(last.status) ?? 'unknown', planId: planProposalForTurn(this.threadId, last) } : undefined;
        }
-        await this.projection.history(thread); await this.queue.reconcile(); this.alive();
+        await this.projectHistory(this.projection, thread); await this.queue.reconcile(); this.alive();
    }
    private async refreshChildren(subscribe: boolean): Promise<void> {
        // 历史回放可省略，子代理发现与实时订阅仍需在重连后恢复。
@@ -376,7 +399,8 @@ export class SharedCodexRoot {
                const thread = record(value); const id = string(thread.id);
                if (id && id !== this.threadId) this.ancestry.set(id, string(thread.parentThreadId) ?? null);
                if (id && id !== this.threadId && await this.ownsThread(id) && !this.children.has(id)) {
-                    this.children.set(id, new SharedCodexProjection(this.session, id, async () => {}, this.threadId));
+                    this.children.set(id, new SharedCodexProjection(this.session, id, async () => {}, this.threadId, undefined,
+                        (threadId, item) => this.asyncQuestions.observeItem(threadId, item)));
                }
            }
            cursor = string(page.nextCursor);
@@ -392,7 +416,7 @@ export class SharedCodexRoot {
            try {
                if (subscribe && loaded.has(id)) await this.client.request('thread/resume', { threadId: id, excludeTurns: true });
                if (replayHistory) {
-                    projection.reset(); await projection.history(await this.readThread(id));
+                    projection.reset(); await this.projectHistory(projection, await this.readThread(id));
                }
            } catch (error) { logger.debug('[Codex shared] child history unavailable', { id, error }); }
        }
@@ -413,7 +437,7 @@ export class SharedCodexRoot {
                try {
                    await initializeSharedClient(this.client);
                    if (this.stopping) return;
-                    this.permissions = new SharedCodexPermissions(this.session, this.client, `${this.host.generation}:${randomUUID()}`);
+                    this.permissions = new SharedCodexPermissions(this.session, this.client, `${this.host.generation}:${randomUUID()}`, reply => this.asyncQuestions.answer(reply));
                    this.client.setServerRequestHandler(request => { void this.receiveRequest(request); });
                    const settingsRevision = this.settingsRevision;
                    const response = record(await this.client.request('thread/resume', { threadId: this.threadId, excludeTurns: true }));
@@ -643,7 +667,7 @@ export class SharedCodexRoot {
    close(archived: boolean): Promise<void> {
        return this.closing ??= (async () => {
            this.stopAccepting(); this.closed = true; clearInterval(this.heartbeat);
-            this.permissions?.close();
+            this.permissions?.close(); this.asyncQuestions?.close();
            await this.client.disconnect(); await this.notifications; await this.queue?.flush(); this.bridge?.server.stop();
            if (archived) this.session.updateMetadata(metadata => ({ ...metadata, lifecycleState: 'archived', lifecycleStateSince: Date.now() }));
            // Inactive is resumable; uploads referenced by pending input must survive.
@@ -11,6 +11,7 @@ import { initializeSharedClient } from './launch';
 import { record } from './gateway';
 import { isProcessAlive } from '@/utils/process';
 import { notifyRunnerSessionStarted } from '@/runner/controlClient';
+import { asyncQuestionsForItem } from './asyncQuestions';

 const state = vi.hoisted(() => ({ home: '', sessions: new Map<string, MockSession>(), beforeBootstrap: undefined as (() => Promise<void>) | undefined }));
 const BOOTSTRAP_START_MARKER = vi.hoisted(() => 'Tue Sep 29 09:00:00 2026');
@@ -53,6 +54,76 @@ vi.mock('../utils/buildHapiMcpBridge', () => ({ buildHapiMcpBridge: async () =>
 // Explicit opt-in: installed official binary, entirely isolated CODEX_HOME, mock Responses API only.
 describe.skipIf(process.env.HAPI_RUN_SHARED_CODEX_TESTS !== '1')('installed Codex shared runtime', () => {
    afterEach(() => { vi.unstubAllEnvs(); state.sessions.clear(); state.beforeBootstrap = undefined; });
+    it('projects native async questions and delivers answers during generation and after completion', async () => {
+        const home = await mkdtemp('/tmp/hapi-shared-async-'); state.home = home;
+        const ch = join(home, 'codex'); const cwd = join(home, 'work'); await mkdir(ch); await mkdir(cwd);
+        const modelRequests: Record<string, unknown>[] = [];
+        let release!: () => void;
+        const gate = new Promise<void>(resolve => { release = resolve; });
+        const questions = [{ title: 'Which client?', options: ['Web', 'Terminal'] }, { title: 'Any details?' }];
+        const http = createServer((request, response) => {
+            const chunks: Buffer[] = [];
+            request.on('data', chunk => chunks.push(Buffer.from(chunk)));
+            request.on('end', () => { void (async () => {
+                if (!request.url?.endsWith('/responses')) { response.writeHead(404).end(); return; }
+                modelRequests.push(record(JSON.parse(Buffer.concat(chunks).toString())));
+                const count = modelRequests.length;
+                if (count === 2) await gate;
+                const id = randomUUID();
+                const item = count === 1 ? { type: 'function_call', call_id: 'call_async', name: 'request_user_input_async', arguments: JSON.stringify({ questions }) }
+                    : { type: 'message', role: 'assistant', id: `msg_${id}`, content: [{ type: 'output_text', text: 'MOCK DONE' }] };
+                const events = [{ type: 'response.created', response: { id } }, { type: 'response.output_item.done', item },
+                    { type: 'response.completed', response: { id, usage: { input_tokens: 0, output_tokens: 0, total_tokens: 0 } } }];
+                response.writeHead(200, { 'Content-Type': 'text/event-stream' });
+                response.end(events.map(event => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`).join(''));
+            })(); });
+        });
+        await new Promise<void>(resolve => http.listen(0, '127.0.0.1', resolve));
+        const port = (http.address() as { port: number }).port;
+        // This model's bundled tool catalog includes request_user_input_async.
+        // All Responses traffic goes to the isolated mock; no model credentials.
+        await writeFile(join(ch, 'config.toml'), `model = "gpt-6.1-sol"\nmodel_provider = "mock_provider"\napproval_policy = "never"\nsandbox_mode = "read-only"\ncheck_for_update_on_startup = false\n[features]\nsend_message_to_user_async = true\n[model_providers.mock_provider]\nname = "Isolated mock"\nbase_url = "http://127.0.0.1:${port}/v1"\nwire_api = "responses"\nrequires_openai_auth = false\nsupports_websockets = false\nrequest_max_retries = 0\nstream_max_retries = 0\n[analytics]\nenabled = false\n[feedback]\nenabled = false\n[projects."${cwd}"]\ntrust_level = "trusted"\n`);
+        vi.stubEnv('CODEX_HOME', ch);
+        const { runSharedRuntime } = await import('./runtime');
+        let ready!: (value: import('./runtime').RuntimeReady) => void;
+        const readiness = new Promise<import('./runtime').RuntimeReady>(resolve => { ready = resolve; });
+        const abort = new AbortController();
+        const running = runSharedRuntime({ workingDirectory: cwd }, ready, abort.signal);
+        let client: CodexAppServerClient | undefined;
+        try {
+            const { runtime, sessionId } = await Promise.race([readiness, running.then(() => { throw new Error('Runtime stopped before ready'); })]);
+            const threadId = runtime.sessions[sessionId].threadId;
+            const web = state.sessions.get(sessionId)!;
+            client = new CodexAppServerClient({ endpoint: runtime.endpoint, token: runtime.token });
+            const nativeRequests = vi.fn(); client.setServerRequestHandler(nativeRequests); await initializeSharedClient(client);
+            await client.request('thread/resume', { threadId });
+            web.user?.({ content: { text: 'ASK ASYNC' } }, 'async-prompt');
+            await vi.waitFor(() => expect(Object.keys(web.state.requests ?? {})).toHaveLength(2), { timeout: 15_000 });
+            expect(nativeRequests).not.toHaveBeenCalled();
+            const [choice, details] = asyncQuestionsForItem(threadId, { type: 'agentMessage', id: 'call_async', delivery: 'async', questions });
+            expect(web.messages).toContainEqual(expect.objectContaining({ type: 'tool-call', name: 'request_user_input', callId: choice.id }));
+            await vi.waitFor(() => expect(modelRequests.length).toBe(2));
+            await web.rpc.get('permission')!({ id: choice.id, approved: true, answers: { [choice.questionItemId]: ['Web'] } });
+            release();
+            await vi.waitFor(() => expect(web.state.completedRequests?.[choice.id]?.status).toBe('resolved'), { timeout: 15_000 });
+            await vi.waitFor(() => expect(web.state.steeringActive).toBe(false), { timeout: 15_000 });
+            expect(web.state.requests?.[details.id]).toBeDefined();
+            await web.rpc.get('permission')!({ id: details.id, approved: true, answers: { [details.questionItemId]: ['user_note: From mobile'] } });
+            await vi.waitFor(() => expect(web.state.completedRequests?.[details.id]?.status).toBe('resolved'), { timeout: 15_000 });
+            await vi.waitFor(() => expect(web.messages).toContainEqual({ user: 'Any details?\nFrom mobile' }));
+            await vi.waitFor(() => expect(JSON.stringify(modelRequests)).toContain('From mobile'), { timeout: 15_000 });
+            expect(JSON.stringify(modelRequests)).toContain('send_user_message_question_reply');
+            const history = record(record(await client.request('thread/read', { threadId, includeTurns: true })).thread);
+            const items = (history.turns as Array<{ items: unknown[] }>).flatMap(turn => turn.items);
+            expect(items).toContainEqual(expect.objectContaining({ type: 'agentMessage', id: 'call_async', delivery: 'async',
+                questions: [{ title: 'Which client?', options: ['Web', 'Terminal'] }, { title: 'Any details?', options: null }] }));
+            expect(Object.keys(web.state.requests ?? {})).toHaveLength(0);
+        } finally {
+            release(); abort.abort(); await running.catch(() => {}); await client?.disconnect();
+            await new Promise<void>(resolve => http.close(() => resolve()));
+            await rm(home, { recursive: true, force: true, maxRetries: 3 });
+        }
+    }, 60_000);
    it('rejects a child ID before creating a HAPI binding or calling native resume', async () => {
        const home = await mkdtemp('/tmp/hapi-shared-child-'); state.home = home;
        const ch = join(home, 'codex'); await mkdir(ch);