mirror of
https://github.com/wu736139669/hapi.git
synced 2026-08-07 06:52:28 +00:00
fix(cli): harden ACP/Codex event handling (#211)
Co-authored-by: Lihengwannafly <Lihengwannafly@users.noreply.github.com>
This commit is contained in:
Lihengwannafly
co-authored by
Lihengwannafly
Lihengwannafly
parent
77cfa715f9
commit
10f7e1404e
@@ -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<typeof setTimeout> | null = null;
|
||||
let scheduleReadyAfterTurn: (() => void) | null = null;
|
||||
let clearReadyAfterTurnTimer: (() => void) | null = null;
|
||||
let turnInFlight = false;
|
||||
let allowAnonymousTerminalEvent = false;
|
||||
|
||||
const handleCodexEvent = (msg: Record<string, unknown>) => {
|
||||
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(),
|
||||
|
||||
Reference in New Issue
Block a user