diff --git a/cli/src/codex/shared/asyncQuestions.test.ts b/cli/src/codex/shared/asyncQuestions.test.ts new file mode 100644 index 00000000..1aee31a0 --- /dev/null +++ b/cli/src/codex/shared/asyncQuestions.test.ts @@ -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: + `\n${JSON.stringify([{ questionItemId, answer, question: 'Which client?' }])}\n` }] }; +} +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', + `\n${JSON.stringify([{ answer, question: q.title, questionItemId: q.questionItemId }])}\n`, `${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 ['bad json', + `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 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({}); + }); +}); diff --git a/cli/src/codex/shared/asyncQuestions.ts b/cli/src/codex/shared/asyncQuestions.ts new file mode 100644 index 00000000..07f2923d --- /dev/null +++ b/cli/src/codex/shared/asyncQuestions.ts @@ -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 = ''; + const closing = ''; + 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(); + private readonly questions = new Map(); + private readonly completed = new Map(); + private closed = false; + + constructor(private readonly session: ApiSessionClient, + private readonly submit: (threadId: string, text: string, clientUserMessageId: string) => Promise) {} + + 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 { + 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 = `\n${JSON.stringify([{ + answer, question: question.title, questionItemId: question.questionItemId + }])}\n`; + 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); + } +} diff --git a/cli/src/codex/shared/permissions.ts b/cli/src/codex/shared/permissions.ts index f8466f6c..2f4cdcf5 100644 --- a/cli/src/codex/shared/permissions.ts +++ b/cli/src/codex/shared/permissions.ts @@ -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; +export type PermissionReply = z.infer; +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) { 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; diff --git a/cli/src/codex/shared/projection.test.ts b/cli/src/codex/shared/projection.test.ts index 7009181a..9394d111 100644 --- a/cli/src/codex/shared/projection.test.ts +++ b/cli/src/codex/shared/projection.test.ts @@ -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 = `\n${JSON.stringify([{ + questionItemId: '["request_user_input_async","call_async",0]', question: 'Which client?', answer: 'Web' + }])}\n`; + 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; diff --git a/cli/src/codex/shared/projection.ts b/cli/src/codex/shared/projection.ts index 0736c1ad..cade3ec5 100644 --- a/cli/src/codex/shared/projection.ts +++ b/cli/src/codex/shared/projection.ts @@ -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, private readonly parentThreadId?: string, - private readonly readSummaryThread?: () => Promise) { + private readonly readSummaryThread?: () => Promise, + 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, 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)) { diff --git a/cli/src/codex/shared/root.test.ts b/cli/src/codex/shared/root.test.ts index e61b1aa8..320f1686 100644 --- a/cli/src/codex/shared/root.test.ts +++ b/cli/src/codex/shared/root.test.ts @@ -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; + 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>, id = 'input', turnId = 'compacting-turn') { f.native.queue = [{ id: `native-${id}`, clientUserMessageId: id, input: [{ type: 'text', text: id }] }]; await f.root.refresh(); diff --git a/cli/src/codex/shared/root.ts b/cli/src/codex/shared/root.ts index 1bf18055..ffd3c92a 100644 --- a/cli/src/codex/shared/root.ts +++ b/cli/src/codex/shared/root.ts @@ -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(); @@ -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 { + 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 { + this.asyncQuestions.history(projection.threadId, thread); + await projection.history(thread); + } private async receiveRequest(request: { id: string | number; method: string; params: unknown }): Promise { 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 { // 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 { // 历史回放可省略，子代理发现与实时订阅仍需在重连后恢复。 @@ -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 { 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. diff --git a/cli/src/codex/shared/runtime.integration.test.ts b/cli/src/codex/shared/runtime.integration.test.ts index 9134b5b7..ce937506 100644 --- a/cli/src/codex/shared/runtime.integration.test.ts +++ b/cli/src/codex/shared/runtime.integration.test.ts @@ -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(), beforeBootstrap: undefined as (() => Promise) | 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[] = []; + let release!: () => void; + const gate = new Promise(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(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(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(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); diff --git a/docs/guide/codex-shared-sessions.md b/docs/guide/codex-shared-sessions.md index 52668a1d..84c27d25 100644 --- a/docs/guide/codex-shared-sessions.md +++ b/docs/guide/codex-shared-sessions.md @@ -2,7 +2,8 @@ Codex **0.154.0 or newer**: the official terminal UI and HAPI Web / native clients share one app-server. No local/remote ownership switch. Both can send -messages, approve tools and answer `request_user_input`. +messages, approve tools and answer `request_user_input`. Codex versions with +`request_user_input_async` also expose their queued follow-up questions in HAPI. ```bash hapi codex # launch an execution owned by this terminal @@ -77,13 +78,21 @@ confirms consumption. Compaction completion, turn completion/interruption or a native disconnect withdraws this temporary status. Unconfirmed input retains the existing unknown-delivery recovery controls and is never automatically replayed. -Codex arbitrates terminal/Web answers. A submitted Web answer is only a +For blocking `request_user_input`, Codex arbitrates terminal/Web answers. A submitted Web answer is only a candidate. Native `serverRequest/resolved` closes the prompt with neutral **Resolved in Codex** status; HAPI does not invent the winner or their answer. Disconnect withdraws the old controls; reconnect replays native pending requests. -Canceling a question interrupts its native turn instead of sending an empty +Canceling a blocking question interrupts its native turn instead of sending an empty fabricated answer. Unknown side-client request types remain terminal-only. +`request_user_input_async` uses a separate path: Codex publishes structured +questions in an async assistant message and continues working. HAPI shows each +question as an answerable card, including choices and free text. Answers are +sent as native user input during the active turn or in a new turn when idle; +native input/history confirms completion. Unanswered questions survive turn +completion and reconnect. Dismissing one in HAPI does not interrupt Codex or +dismiss the terminal's local question. + Each root has its own HAPI socket, MCP bridge and shell `HAPI_SESSION_ID`. Child-agent traces/permissions are attributed by native ancestry, not merely by being received on the same app-server connection.