From 10f7e1404e02a00d165b20b545245fda5627b3d2 Mon Sep 17 00:00:00 2001 From: Lihengwannafly <624260029@qq.com> Date: Wed, 25 Feb 2026 16:57:28 +0800 Subject: [PATCH] fix(cli): harden ACP/Codex event handling (#211) Co-authored-by: Lihengwannafly --- .../backends/acp/AcpMessageHandler.test.ts | 274 ++++++++++++++++++ .../agent/backends/acp/AcpMessageHandler.ts | 105 ++++++- .../agent/backends/acp/AcpSdkBackend.test.ts | 104 +++++++ cli/src/agent/backends/acp/AcpSdkBackend.ts | 71 ++++- cli/src/agent/messageConverter.test.ts | 38 +++ cli/src/agent/messageConverter.ts | 21 +- cli/src/agent/utils.test.ts | 26 ++ cli/src/agent/utils.ts | 67 +++-- cli/src/codex/codexRemoteLauncher.test.ts | 177 +++++++++++ cli/src/codex/codexRemoteLauncher.ts | 154 +++++++++- cli/src/codex/utils/appServerConfig.test.ts | 22 +- cli/src/codex/utils/appServerConfig.ts | 11 +- .../utils/appServerEventConverter.test.ts | 177 +++++++++++ .../codex/utils/appServerEventConverter.ts | 221 +++++++++++++- cli/src/codex/utils/systemPrompt.ts | 5 +- .../codex/utils/terminalEventGuard.test.ts | 94 ++++++ cli/src/codex/utils/terminalEventGuard.ts | 29 ++ 17 files changed, 1529 insertions(+), 67 deletions(-) create mode 100644 cli/src/agent/backends/acp/AcpMessageHandler.test.ts create mode 100644 cli/src/agent/backends/acp/AcpSdkBackend.test.ts create mode 100644 cli/src/agent/messageConverter.test.ts create mode 100644 cli/src/agent/utils.test.ts create mode 100644 cli/src/codex/codexRemoteLauncher.test.ts create mode 100644 cli/src/codex/utils/terminalEventGuard.test.ts create mode 100644 cli/src/codex/utils/terminalEventGuard.ts diff --git a/cli/src/agent/backends/acp/AcpMessageHandler.test.ts b/cli/src/agent/backends/acp/AcpMessageHandler.test.ts new file mode 100644 index 00000000..40f3e2ac --- /dev/null +++ b/cli/src/agent/backends/acp/AcpMessageHandler.test.ts @@ -0,0 +1,274 @@ +import { describe, expect, it } from 'vitest'; +import type { AgentMessage } from '@/agent/types'; +import { AcpMessageHandler } from './AcpMessageHandler'; +import { ACP_SESSION_UPDATE_TYPES } from './constants'; + +function getToolResult(messages: AgentMessage[], id: string): Extract { + const result = messages.find((message): message is Extract => + message.type === 'tool_result' && message.id === id + ); + if (!result) { + throw new Error(`Missing tool_result for ${id}`); + } + return result; +} + +describe('AcpMessageHandler', () => { + it('does not synthesize {status} output when tool completes without payload', () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCall, + toolCallId: 'tool-1', + title: 'Read', + rawInput: { path: 'README.md' }, + status: 'in_progress' + }); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCallUpdate, + toolCallId: 'tool-1', + status: 'completed' + }); + + const result = getToolResult(messages, 'tool-1'); + expect(result.status).toBe('completed'); + expect(result.output).toBeUndefined(); + }); + + it('keeps raw output when provided by ACP update', () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCall, + toolCallId: 'tool-2', + title: 'Bash', + rawInput: { cmd: 'echo ok' }, + status: 'in_progress' + }); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCallUpdate, + toolCallId: 'tool-2', + status: 'completed', + rawOutput: { stdout: 'ok\n' } + }); + + const result = getToolResult(messages, 'tool-2'); + expect(result.status).toBe('completed'); + expect(result.output).toEqual({ stdout: 'ok\n' }); + }); + + it('keeps buffered text behind tool lifecycle events', () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'final answer' } + }); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCall, + toolCallId: 'tool-3', + title: 'Read', + rawInput: { path: 'README.md' }, + status: 'in_progress' + }); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCallUpdate, + toolCallId: 'tool-3', + status: 'completed', + rawOutput: { content: 'ok' } + }); + + handler.flushText(); + + expect(messages.map((message) => message.type)).toEqual(['tool_call', 'tool_result', 'text']); + const textMessage = messages[messages.length - 1]; + expect(textMessage).toEqual({ type: 'text', text: 'final answer' }); + }); + + it('ignores text chunks targeted only to user audience', () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { + type: 'text', + text: 'user-visible only', + annotations: { + audience: ['user'] + } + } + }); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { + type: 'text', + text: 'assistant-visible', + annotations: { + audience: ['assistant'] + } + } + }); + + handler.flushText(); + + expect(messages).toEqual([{ type: 'text', text: 'assistant-visible' }]); + }); + + it('supports annotations array format for audience filtering', () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { + type: 'text', + text: 'user-only', + annotations: [ + { + audience: ['user'] + } + ] + } + }); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { + type: 'text', + text: 'assistant-only', + annotations: [ + { + audience: ['assistant'] + } + ] + } + }); + + handler.flushText(); + + expect(messages).toEqual([{ type: 'text', text: 'assistant-only' }]); + }); + + it('supports annotations object value.audience format for filtering', () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { + type: 'text', + text: 'user-only', + annotations: { + value: { + audience: ['user'] + } + } + } + }); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { + type: 'text', + text: 'assistant-only', + annotations: { + value: { + audience: ['assistant'] + } + } + } + }); + + handler.flushText(); + + expect(messages).toEqual([{ type: 'text', text: 'assistant-only' }]); + }); + + it('deduplicates overlapping text chunks', () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'hello wo' } + }); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'world' } + }); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'world' } + }); + + handler.flushText(); + + expect(messages).toEqual([{ type: 'text', text: 'hello world' }]); + }); + + it('keeps existing tool name when update only has kind fallback', () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCall, + toolCallId: 'tool-4', + title: 'hapi_change_title', + rawInput: { title: 'A' }, + status: 'in_progress' + }); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCallUpdate, + toolCallId: 'tool-4', + kind: 'other', + rawInput: { title: 'B' }, + status: 'in_progress' + }); + + const calls = messages.filter((message): message is Extract => + message.type === 'tool_call' + ); + expect(calls).toHaveLength(2); + expect(calls[0].name).toBe('hapi_change_title'); + expect(calls[1].name).toBe('hapi_change_title'); + }); + + it('allows kind fallback to replace placeholder tool name', () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCall, + toolCallId: 'tool-5', + rawInput: { foo: 'bar' }, + status: 'in_progress' + }); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCallUpdate, + toolCallId: 'tool-5', + kind: 'search', + rawInput: { foo: 'baz' }, + status: 'in_progress' + }); + + const calls = messages.filter((message): message is Extract => + message.type === 'tool_call' + ); + expect(calls).toHaveLength(2); + expect(calls[0].name).toBe('Tool'); + expect(calls[1].name).toBe('search'); + }); +}); diff --git a/cli/src/agent/backends/acp/AcpMessageHandler.ts b/cli/src/agent/backends/acp/AcpMessageHandler.ts index de469a45..2b80439e 100644 --- a/cli/src/agent/backends/acp/AcpMessageHandler.ts +++ b/cli/src/agent/backends/acp/AcpMessageHandler.ts @@ -1,6 +1,6 @@ import type { AgentMessage, PlanItem } from '@/agent/types'; import { asString, isObject } from '@hapi/protocol'; -import { deriveToolName } from '@/agent/utils'; +import { deriveToolNameWithSource, isPlaceholderToolName } from '@/agent/utils'; import { ACP_SESSION_UPDATE_TYPES } from './constants'; function normalizeStatus(status: unknown): 'pending' | 'in_progress' | 'completed' | 'failed' { @@ -10,8 +10,10 @@ function normalizeStatus(status: unknown): 'pending' | 'in_progress' | 'complete return 'pending'; } -function deriveToolNameFromUpdate(update: Record): string { - return deriveToolName({ +type DerivedToolName = ReturnType; + +function deriveToolNameFromUpdate(update: Record): DerivedToolName { + return deriveToolNameWithSource({ title: asString(update.title), kind: asString(update.kind), rawInput: update.rawInput @@ -21,10 +23,51 @@ function deriveToolNameFromUpdate(update: Record): string { function extractTextContent(block: unknown): string | null { if (!isObject(block)) return null; if (block.type !== 'text') return null; + const explicitAudience = extractExplicitAudience(block.annotations); + if (explicitAudience.length > 0 && !explicitAudience.includes('assistant')) { + return null; + } const text = block.text; return typeof text === 'string' ? text : null; } +function extractExplicitAudience(annotations: unknown): string[] { + if (Array.isArray(annotations)) { + const audiences: string[] = []; + for (const entry of annotations) { + if (typeof entry === 'string') { + audiences.push(entry); + continue; + } + if (!isObject(entry)) { + continue; + } + audiences.push(...extractAudienceField(entry.audience)); + if (isObject(entry.value)) { + audiences.push(...extractAudienceField(entry.value.audience)); + } + } + return audiences; + } + if (isObject(annotations)) { + return [ + ...extractAudienceField(annotations.audience), + ...(isObject(annotations.value) ? extractAudienceField(annotations.value.audience) : []) + ]; + } + return []; +} + +function extractAudienceField(value: unknown): string[] { + if (typeof value === 'string') { + return [value]; + } + if (!Array.isArray(value)) { + return []; + } + return value.filter((entry): entry is string => typeof entry === 'string'); +} + function normalizePlanEntries(entries: unknown): PlanItem[] { if (!Array.isArray(entries)) return []; @@ -45,6 +88,16 @@ function normalizePlanEntries(entries: unknown): PlanItem[] { return items; } +function getSuffixPrefixOverlap(base: string, next: string): number { + const maxOverlap = Math.min(base.length, next.length); + for (let length = maxOverlap; length > 0; length -= 1) { + if (base.endsWith(next.slice(0, length))) { + return length; + } + } + return 0; +} + export class AcpMessageHandler { private readonly toolCalls = new Map(); private bufferedText = ''; @@ -77,6 +130,20 @@ export class AcpMessageHandler { if (this.bufferedText.startsWith(text)) { return; } + if (this.bufferedText.endsWith(text)) { + return; + } + if (text.endsWith(this.bufferedText)) { + this.bufferedText = text; + return; + } + + const overlap = getSuffixPrefixOverlap(this.bufferedText, text); + if (overlap > 0) { + this.bufferedText += text.slice(overlap); + return; + } + this.bufferedText += text; } @@ -99,19 +166,16 @@ export class AcpMessageHandler { } if (updateType === ACP_SESSION_UPDATE_TYPES.toolCall) { - this.flushText(); this.handleToolCall(update); return; } if (updateType === ACP_SESSION_UPDATE_TYPES.toolCallUpdate) { - this.flushText(); this.handleToolCallUpdate(update); return; } if (updateType === ACP_SESSION_UPDATE_TYPES.plan) { - this.flushText(); const items = normalizePlanEntries(update.entries); if (items.length > 0) { this.onMessage({ type: 'plan', items }); @@ -123,7 +187,8 @@ export class AcpMessageHandler { const toolCallId = asString(update.toolCallId); if (!toolCallId) return; - const name = deriveToolNameFromUpdate(update); + const derivedName = deriveToolNameFromUpdate(update); + const name = derivedName.name; const input = update.rawInput ?? null; const status = normalizeStatus(update.status); @@ -146,7 +211,8 @@ export class AcpMessageHandler { const existing = this.toolCalls.get(toolCallId); if (update.rawInput !== undefined) { - const name = deriveToolNameFromUpdate(update); + const derivedName = deriveToolNameFromUpdate(update); + const name = this.selectToolNameForUpdate(existing?.name ?? null, derivedName); const input = update.rawInput; this.toolCalls.set(toolCallId, { name, input }); this.onMessage({ @@ -167,8 +233,7 @@ export class AcpMessageHandler { } if (status === 'completed' || status === 'failed') { - const output = update.rawOutput ?? update.content; - const result = output ?? { status }; + const result = update.rawOutput ?? update.content; this.onMessage({ type: 'tool_result', id: toolCallId, @@ -177,4 +242,24 @@ export class AcpMessageHandler { }); } } + + private selectToolNameForUpdate(existingName: string | null, derivedName: DerivedToolName): string { + if (!existingName) { + return derivedName.name; + } + + if ( + derivedName.source === 'title' || + derivedName.source === 'raw_input_name' || + derivedName.source === 'raw_input_tool' + ) { + return derivedName.name; + } + + if (isPlaceholderToolName(existingName)) { + return derivedName.name; + } + + return existingName; + } } diff --git a/cli/src/agent/backends/acp/AcpSdkBackend.test.ts b/cli/src/agent/backends/acp/AcpSdkBackend.test.ts new file mode 100644 index 00000000..5bc86812 --- /dev/null +++ b/cli/src/agent/backends/acp/AcpSdkBackend.test.ts @@ -0,0 +1,104 @@ +import { afterEach, describe, expect, it } from 'vitest'; +import type { AgentMessage } from '@/agent/types'; +import { AcpSdkBackend } from './AcpSdkBackend'; +import { ACP_SESSION_UPDATE_TYPES } from './constants'; + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +type BackendStatics = { + UPDATE_QUIET_PERIOD_MS: number; + UPDATE_DRAIN_TIMEOUT_MS: number; + PRE_PROMPT_UPDATE_QUIET_PERIOD_MS: number; + PRE_PROMPT_UPDATE_DRAIN_TIMEOUT_MS: number; +}; + +const backendStatics = AcpSdkBackend as unknown as BackendStatics; +const originalStatics = { + updateQuietPeriodMs: backendStatics.UPDATE_QUIET_PERIOD_MS, + updateDrainTimeoutMs: backendStatics.UPDATE_DRAIN_TIMEOUT_MS, + prePromptUpdateQuietPeriodMs: backendStatics.PRE_PROMPT_UPDATE_QUIET_PERIOD_MS, + prePromptUpdateDrainTimeoutMs: backendStatics.PRE_PROMPT_UPDATE_DRAIN_TIMEOUT_MS +}; + +afterEach(() => { + backendStatics.UPDATE_QUIET_PERIOD_MS = originalStatics.updateQuietPeriodMs; + backendStatics.UPDATE_DRAIN_TIMEOUT_MS = originalStatics.updateDrainTimeoutMs; + backendStatics.PRE_PROMPT_UPDATE_QUIET_PERIOD_MS = originalStatics.prePromptUpdateQuietPeriodMs; + backendStatics.PRE_PROMPT_UPDATE_DRAIN_TIMEOUT_MS = originalStatics.prePromptUpdateDrainTimeoutMs; +}); + +describe('AcpSdkBackend', () => { + it('emits turn_complete after trailing tool updates from the same turn', async () => { + backendStatics.UPDATE_QUIET_PERIOD_MS = 8; + backendStatics.UPDATE_DRAIN_TIMEOUT_MS = 200; + backendStatics.PRE_PROMPT_UPDATE_QUIET_PERIOD_MS = 1; + backendStatics.PRE_PROMPT_UPDATE_DRAIN_TIMEOUT_MS = 50; + + const backend = new AcpSdkBackend({ command: 'opencode' }); + const backendInternal = backend as unknown as { + transport: { + sendRequest: (...args: unknown[]) => Promise; + close: () => Promise; + } | null; + handleSessionUpdate: (params: unknown) => void; + }; + + const messages: AgentMessage[] = []; + backendInternal.transport = { + sendRequest: async () => { + setTimeout(() => { + backendInternal.handleSessionUpdate({ + sessionId: 'session-1', + update: { + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'final answer' } + } + }); + }, 0); + + await sleep(5); + + setTimeout(() => { + backendInternal.handleSessionUpdate({ + sessionId: 'session-1', + update: { + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCall, + toolCallId: 'tool-1', + title: 'Read', + rawInput: { path: 'README.md' }, + status: 'in_progress' + } + }); + }, 3); + + setTimeout(() => { + backendInternal.handleSessionUpdate({ + sessionId: 'session-1', + update: { + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCallUpdate, + toolCallId: 'tool-1', + status: 'completed', + rawOutput: { ok: true } + } + }); + }, 6); + + return { stopReason: 'end_turn' }; + }, + close: async () => {} + }; + + await backend.prompt('session-1', [{ type: 'text', text: 'hello' }], (message) => { + messages.push(message); + }); + + expect(messages.map((message) => message.type)).toEqual([ + 'tool_call', + 'tool_result', + 'text', + 'turn_complete' + ]); + }); +}); diff --git a/cli/src/agent/backends/acp/AcpSdkBackend.ts b/cli/src/agent/backends/acp/AcpSdkBackend.ts index dac226aa..aa797e96 100644 --- a/cli/src/agent/backends/acp/AcpSdkBackend.ts +++ b/cli/src/agent/backends/acp/AcpSdkBackend.ts @@ -19,6 +19,7 @@ export class AcpSdkBackend implements AgentBackend { private activeSessionId: string | null = null; private isProcessingMessage = false; private responseCompleteResolvers: Array<() => void> = []; + private lastSessionUpdateAt = 0; /** Retry configuration for ACP initialization */ private static readonly INIT_RETRY_OPTIONS = { @@ -26,6 +27,10 @@ export class AcpSdkBackend implements AgentBackend { minDelay: 1000, maxDelay: 5000 }; + private static readonly UPDATE_QUIET_PERIOD_MS = 120; + private static readonly UPDATE_DRAIN_TIMEOUT_MS = 2000; + private static readonly PRE_PROMPT_UPDATE_QUIET_PERIOD_MS = 200; + private static readonly PRE_PROMPT_UPDATE_DRAIN_TIMEOUT_MS = 1200; constructor(private readonly options: { command: string; args?: string[]; env?: Record }) {} @@ -141,8 +146,20 @@ export class AcpSdkBackend implements AgentBackend { } this.activeSessionId = sessionId; + await this.waitForSessionUpdateQuiet( + AcpSdkBackend.PRE_PROMPT_UPDATE_QUIET_PERIOD_MS, + AcpSdkBackend.PRE_PROMPT_UPDATE_DRAIN_TIMEOUT_MS + ); + this.messageHandler?.flushText(); + this.messageHandler = null; + await this.waitForSessionUpdateQuiet( + AcpSdkBackend.PRE_PROMPT_UPDATE_QUIET_PERIOD_MS, + AcpSdkBackend.PRE_PROMPT_UPDATE_DRAIN_TIMEOUT_MS + ); this.messageHandler = new AcpMessageHandler(onUpdate); this.isProcessingMessage = true; + this.lastSessionUpdateAt = Date.now(); + let stopReason: string | null = null; try { // No timeout for prompt requests - they can run for extended periods @@ -152,16 +169,21 @@ export class AcpSdkBackend implements AgentBackend { prompt: content }, { timeoutMs: Infinity }); - const stopReason = isObject(response) ? asString(response.stopReason) : null; - if (stopReason) { - this.messageHandler?.flushText(); - onUpdate({ type: 'turn_complete', stopReason }); - } + stopReason = isObject(response) ? asString(response.stopReason) : null; } finally { + await this.waitForSessionUpdateQuiet( + AcpSdkBackend.UPDATE_QUIET_PERIOD_MS, + AcpSdkBackend.UPDATE_DRAIN_TIMEOUT_MS + ); this.messageHandler?.flushText(); - this.messageHandler = null; - this.isProcessingMessage = false; - this.notifyResponseComplete(); + try { + if (stopReason) { + onUpdate({ type: 'turn_complete', stopReason }); + } + } finally { + this.isProcessingMessage = false; + this.notifyResponseComplete(); + } } } @@ -215,6 +237,10 @@ export class AcpSdkBackend implements AgentBackend { return this.isProcessingMessage; } + getLastSessionUpdateAt(): number { + return this.lastSessionUpdateAt; + } + /** * Wait for any in-progress response to complete. * Resolves immediately if no response is being processed. @@ -232,6 +258,11 @@ export class AcpSdkBackend implements AgentBackend { async disconnect(): Promise { if (!this.transport) return; + this.messageHandler?.flushText(); + this.messageHandler = null; + this.activeSessionId = null; + this.isProcessingMessage = false; + this.notifyResponseComplete(); await this.transport.close(); this.transport = null; } @@ -242,9 +273,29 @@ export class AcpSdkBackend implements AgentBackend { if (this.activeSessionId && sessionId && sessionId !== this.activeSessionId) { return; } + this.lastSessionUpdateAt = Date.now(); const update = params.update; - if (!this.messageHandler) return; - this.messageHandler.handleUpdate(update); + this.messageHandler?.handleUpdate(update); + } + + private async waitForSessionUpdateQuiet(quietMs: number, timeoutMs: number): Promise { + if (quietMs <= 0 || timeoutMs <= 0) { + return; + } + + const deadline = Date.now() + timeoutMs; + + while (Date.now() < deadline) { + const elapsedSinceUpdate = Date.now() - this.lastSessionUpdateAt; + if (elapsedSinceUpdate >= quietMs) { + return; + } + + const remainingToQuiet = quietMs - elapsedSinceUpdate; + const remainingBudget = deadline - Date.now(); + const waitMs = Math.max(1, Math.min(remainingToQuiet, remainingBudget)); + await new Promise((resolve) => setTimeout(resolve, waitMs)); + } } private async handlePermissionRequest(params: unknown, requestId: string | number | null): Promise { diff --git a/cli/src/agent/messageConverter.test.ts b/cli/src/agent/messageConverter.test.ts new file mode 100644 index 00000000..97dae110 --- /dev/null +++ b/cli/src/agent/messageConverter.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from 'vitest'; +import { convertAgentMessage } from './messageConverter'; + +describe('convertAgentMessage', () => { + it('keeps tool-call status when converting ACP tool events', () => { + const converted = convertAgentMessage({ + type: 'tool_call', + id: 'call-1', + name: 'Bash', + input: { cmd: 'echo test' }, + status: 'completed' + }); + + expect(converted).toEqual({ + type: 'tool-call', + callId: 'call-1', + name: 'Bash', + input: { cmd: 'echo test' }, + status: 'completed' + }); + }); + + it('marks failed tool results as error', () => { + const converted = convertAgentMessage({ + type: 'tool_result', + id: 'call-2', + output: { message: 'boom' }, + status: 'failed' + }); + + expect(converted).toEqual({ + type: 'tool-call-result', + callId: 'call-2', + output: { message: 'boom' }, + is_error: true + }); + }); +}); diff --git a/cli/src/agent/messageConverter.ts b/cli/src/agent/messageConverter.ts index d3b95a22..9bf8d6a1 100644 --- a/cli/src/agent/messageConverter.ts +++ b/cli/src/agent/messageConverter.ts @@ -2,8 +2,19 @@ import type { AgentMessage, PlanItem } from './types'; export type CodexMessage = | { type: 'message'; message: string } - | { type: 'tool-call'; name: string; callId: string; input: unknown } - | { type: 'tool-call-result'; callId: string; output: unknown } + | { + type: 'tool-call'; + name: string; + callId: string; + input: unknown; + status?: 'pending' | 'in_progress' | 'completed' | 'failed'; + } + | { + type: 'tool-call-result'; + callId: string; + output: unknown; + is_error?: boolean; + } | { type: 'plan'; entries: PlanItem[] } | { type: 'error'; message: string }; @@ -16,13 +27,15 @@ export function convertAgentMessage(message: AgentMessage): CodexMessage | null type: 'tool-call', name: message.name, callId: message.id, - input: message.input + input: message.input, + status: message.status }; case 'tool_result': return { type: 'tool-call-result', callId: message.id, - output: message.output + output: message.output, + is_error: message.status === 'failed' }; case 'plan': return { diff --git a/cli/src/agent/utils.test.ts b/cli/src/agent/utils.test.ts new file mode 100644 index 00000000..763ece85 --- /dev/null +++ b/cli/src/agent/utils.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from 'vitest'; +import { deriveToolName, deriveToolNameWithSource, isPlaceholderToolName } from './utils'; + +describe('agent tool name helpers', () => { + it('treats generic kind fallback as placeholder', () => { + expect(deriveToolName({ kind: 'other' })).toBe('Tool'); + expect(deriveToolName({ kind: 'unknown' })).toBe('Tool'); + }); + + it('keeps source metadata for explicit raw input names', () => { + const derived = deriveToolNameWithSource({ + kind: 'execute', + rawInput: { name: 'Bash' } + }); + expect(derived).toEqual({ + name: 'Bash', + source: 'raw_input_name' + }); + }); + + it('marks placeholder tool names', () => { + expect(isPlaceholderToolName('other')).toBe(true); + expect(isPlaceholderToolName('tool')).toBe(true); + expect(isPlaceholderToolName('search')).toBe(false); + }); +}); diff --git a/cli/src/agent/utils.ts b/cli/src/agent/utils.ts index 4f9830fb..f529332b 100644 --- a/cli/src/agent/utils.ts +++ b/cli/src/agent/utils.ts @@ -1,29 +1,54 @@ import { isObject } from '@hapi/protocol'; +type ToolNameSource = 'title' | 'raw_input_name' | 'raw_input_tool' | 'kind' | 'default'; + +function normalizeToolName(value: unknown): string | null { + if (typeof value !== 'string') { + return null; + } + const trimmed = value.trim(); + return trimmed.length > 0 ? trimmed : null; +} + +export function isPlaceholderToolName(name: string): boolean { + const normalized = name.trim().toLowerCase(); + return normalized === '' || normalized === 'tool' || normalized === 'unknown' || normalized === 'other'; +} + +export function deriveToolNameWithSource(input: { + title?: string | null; + kind?: string | null; + rawInput?: unknown; +}): { name: string; source: ToolNameSource } { + const title = normalizeToolName(input.title); + if (title) { + return { name: title, source: 'title' }; + } + + if (isObject(input.rawInput)) { + const fromName = normalizeToolName(input.rawInput.name); + if (fromName) { + return { name: fromName, source: 'raw_input_name' }; + } + + const fromTool = normalizeToolName(input.rawInput.tool); + if (fromTool) { + return { name: fromTool, source: 'raw_input_tool' }; + } + } + + const kind = normalizeToolName(input.kind); + if (kind && !isPlaceholderToolName(kind)) { + return { name: kind, source: 'kind' }; + } + + return { name: 'Tool', source: 'default' }; +} + export function deriveToolName(input: { title?: string | null; kind?: string | null; rawInput?: unknown; }): string { - if (input.title && input.title.trim().length > 0) { - return input.title.trim(); - } - - if (isObject(input.rawInput)) { - const fromName = input.rawInput.name; - if (typeof fromName === 'string' && fromName.trim().length > 0) { - return fromName.trim(); - } - - const fromTool = input.rawInput.tool; - if (typeof fromTool === 'string' && fromTool.trim().length > 0) { - return fromTool.trim(); - } - } - - if (input.kind && input.kind.trim().length > 0) { - return input.kind.trim(); - } - - return 'Tool'; + return deriveToolNameWithSource(input).name; } diff --git a/cli/src/codex/codexRemoteLauncher.test.ts b/cli/src/codex/codexRemoteLauncher.test.ts new file mode 100644 index 00000000..5d3ca2eb --- /dev/null +++ b/cli/src/codex/codexRemoteLauncher.test.ts @@ -0,0 +1,177 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { MessageQueue2 } from '@/utils/MessageQueue2'; +import type { EnhancedMode } from './loop'; + +const harness = vi.hoisted(() => ({ + notifications: [] as Array<{ method: string; params: unknown }>, + registerRequestCalls: [] as string[] +})); + +vi.mock('./codexAppServerClient', () => { + class MockCodexAppServerClient { + private notificationHandler: ((method: string, params: unknown) => void) | null = null; + + async connect(): Promise {} + + async initialize(): Promise<{ protocolVersion: number }> { + return { protocolVersion: 1 }; + } + + setNotificationHandler(handler: ((method: string, params: unknown) => void) | null): void { + this.notificationHandler = handler; + } + + registerRequestHandler(method: string): void { + harness.registerRequestCalls.push(method); + } + + async startThread(): Promise<{ thread: { id: string } }> { + return { thread: { id: 'thread-anonymous' } }; + } + + async resumeThread(): Promise<{ thread: { id: string } }> { + return { thread: { id: 'thread-anonymous' } }; + } + + async startTurn(): Promise<{ turn: Record }> { + const started = { turn: {} }; + harness.notifications.push({ method: 'turn/started', params: started }); + this.notificationHandler?.('turn/started', started); + + const completed = { status: 'Completed', turn: {} }; + harness.notifications.push({ method: 'turn/completed', params: completed }); + this.notificationHandler?.('turn/completed', completed); + + return { turn: {} }; + } + + async interruptTurn(): Promise> { + return {}; + } + + async disconnect(): Promise {} + } + + return { CodexAppServerClient: MockCodexAppServerClient }; +}); + +vi.mock('./utils/buildHapiMcpBridge', () => ({ + buildHapiMcpBridge: async () => ({ + server: { + stop: () => {} + }, + mcpServers: {} + }) +})); + +import { codexRemoteLauncher } from './codexRemoteLauncher'; + +type FakeAgentState = { + requests: Record; + completedRequests: Record; +}; + +function createMode(): EnhancedMode { + return { + permissionMode: 'default' + }; +} + +function createSessionStub() { + const queue = new MessageQueue2((mode) => JSON.stringify(mode)); + queue.push('hello from launcher test', createMode()); + queue.close(); + + const sessionEvents: Array<{ type: string; [key: string]: unknown }> = []; + const codexMessages: unknown[] = []; + const thinkingChanges: boolean[] = []; + const foundSessionIds: string[] = []; + let agentState: FakeAgentState = { + requests: {}, + completedRequests: {} + }; + + const rpcHandlers = new Map unknown>(); + const client = { + rpcHandlerManager: { + registerHandler(method: string, handler: (params: unknown) => unknown) { + rpcHandlers.set(method, handler); + } + }, + updateAgentState(handler: (state: FakeAgentState) => FakeAgentState) { + agentState = handler(agentState); + }, + sendCodexMessage(message: unknown) { + codexMessages.push(message); + }, + sendUserMessage(_text: string) {}, + sendSessionEvent(event: { type: string; [key: string]: unknown }) { + sessionEvents.push(event); + } + }; + + const session = { + path: '/tmp/hapi-update', + logPath: '/tmp/hapi-update/test.log', + client, + queue, + codexArgs: undefined, + codexCliOverrides: undefined, + sessionId: null as string | null, + thinking: false, + onThinkingChange(nextThinking: boolean) { + session.thinking = nextThinking; + thinkingChanges.push(nextThinking); + }, + onSessionFound(id: string) { + session.sessionId = id; + foundSessionIds.push(id); + }, + sendCodexMessage(message: unknown) { + client.sendCodexMessage(message); + }, + sendSessionEvent(event: { type: string; [key: string]: unknown }) { + client.sendSessionEvent(event); + }, + sendUserMessage(text: string) { + client.sendUserMessage(text); + } + }; + + return { + session, + sessionEvents, + codexMessages, + thinkingChanges, + foundSessionIds, + rpcHandlers, + getAgentState: () => agentState + }; +} + +describe('codexRemoteLauncher', () => { + afterEach(() => { + harness.notifications = []; + harness.registerRequestCalls = []; + delete process.env.CODEX_USE_MCP_SERVER; + }); + + it('finishes a turn and emits ready when task lifecycle events omit turn_id', async () => { + delete process.env.CODEX_USE_MCP_SERVER; + const { + session, + sessionEvents, + thinkingChanges, + foundSessionIds + } = createSessionStub(); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(foundSessionIds).toContain('thread-anonymous'); + expect(harness.notifications.map((entry) => entry.method)).toEqual(['turn/started', 'turn/completed']); + expect(sessionEvents.filter((event) => event.type === 'ready').length).toBeGreaterThanOrEqual(1); + expect(thinkingChanges).toContain(true); + expect(session.thinking).toBe(false); + }); +}); diff --git a/cli/src/codex/codexRemoteLauncher.ts b/cli/src/codex/codexRemoteLauncher.ts index c42054fa..0326320e 100644 --- a/cli/src/codex/codexRemoteLauncher.ts +++ b/cli/src/codex/codexRemoteLauncher.ts @@ -18,6 +18,7 @@ import { buildCodexStartConfig } from './utils/codexStartConfig'; import { AppServerEventConverter } from './utils/appServerEventConverter'; import { registerAppServerPermissionHandlers } from './utils/appServerPermissionAdapter'; import { buildThreadStartParams, buildTurnStartParams } from './utils/appServerConfig'; +import { shouldIgnoreTerminalEvent } from './utils/terminalEventGuard'; import { RemoteLauncherBase, type RemoteLauncherDisplayContext, @@ -155,6 +156,15 @@ class CodexRemoteLauncher extends RemoteLauncherBase { return typeof value === 'string' && value.length > 0 ? value : null; }; + const buildMcpToolName = (server: unknown, tool: unknown): string | null => { + const serverName = asString(server); + const toolName = asString(tool); + if (!serverName || !toolName) { + return null; + } + return `mcp__${serverName}__${toolName}`; + }; + const formatOutputPreview = (value: unknown): string => { if (typeof value === 'string') return value; if (typeof value === 'number' || typeof value === 'boolean') return String(value); @@ -214,10 +224,17 @@ class CodexRemoteLauncher extends RemoteLauncherBase { this.permissionHandler = permissionHandler; this.reasoningProcessor = reasoningProcessor; this.diffProcessor = diffProcessor; + let readyAfterTurnTimer: ReturnType | null = null; + let scheduleReadyAfterTurn: (() => void) | null = null; + let clearReadyAfterTurnTimer: (() => void) | null = null; + let turnInFlight = false; + let allowAnonymousTerminalEvent = false; const handleCodexEvent = (msg: Record) => { const msgType = asString(msg.type); if (!msgType) return; + const eventTurnId = asString(msg.turn_id ?? msg.turnId); + const isTerminalEvent = msgType === 'task_complete' || msgType === 'turn_aborted' || msgType === 'task_failed'; if (msgType === 'thread_started') { const threadId = asString(msg.thread_id ?? msg.threadId); @@ -229,14 +246,32 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } if (msgType === 'task_started') { - const turnId = asString(msg.turn_id ?? msg.turnId); + const turnId = eventTurnId; if (turnId) { this.currentTurnId = turnId; + allowAnonymousTerminalEvent = false; + } else if (useAppServer && !this.currentTurnId) { + allowAnonymousTerminalEvent = true; } } - if (msgType === 'task_complete' || msgType === 'turn_aborted' || msgType === 'task_failed') { + if (isTerminalEvent) { + if (shouldIgnoreTerminalEvent({ + useAppServer, + eventTurnId, + currentTurnId: this.currentTurnId, + turnInFlight, + allowAnonymousTerminalEvent + })) { + logger.debug( + `[Codex] Ignoring terminal event ${msgType} without matching turn context; ` + + `eventTurnId=${eventTurnId ?? 'none'}, activeTurn=${this.currentTurnId ?? 'none'}, ` + + `turnInFlight=${turnInFlight}, allowAnonymous=${allowAnonymousTerminalEvent}` + ); + return; + } this.currentTurnId = null; + allowAnonymousTerminalEvent = false; } if (!useAppServer) { @@ -274,28 +309,39 @@ class CodexRemoteLauncher extends RemoteLauncherBase { messageBuffer.addMessage('Starting task...', 'status'); } else if (msgType === 'task_complete') { messageBuffer.addMessage('Task completed', 'status'); - sendReady(); + if (!useAppServer) { + sendReady(); + } } else if (msgType === 'turn_aborted') { messageBuffer.addMessage('Turn aborted', 'status'); - sendReady(); + if (!useAppServer) { + sendReady(); + } } else if (msgType === 'task_failed') { const error = asString(msg.error); messageBuffer.addMessage(error ? `Task failed: ${error}` : 'Task failed', 'status'); - sendReady(); + if (!useAppServer) { + sendReady(); + } } if (msgType === 'task_started') { + clearReadyAfterTurnTimer?.(); if (useAppServer) { turnInFlight = true; + if (!eventTurnId && !this.currentTurnId) { + allowAnonymousTerminalEvent = true; + } } if (!session.thinking) { logger.debug('thinking started'); session.onThinkingChange(true); } } - if (msgType === 'task_complete' || msgType === 'turn_aborted' || msgType === 'task_failed') { + if (isTerminalEvent) { if (useAppServer) { turnInFlight = false; + allowAnonymousTerminalEvent = false; } if (session.thinking) { logger.debug('thinking completed'); @@ -304,6 +350,15 @@ class CodexRemoteLauncher extends RemoteLauncherBase { diffProcessor.reset(); appServerEventConverter?.reset(); } + + if (useAppServer) { + if (isTerminalEvent && !turnInFlight) { + scheduleReadyAfterTurn?.(); + } else if (readyAfterTurnTimer && msgType !== 'task_started') { + scheduleReadyAfterTurn?.(); + } + } + if (msgType === 'agent_reasoning_section_break') { reasoningProcessor.handleSectionBreak(); } @@ -415,6 +470,48 @@ class CodexRemoteLauncher extends RemoteLauncherBase { }); } } + if (msgType === 'mcp_tool_call_begin') { + const callId = asString(msg.call_id ?? msg.callId); + const invocation = asRecord(msg.invocation) ?? {}; + const name = buildMcpToolName( + invocation.server ?? invocation.server_name ?? msg.server, + invocation.tool ?? invocation.tool_name ?? msg.tool + ); + if (callId && name) { + session.sendCodexMessage({ + type: 'tool-call', + name, + callId, + input: invocation.arguments ?? invocation.input ?? msg.arguments ?? msg.input ?? {}, + id: randomUUID() + }); + } + } + if (msgType === 'mcp_tool_call_end') { + const callId = asString(msg.call_id ?? msg.callId); + const rawResult = msg.result; + let output = rawResult; + let isError = false; + const resultRecord = asRecord(rawResult); + if (resultRecord) { + if (Object.prototype.hasOwnProperty.call(resultRecord, 'Ok')) { + output = resultRecord.Ok; + } else if (Object.prototype.hasOwnProperty.call(resultRecord, 'Err')) { + output = resultRecord.Err; + isError = true; + } + } + + if (callId) { + session.sendCodexMessage({ + type: 'tool-call-result', + callId, + output, + is_error: isError, + id: randomUUID() + }); + } + } if (msgType === 'turn_diff') { const diff = asString(msg.unified_diff); if (diff) { @@ -492,7 +589,28 @@ class CodexRemoteLauncher extends RemoteLauncherBase { let currentModeHash: string | null = null; let pending: { message: string; mode: EnhancedMode; isolate: boolean; hash: string } | null = null; let first = true; - let turnInFlight = false; + + clearReadyAfterTurnTimer = () => { + if (!readyAfterTurnTimer) { + return; + } + clearTimeout(readyAfterTurnTimer); + readyAfterTurnTimer = null; + }; + + scheduleReadyAfterTurn = () => { + clearReadyAfterTurnTimer?.(); + readyAfterTurnTimer = setTimeout(() => { + readyAfterTurnTimer = null; + emitReadyIfIdle({ + pending, + queueSize: () => session.queue.size(), + shouldExit: this.shouldExit, + sendReady + }); + }, 120); + readyAfterTurnTimer.unref?.(); + }; while (!this.shouldExit) { logActiveHandles('loop-top'); @@ -589,6 +707,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { cliOverrides: session.codexCliOverrides }); turnInFlight = true; + allowAnonymousTerminalEvent = false; const turnResponse = await appServerClient.startTurn(turnParams, { signal: this.abortController.signal }); @@ -597,6 +716,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase { const turnId = asString(turn?.id); if (turnId) { this.currentTurnId = turnId; + } else if (!this.currentTurnId) { + allowAnonymousTerminalEvent = true; } } else if (mcpClient) { const startConfig: CodexSessionConfig = buildCodexStartConfig({ @@ -628,6 +749,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { cliOverrides: session.codexCliOverrides }); turnInFlight = true; + allowAnonymousTerminalEvent = false; const turnResponse = await appServerClient.startTurn(turnParams, { signal: this.abortController.signal }); @@ -636,6 +758,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase { const turnId = asString(turn?.id); if (turnId) { this.currentTurnId = turnId; + } else if (!this.currentTurnId) { + allowAnonymousTerminalEvent = true; } } else if (mcpClient) { await mcpClient.continueSession(message.message, { signal: this.abortController.signal }); @@ -646,6 +770,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase { const isAbortError = error instanceof Error && error.name === 'AbortError'; if (useAppServer) { turnInFlight = false; + allowAnonymousTerminalEvent = false; + this.currentTurnId = null; } if (isAbortError) { @@ -666,12 +792,14 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } } } finally { - permissionHandler.reset(); - reasoningProcessor.abort(); - diffProcessor.reset(); - appServerEventConverter?.reset(); - session.onThinkingChange(false); - if (!useAppServer || !turnInFlight) { + const shouldFinalizeTurnState = !useAppServer || !turnInFlight; + if (shouldFinalizeTurnState) { + permissionHandler.reset(); + reasoningProcessor.abort(); + diffProcessor.reset(); + appServerEventConverter?.reset(); + session.onThinkingChange(false); + clearReadyAfterTurnTimer?.(); emitReadyIfIdle({ pending, queueSize: () => session.queue.size(), diff --git a/cli/src/codex/utils/appServerConfig.test.ts b/cli/src/codex/utils/appServerConfig.test.ts index 1155b7bd..87ffc62c 100644 --- a/cli/src/codex/utils/appServerConfig.test.ts +++ b/cli/src/codex/utils/appServerConfig.test.ts @@ -15,11 +15,13 @@ describe('appServerConfig', () => { expect(params.sandbox).toBe('danger-full-access'); expect(params.approvalPolicy).toBe('never'); expect(params.baseInstructions).toBe(codexSystemPrompt); + expect(params.developerInstructions).toBe(codexSystemPrompt); expect(params.config).toEqual({ 'mcp_servers.hapi': { command: 'node', args: ['mcp'] - } + }, + developer_instructions: codexSystemPrompt }); }); @@ -34,6 +36,24 @@ describe('appServerConfig', () => { expect(params.approvalPolicy).toBe('on-failure'); }); + it('concatenates custom developer instructions after base instructions', () => { + const params = buildThreadStartParams({ + mode: { permissionMode: 'default' }, + mcpServers, + developerInstructions: 'Only respond in Chinese.' + }); + + expect(params.baseInstructions).toBe(codexSystemPrompt); + expect(params.developerInstructions).toBe(`${codexSystemPrompt}\n\nOnly respond in Chinese.`); + expect(params.config).toEqual({ + 'mcp_servers.hapi': { + command: 'node', + args: ['mcp'] + }, + developer_instructions: `${codexSystemPrompt}\n\nOnly respond in Chinese.` + }); + }); + it('builds turn params with mode defaults', () => { const params = buildTurnStartParams({ threadId: 'thread-1', diff --git a/cli/src/codex/utils/appServerConfig.ts b/cli/src/codex/utils/appServerConfig.ts index 5a2aa102..0804c9e5 100644 --- a/cli/src/codex/utils/appServerConfig.ts +++ b/cli/src/codex/utils/appServerConfig.ts @@ -88,13 +88,20 @@ export function buildThreadStartParams(args: { const config = buildMcpServerConfig(args.mcpServers); const baseInstructions = args.baseInstructions ?? codexSystemPrompt; + const resolvedDeveloperInstructions = args.developerInstructions + ? `${baseInstructions}\n\n${args.developerInstructions}` + : baseInstructions; + const configWithInstructions = { + ...config, + developer_instructions: resolvedDeveloperInstructions + }; const params: ThreadStartParams = { approvalPolicy: resolvedApprovalPolicy, sandbox: resolvedSandbox, baseInstructions, - ...(args.developerInstructions ? { developerInstructions: args.developerInstructions } : {}), - ...(Object.keys(config).length > 0 ? { config } : {}) + developerInstructions: resolvedDeveloperInstructions, + ...(Object.keys(configWithInstructions).length > 0 ? { config: configWithInstructions } : {}) }; if (args.mode.model) { diff --git a/cli/src/codex/utils/appServerEventConverter.test.ts b/cli/src/codex/utils/appServerEventConverter.test.ts index 73459c70..27276926 100644 --- a/cli/src/codex/utils/appServerEventConverter.test.ts +++ b/cli/src/codex/utils/appServerEventConverter.test.ts @@ -44,6 +44,21 @@ describe('AppServerEventConverter', () => { expect(completed).toEqual([{ type: 'agent_message', message: 'Hello world' }]); }); + it('deduplicates repeated agent message completions for the same item', () => { + const converter = new AppServerEventConverter(); + + converter.handleNotification('item/agentMessage/delta', { itemId: 'msg-1', delta: 'Hello' }); + const first = converter.handleNotification('item/completed', { + item: { id: 'msg-1', type: 'AgentMessage' } + }); + const second = converter.handleNotification('item/completed', { + item: { id: 'msg-1', type: 'agentMessage' } + }); + + expect(first).toEqual([{ type: 'agent_message', message: 'Hello' }]); + expect(second).toEqual([]); + }); + it('maps command execution items and output deltas', () => { const converter = new AppServerEventConverter(); @@ -77,10 +92,172 @@ describe('AppServerEventConverter', () => { expect(events).toEqual([{ type: 'agent_reasoning_delta', delta: 'step' }]); }); + it('dedupes duplicate reasoning deltas', () => { + const converter = new AppServerEventConverter(); + + expect(converter.handleNotification('item/reasoning/textDelta', { itemId: 'r1', delta: 'Hello ' })) + .toEqual([{ type: 'agent_reasoning_delta', delta: 'Hello ' }]); + expect(converter.handleNotification('item/reasoning/textDelta', { itemId: 'r1', delta: 'Hello ' })) + .toEqual([]); + converter.handleNotification('item/reasoning/textDelta', { itemId: 'r1', delta: 'world' }); + + const completed = converter.handleNotification('item/completed', { + item: { id: 'r1', type: 'reasoning' } + }); + + expect(completed).toEqual([{ type: 'agent_reasoning', text: 'Hello world' }]); + }); + + it('maps reasoning summary deltas', () => { + const converter = new AppServerEventConverter(); + + const events = converter.handleNotification('item/reasoning/summaryTextDelta', { itemId: 'r1', delta: 'step' }); + expect(events).toEqual([{ type: 'agent_reasoning_delta', delta: 'step' }]); + }); + + it('deduplicates repeated reasoning completions for the same item', () => { + const converter = new AppServerEventConverter(); + + const first = converter.handleNotification('item/completed', { + item: { id: 'r1', type: 'Reasoning', summary_text: ['Plan'] } + }); + const second = converter.handleNotification('item/completed', { + item: { id: 'r1', type: 'reasoning', summary_text: ['Plan'] } + }); + + expect(first).toEqual([{ type: 'agent_reasoning', text: 'Plan' }]); + expect(second).toEqual([]); + }); + it('maps diff updates', () => { const converter = new AppServerEventConverter(); const events = converter.handleNotification('turn/diff/updated', { diff: 'diff --git a b' }); expect(events).toEqual([{ type: 'turn_diff', unified_diff: 'diff --git a b' }]); }); + + it('unwraps codex/event task lifecycle', () => { + const converter = new AppServerEventConverter(); + + const started = converter.handleNotification('codex/event/task_started', { + msg: { type: 'task_started', turn_id: 'turn-1' } + }); + expect(started).toEqual([{ type: 'task_started', turn_id: 'turn-1' }]); + + const completed = converter.handleNotification('codex/event/task_complete', { + msg: { type: 'task_complete', turn_id: 'turn-1' } + }); + expect(completed).toEqual([{ type: 'task_complete', turn_id: 'turn-1' }]); + }); + + it('ignores wrapped terminal lifecycle events without turn_id', () => { + const converter = new AppServerEventConverter(); + + const completed = converter.handleNotification('codex/event/task_complete', { + msg: { type: 'task_complete' } + }); + + expect(completed).toEqual([]); + }); + + it('unwraps codex/event agent deltas and item completion', () => { + const converter = new AppServerEventConverter(); + + converter.handleNotification('codex/event/agent_message_delta', { + msg: { type: 'agent_message_delta', item_id: 'msg-1', delta: 'Hello' } + }); + converter.handleNotification('codex/event/agent_message_content_delta', { + msg: { type: 'agent_message_content_delta', item_id: 'msg-1', delta: ' world' } + }); + + const completed = converter.handleNotification('codex/event/item_completed', { + msg: { + type: 'item_completed', + item_id: 'msg-1', + item: { id: 'msg-1', type: 'AgentMessage' } + } + }); + + expect(completed).toEqual([{ type: 'agent_message', message: 'Hello world' }]); + }); + + it('unwraps codex/event reasoning completion from summary text', () => { + const converter = new AppServerEventConverter(); + + converter.handleNotification('codex/event/reasoning_content_delta', { + msg: { type: 'reasoning_content_delta', item_id: 'r1', delta: 'Plan' } + }); + const completed = converter.handleNotification('codex/event/item_completed', { + msg: { + type: 'item_completed', + item_id: 'r1', + item: { id: 'r1', type: 'Reasoning', summary_text: ['Plan done'] } + } + }); + + expect(completed).toEqual([{ type: 'agent_reasoning', text: 'Plan done' }]); + }); + + it('prefers canonical reasoning stream over wrapped agent_reasoning events', () => { + const converter = new AppServerEventConverter(); + + const section = converter.handleNotification('codex/event/agent_reasoning_section_break', { + msg: { type: 'agent_reasoning_section_break', item_id: 'r1' } + }); + const delta = converter.handleNotification('codex/event/agent_reasoning_delta', { + msg: { type: 'agent_reasoning_delta', item_id: 'r1', delta: 'step' } + }); + const reasoning = converter.handleNotification('codex/event/agent_reasoning', { + msg: { type: 'agent_reasoning', item_id: 'r1', text: 'Plan' } + }); + + expect(section).toEqual([{ type: 'agent_reasoning_section_break' }]); + expect(delta).toEqual([]); + expect(reasoning).toEqual([]); + }); + + it('deduplicates section break when wrapped and direct summary part events share the same index', () => { + const converter = new AppServerEventConverter(); + + const wrapped = converter.handleNotification('codex/event/agent_reasoning_section_break', { + msg: { type: 'agent_reasoning_section_break', item_id: 'r1', summary_index: 0 } + }); + const direct = converter.handleNotification('item/reasoning/summaryPartAdded', { + itemId: 'r1', + summaryIndex: 0 + }); + + expect(wrapped).toEqual([{ type: 'agent_reasoning_section_break' }]); + expect(direct).toEqual([]); + }); + + it('ignores wrapped final agent message and relies on item completion', () => { + const converter = new AppServerEventConverter(); + + const wrapped = converter.handleNotification('codex/event/agent_message', { + msg: { type: 'agent_message', item_id: 'msg-1', message: 'Hello' } + }); + + expect(wrapped).toEqual([]); + }); + + it('ignores wrapped retryable errors', () => { + const converter = new AppServerEventConverter(); + + const events = converter.handleNotification('codex/event/error', { + msg: { type: 'error', message: 'temporary', will_retry: true } + }); + + expect(events).toEqual([]); + }); + + it('maps wrapped non-retryable errors to task_failed', () => { + const converter = new AppServerEventConverter(); + + const events = converter.handleNotification('codex/event/error', { + msg: { type: 'error', message: 'fatal' } + }); + + expect(events).toEqual([{ type: 'task_failed', error: 'fatal' }]); + }); }); diff --git a/cli/src/codex/utils/appServerEventConverter.ts b/cli/src/codex/utils/appServerEventConverter.ts index 8760af48..08a95763 100644 --- a/cli/src/codex/utils/appServerEventConverter.ts +++ b/cli/src/codex/utils/appServerEventConverter.ts @@ -76,17 +76,187 @@ function extractChanges(value: unknown): Record | null { return null; } +function extractTextFromContent(value: unknown): string | null { + if (typeof value === 'string' && value.length > 0) { + return value; + } + + if (!Array.isArray(value)) { + return null; + } + + const chunks: string[] = []; + for (const entry of value) { + const record = asRecord(entry); + if (!record) continue; + const text = asString(record.text ?? record.message ?? record.content); + if (text) { + chunks.push(text); + } + } + + if (chunks.length === 0) { + return null; + } + + return chunks.join(''); +} + +function extractItemText(item: Record): string | null { + return asString(item.text ?? item.message) ?? extractTextFromContent(item.content); +} + +function extractReasoningText(item: Record): string | null { + const direct = extractItemText(item); + if (direct) { + return direct; + } + + const summary = item.summary_text ?? item.summaryText; + if (Array.isArray(summary)) { + const chunks = summary.filter((part): part is string => typeof part === 'string' && part.length > 0); + if (chunks.length > 0) { + return chunks.join('\n'); + } + } + + return null; +} + export class AppServerEventConverter { private readonly agentMessageBuffers = new Map(); private readonly reasoningBuffers = new Map(); private readonly commandOutputBuffers = new Map(); private readonly commandMeta = new Map>(); private readonly fileChangeMeta = new Map>(); + private readonly completedAgentMessageItems = new Set(); + private readonly completedReasoningItems = new Set(); + private readonly reasoningSectionBreakKeys = new Set(); + private readonly lastAgentMessageDeltaByItemId = new Map(); + private readonly lastReasoningDeltaByItemId = new Map(); + private readonly lastCommandOutputDeltaByItemId = new Map(); + + private handleWrappedCodexEvent(paramsRecord: Record): ConvertedEvent[] | null { + const msg = asRecord(paramsRecord.msg); + if (!msg) { + return []; + } + + const msgType = asString(msg.type); + if (!msgType) { + return []; + } + + if (msgType === 'item_started' || msgType === 'item_completed') { + const itemMethod = msgType === 'item_started' ? 'item/started' : 'item/completed'; + const item = asRecord(msg.item) ?? {}; + const params: Record = { + item, + itemId: asString(msg.item_id ?? msg.itemId ?? item.id), + threadId: asString(msg.thread_id ?? msg.threadId), + turnId: asString(msg.turn_id ?? msg.turnId) + }; + return this.handleNotification(itemMethod, params); + } + + if ( + msgType === 'task_started' || + msgType === 'task_complete' || + msgType === 'turn_aborted' || + msgType === 'task_failed' + ) { + const turnId = asString(msg.turn_id ?? msg.turnId); + if ((msgType === 'task_complete' || msgType === 'turn_aborted' || msgType === 'task_failed') && !turnId) { + logger.debug('[AppServerEventConverter] Ignoring wrapped terminal event without turn_id', { msgType }); + return []; + } + + const event: ConvertedEvent = { type: msgType }; + if (turnId) { + event.turn_id = turnId; + } + if (msgType === 'task_failed') { + const error = asString(msg.error ?? msg.message ?? asRecord(msg.error)?.message); + if (error) { + event.error = error; + } + } + return [event]; + } + + if (msgType === 'agent_message_delta' || msgType === 'agent_message_content_delta') { + const itemId = asString(msg.item_id ?? msg.itemId ?? msg.id) ?? 'agent-message'; + const delta = asString(msg.delta ?? msg.text ?? msg.message); + if (!delta) return []; + return this.handleNotification('item/agentMessage/delta', { itemId, delta }); + } + + if (msgType === 'reasoning_content_delta') { + const itemId = asString(msg.item_id ?? msg.itemId ?? msg.id) ?? 'reasoning'; + const delta = asString(msg.delta ?? msg.text ?? msg.message); + if (!delta) return []; + return this.handleNotification('item/reasoning/summaryTextDelta', { itemId, delta }); + } + + if (msgType === 'agent_reasoning_section_break') { + const itemId = asString(msg.item_id ?? msg.itemId ?? msg.id) ?? 'reasoning'; + const summaryIndex = asNumber(msg.summary_index ?? msg.summaryIndex); + return this.handleNotification('item/reasoning/summaryPartAdded', { + itemId, + ...(summaryIndex !== null ? { summaryIndex } : {}) + }); + } + + if (msgType === 'agent_reasoning_delta' || msgType === 'agent_reasoning' || msgType === 'agent_message') { + return []; + } + + if (msgType === 'exec_command_output_delta') { + const itemId = asString(msg.call_id ?? msg.callId ?? msg.item_id ?? msg.itemId ?? msg.id); + const delta = asString(msg.delta ?? msg.output ?? msg.stdout ?? msg.text); + if (!itemId || !delta) return []; + return this.handleNotification('item/commandExecution/outputDelta', { itemId, delta }); + } + + if (msgType === 'error') { + const errorRecord = asRecord(msg.error); + const willRetry = asBoolean(msg.will_retry ?? msg.willRetry ?? errorRecord?.will_retry ?? errorRecord?.willRetry) ?? false; + if (willRetry) { + return []; + } + const error = asString(msg.message ?? msg.reason ?? errorRecord?.message); + return error ? [{ type: 'task_failed', error }] : []; + } + + if ( + msgType === 'mcp_startup_update' || + msgType === 'mcp_startup_complete' || + msgType === 'plan_update' || + msgType === 'skills_update_available' || + msgType === 'stream_error' || + msgType === 'warning' || + msgType === 'context_compacted' || + msgType === 'terminal_interaction' || + msgType === 'user_message' + ) { + return []; + } + + return [msg as ConvertedEvent]; + } handleNotification(method: string, params: unknown): ConvertedEvent[] { const events: ConvertedEvent[] = []; const paramsRecord = asRecord(params) ?? {}; + if (method.startsWith('codex/event/')) { + return this.handleWrappedCodexEvent(paramsRecord) ?? events; + } + + if (method === 'account/rateLimits/updated' || method === 'turn/plan/updated' || method === 'thread/compacted') { + return events; + } + if (method === 'thread/started' || method === 'thread/resumed') { const thread = asRecord(paramsRecord.thread) ?? paramsRecord; const threadId = asString(thread.threadId ?? thread.thread_id ?? thread.id); @@ -152,16 +322,26 @@ export class AppServerEventConverter { const itemId = extractItemId(paramsRecord); const delta = asString(paramsRecord.delta ?? paramsRecord.text ?? paramsRecord.message); if (itemId && delta) { + const lastDelta = this.lastAgentMessageDeltaByItemId.get(itemId); + if (lastDelta === delta) { + return events; + } + this.lastAgentMessageDeltaByItemId.set(itemId, delta); const prev = this.agentMessageBuffers.get(itemId) ?? ''; this.agentMessageBuffers.set(itemId, prev + delta); } return events; } - if (method === 'item/reasoning/textDelta') { + if (method === 'item/reasoning/textDelta' || method === 'item/reasoning/summaryTextDelta') { const itemId = extractItemId(paramsRecord) ?? 'reasoning'; const delta = asString(paramsRecord.delta ?? paramsRecord.text ?? paramsRecord.message); if (delta) { + const lastDelta = this.lastReasoningDeltaByItemId.get(itemId); + if (lastDelta === delta) { + return events; + } + this.lastReasoningDeltaByItemId.set(itemId, delta); const prev = this.reasoningBuffers.get(itemId) ?? ''; this.reasoningBuffers.set(itemId, prev + delta); events.push({ type: 'agent_reasoning_delta', delta }); @@ -170,6 +350,15 @@ export class AppServerEventConverter { } if (method === 'item/reasoning/summaryPartAdded') { + const itemId = extractItemId(paramsRecord) ?? 'reasoning'; + const summaryIndex = asNumber(paramsRecord.summaryIndex ?? paramsRecord.summary_index); + if (summaryIndex !== null) { + const key = `${itemId}:${summaryIndex}`; + if (this.reasoningSectionBreakKeys.has(key)) { + return events; + } + this.reasoningSectionBreakKeys.add(key); + } events.push({ type: 'agent_reasoning_section_break' }); return events; } @@ -178,6 +367,11 @@ export class AppServerEventConverter { const itemId = extractItemId(paramsRecord); const delta = asString(paramsRecord.delta ?? paramsRecord.text ?? paramsRecord.output ?? paramsRecord.stdout); if (itemId && delta) { + const lastDelta = this.lastCommandOutputDeltaByItemId.get(itemId); + if (lastDelta === delta) { + return events; + } + this.lastCommandOutputDeltaByItemId.set(itemId, delta); const prev = this.commandOutputBuffers.get(itemId) ?? ''; this.commandOutputBuffers.set(itemId, prev + delta); } @@ -197,22 +391,32 @@ export class AppServerEventConverter { if (itemType === 'agentmessage') { if (method === 'item/completed') { - const text = asString(item.text ?? item.message ?? item.content) ?? this.agentMessageBuffers.get(itemId); + if (this.completedAgentMessageItems.has(itemId)) { + return events; + } + const text = extractItemText(item) ?? this.agentMessageBuffers.get(itemId); if (text) { events.push({ type: 'agent_message', message: text }); + this.completedAgentMessageItems.add(itemId); + this.agentMessageBuffers.delete(itemId); } - this.agentMessageBuffers.delete(itemId); + this.lastAgentMessageDeltaByItemId.delete(itemId); } return events; } if (itemType === 'reasoning') { if (method === 'item/completed') { - const text = asString(item.text ?? item.message ?? item.content) ?? this.reasoningBuffers.get(itemId); + if (this.completedReasoningItems.has(itemId)) { + return events; + } + const text = extractReasoningText(item) ?? this.reasoningBuffers.get(itemId); if (text) { events.push({ type: 'agent_reasoning', text }); + this.completedReasoningItems.add(itemId); + this.reasoningBuffers.delete(itemId); } - this.reasoningBuffers.delete(itemId); + this.lastReasoningDeltaByItemId.delete(itemId); } return events; } @@ -256,6 +460,7 @@ export class AppServerEventConverter { this.commandMeta.delete(itemId); this.commandOutputBuffers.delete(itemId); + this.lastCommandOutputDeltaByItemId.delete(itemId); } return events; @@ -309,5 +514,11 @@ export class AppServerEventConverter { this.commandOutputBuffers.clear(); this.commandMeta.clear(); this.fileChangeMeta.clear(); + this.completedAgentMessageItems.clear(); + this.completedReasoningItems.clear(); + this.reasoningSectionBreakKeys.clear(); + this.lastAgentMessageDeltaByItemId.clear(); + this.lastReasoningDeltaByItemId.clear(); + this.lastCommandOutputDeltaByItemId.clear(); } } diff --git a/cli/src/codex/utils/systemPrompt.ts b/cli/src/codex/utils/systemPrompt.ts index cfa55ccf..c8be6620 100644 --- a/cli/src/codex/utils/systemPrompt.ts +++ b/cli/src/codex/utils/systemPrompt.ts @@ -13,7 +13,10 @@ import { trimIdent } from '@/utils/trimIdent'; * so the tool is called as `functions.hapi__change_title`. */ export const TITLE_INSTRUCTION = trimIdent(` - Based on this message, call functions.hapi__change_title to change chat session title that would represent the current task. If chat idea would change dramatically - call this function again to update the title. + ALWAYS when you start a new chat, call the title tool to set a concise task title. + Prefer calling functions.hapi__change_title. + If that exact tool name is unavailable, call an equivalent alias such as hapi__change_title, mcp__hapi__change_title, or hapi_change_title. + If the task focus changes significantly later, call the title tool again with a better title. `); /** diff --git a/cli/src/codex/utils/terminalEventGuard.test.ts b/cli/src/codex/utils/terminalEventGuard.test.ts new file mode 100644 index 00000000..c9c9947f --- /dev/null +++ b/cli/src/codex/utils/terminalEventGuard.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, it } from 'vitest'; +import { shouldIgnoreTerminalEvent } from './terminalEventGuard'; + +describe('shouldIgnoreTerminalEvent', () => { + it('returns false for non app-server mode', () => { + const ignored = shouldIgnoreTerminalEvent({ + useAppServer: false, + eventTurnId: null, + currentTurnId: 'turn-1', + turnInFlight: true + }); + + expect(ignored).toBe(false); + }); + + it('ignores terminal events without turn_id when current turn id exists', () => { + const ignored = shouldIgnoreTerminalEvent({ + useAppServer: true, + eventTurnId: null, + currentTurnId: 'turn-1', + turnInFlight: true + }); + + expect(ignored).toBe(true); + }); + + it('ignores terminal events without turn_id while a turn is still in flight', () => { + const ignored = shouldIgnoreTerminalEvent({ + useAppServer: true, + eventTurnId: null, + currentTurnId: null, + turnInFlight: true + }); + + expect(ignored).toBe(true); + }); + + it('accepts terminal events without turn_id when anonymous terminal is explicitly allowed', () => { + const ignored = shouldIgnoreTerminalEvent({ + useAppServer: true, + eventTurnId: null, + currentTurnId: null, + turnInFlight: true, + allowAnonymousTerminalEvent: true + }); + + expect(ignored).toBe(false); + }); + + it('still ignores terminal events without turn_id when current turn id exists', () => { + const ignored = shouldIgnoreTerminalEvent({ + useAppServer: true, + eventTurnId: null, + currentTurnId: 'turn-1', + turnInFlight: true, + allowAnonymousTerminalEvent: true + }); + + expect(ignored).toBe(true); + }); + + it('ignores stale terminal events from another turn', () => { + const ignored = shouldIgnoreTerminalEvent({ + useAppServer: true, + eventTurnId: 'turn-old', + currentTurnId: 'turn-current', + turnInFlight: true + }); + + expect(ignored).toBe(true); + }); + + it('accepts terminal events that match the current turn id', () => { + const ignored = shouldIgnoreTerminalEvent({ + useAppServer: true, + eventTurnId: 'turn-current', + currentTurnId: 'turn-current', + turnInFlight: true + }); + + expect(ignored).toBe(false); + }); + + it('accepts terminal events without turn_id when no turn is active', () => { + const ignored = shouldIgnoreTerminalEvent({ + useAppServer: true, + eventTurnId: null, + currentTurnId: null, + turnInFlight: false + }); + + expect(ignored).toBe(false); + }); +}); diff --git a/cli/src/codex/utils/terminalEventGuard.ts b/cli/src/codex/utils/terminalEventGuard.ts new file mode 100644 index 00000000..0319a7f0 --- /dev/null +++ b/cli/src/codex/utils/terminalEventGuard.ts @@ -0,0 +1,29 @@ +export type TerminalEventGuardInput = { + useAppServer: boolean; + eventTurnId: string | null; + currentTurnId: string | null; + turnInFlight: boolean; + allowAnonymousTerminalEvent?: boolean; +}; + +export function shouldIgnoreTerminalEvent(input: TerminalEventGuardInput): boolean { + const allowAnonymousTerminalEvent = input.allowAnonymousTerminalEvent === true; + + if (!input.useAppServer) { + return false; + } + + if (input.eventTurnId) { + return Boolean(input.currentTurnId && input.eventTurnId !== input.currentTurnId); + } + + if (input.currentTurnId) { + return true; + } + + if (input.turnInFlight && !allowAnonymousTerminalEvent) { + return true; + } + + return false; +}