mirror of
https://github.com/wu736139669/hapi.git
synced 2026-08-07 06:52:28 +00:00
fix(codex): surface thread system errors (#519)
Co-authored-by: Liu-KM <Liu-KM@users.noreply.github.com>
This commit is contained in:
@@ -5,7 +5,11 @@ import type { EnhancedMode } from './loop';
|
||||
const harness = vi.hoisted(() => ({
|
||||
notifications: [] as Array<{ method: string; params: unknown }>,
|
||||
registerRequestCalls: [] as string[],
|
||||
initializeCalls: [] as unknown[]
|
||||
initializeCalls: [] as unknown[],
|
||||
startThreadIds: [] as string[],
|
||||
resumeThreadIds: [] as string[],
|
||||
startTurnThreadIds: [] as string[],
|
||||
remainingThreadSystemErrors: 0
|
||||
}));
|
||||
|
||||
vi.mock('./codexAppServerClient', () => {
|
||||
@@ -28,23 +32,41 @@ vi.mock('./codexAppServerClient', () => {
|
||||
}
|
||||
|
||||
async startThread(): Promise<{ thread: { id: string }; model: string }> {
|
||||
return { thread: { id: 'thread-anonymous' }, model: 'gpt-5.4' };
|
||||
const id = `thread-${harness.startThreadIds.length + 1}`;
|
||||
harness.startThreadIds.push(id);
|
||||
return { thread: { id }, model: 'gpt-5.4' };
|
||||
}
|
||||
|
||||
async resumeThread(): Promise<{ thread: { id: string }; model: string }> {
|
||||
return { thread: { id: 'thread-anonymous' }, model: 'gpt-5.4' };
|
||||
async resumeThread(params?: { threadId?: string }): Promise<{ thread: { id: string }; model: string }> {
|
||||
const id = params?.threadId ?? 'thread-resumed';
|
||||
harness.resumeThreadIds.push(id);
|
||||
return { thread: { id }, model: 'gpt-5.4' };
|
||||
}
|
||||
|
||||
async startTurn(): Promise<{ turn: Record<string, never> }> {
|
||||
const started = { turn: {} };
|
||||
async startTurn(params?: { threadId?: string }): Promise<{ turn: { id?: string } }> {
|
||||
const threadId = params?.threadId ?? 'thread-unknown';
|
||||
harness.startTurnThreadIds.push(threadId);
|
||||
const turnId = `turn-${harness.startTurnThreadIds.length}`;
|
||||
const started = { turn: { id: turnId } };
|
||||
harness.notifications.push({ method: 'turn/started', params: started });
|
||||
this.notificationHandler?.('turn/started', started);
|
||||
|
||||
const completed = { status: 'Completed', turn: {} };
|
||||
if (harness.remainingThreadSystemErrors > 0) {
|
||||
harness.remainingThreadSystemErrors -= 1;
|
||||
const failed = {
|
||||
thread: { id: threadId },
|
||||
status: { type: 'systemError' }
|
||||
};
|
||||
harness.notifications.push({ method: 'thread/status/changed', params: failed });
|
||||
this.notificationHandler?.('thread/status/changed', failed);
|
||||
return { turn: { id: turnId } };
|
||||
}
|
||||
|
||||
const completed = { status: 'Completed', turn: { id: turnId } };
|
||||
harness.notifications.push({ method: 'turn/completed', params: completed });
|
||||
this.notificationHandler?.('turn/completed', completed);
|
||||
|
||||
return { turn: {} };
|
||||
return { turn: { id: turnId } };
|
||||
}
|
||||
|
||||
async interruptTurn(): Promise<Record<string, never>> {
|
||||
@@ -80,9 +102,15 @@ function createMode(): EnhancedMode {
|
||||
};
|
||||
}
|
||||
|
||||
function createSessionStub() {
|
||||
function createSessionStub(messages = ['hello from launcher test']) {
|
||||
const queue = new MessageQueue2<EnhancedMode>((mode) => JSON.stringify(mode));
|
||||
queue.push('hello from launcher test', createMode());
|
||||
messages.forEach((message, index) => {
|
||||
if (index === 0 && messages.length > 1) {
|
||||
queue.pushIsolateAndClear(message, createMode());
|
||||
} else {
|
||||
queue.push(message, createMode());
|
||||
}
|
||||
});
|
||||
queue.close();
|
||||
|
||||
const sessionEvents: Array<{ type: string; [key: string]: unknown }> = [];
|
||||
@@ -168,9 +196,13 @@ describe('codexRemoteLauncher', () => {
|
||||
harness.notifications = [];
|
||||
harness.registerRequestCalls = [];
|
||||
harness.initializeCalls = [];
|
||||
harness.startThreadIds = [];
|
||||
harness.resumeThreadIds = [];
|
||||
harness.startTurnThreadIds = [];
|
||||
harness.remainingThreadSystemErrors = 0;
|
||||
});
|
||||
|
||||
it('finishes a turn and emits ready when task lifecycle events omit turn_id', async () => {
|
||||
it('finishes a turn and emits ready when task lifecycle events include turn_id', async () => {
|
||||
const {
|
||||
session,
|
||||
sessionEvents,
|
||||
@@ -182,7 +214,7 @@ describe('codexRemoteLauncher', () => {
|
||||
const exitReason = await codexRemoteLauncher(session as never);
|
||||
|
||||
expect(exitReason).toBe('exit');
|
||||
expect(foundSessionIds).toContain('thread-anonymous');
|
||||
expect(foundSessionIds).toContain('thread-1');
|
||||
expect(getModel()).toBe('gpt-5.4');
|
||||
expect(harness.initializeCalls).toEqual([{
|
||||
clientInfo: {
|
||||
@@ -198,4 +230,34 @@ describe('codexRemoteLauncher', () => {
|
||||
expect(thinkingChanges).toContain(true);
|
||||
expect(session.thinking).toBe(false);
|
||||
});
|
||||
|
||||
it('surfaces thread-level systemError as a visible failure and emits ready', async () => {
|
||||
harness.remainingThreadSystemErrors = 1;
|
||||
const { session, sessionEvents } = createSessionStub();
|
||||
|
||||
const exitReason = await codexRemoteLauncher(session as never);
|
||||
|
||||
expect(exitReason).toBe('exit');
|
||||
expect(harness.notifications.map((entry) => entry.method)).toEqual(['turn/started', 'thread/status/changed']);
|
||||
expect(sessionEvents).toContainEqual({
|
||||
type: 'message',
|
||||
message: 'Task failed: Codex thread entered systemError'
|
||||
});
|
||||
expect(sessionEvents.filter((event) => event.type === 'ready').length).toBeGreaterThanOrEqual(1);
|
||||
expect(session.thinking).toBe(false);
|
||||
});
|
||||
|
||||
it('starts a fresh thread for the next queued message after thread-level systemError', async () => {
|
||||
harness.remainingThreadSystemErrors = 1;
|
||||
const { session } = createSessionStub(['first message', 'second message']);
|
||||
|
||||
const exitReason = await codexRemoteLauncher(session as never);
|
||||
|
||||
expect(exitReason).toBe('exit');
|
||||
expect(harness.startThreadIds).toEqual(['thread-1', 'thread-2']);
|
||||
expect(harness.resumeThreadIds).toEqual([]);
|
||||
expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-2']);
|
||||
expect(session.sessionId).toBe('thread-2');
|
||||
expect(session.thinking).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -241,11 +241,13 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
let clearReadyAfterTurnTimer: (() => void) | null = null;
|
||||
let turnInFlight = false;
|
||||
let allowAnonymousTerminalEvent = false;
|
||||
let invalidThreadId: string | null = null;
|
||||
|
||||
const handleCodexEvent = (msg: Record<string, unknown>) => {
|
||||
const msgType = asString(msg.type);
|
||||
if (!msgType) return;
|
||||
const eventTurnId = asString(msg.turn_id ?? msg.turnId);
|
||||
const eventThreadId = asString(msg.thread_id ?? msg.threadId);
|
||||
const isTerminalEvent = msgType === 'task_complete' || msgType === 'turn_aborted' || msgType === 'task_failed';
|
||||
|
||||
if (msgType === 'thread_started') {
|
||||
@@ -267,22 +269,33 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
}
|
||||
}
|
||||
|
||||
const isThreadStatusFailure = msgType === 'task_failed' && msg.terminal_source === 'thread_status';
|
||||
|
||||
if (isTerminalEvent) {
|
||||
if (shouldIgnoreTerminalEvent({
|
||||
eventTurnId,
|
||||
currentTurnId: this.currentTurnId,
|
||||
turnInFlight,
|
||||
allowAnonymousTerminalEvent
|
||||
allowAnonymousTerminalEvent,
|
||||
eventThreadId,
|
||||
currentThreadId: this.currentThreadId,
|
||||
allowMatchingThreadIdTerminalEvent: msg.terminal_source === 'thread_status'
|
||||
})) {
|
||||
logger.debug(
|
||||
`[Codex] Ignoring terminal event ${msgType} without matching turn context; ` +
|
||||
`eventTurnId=${eventTurnId ?? 'none'}, activeTurn=${this.currentTurnId ?? 'none'}, ` +
|
||||
`eventThreadId=${eventThreadId ?? 'none'}, activeThread=${this.currentThreadId ?? 'none'}, ` +
|
||||
`turnInFlight=${turnInFlight}, allowAnonymous=${allowAnonymousTerminalEvent}`
|
||||
);
|
||||
return;
|
||||
}
|
||||
this.currentTurnId = null;
|
||||
allowAnonymousTerminalEvent = false;
|
||||
if (isThreadStatusFailure) {
|
||||
invalidThreadId = eventThreadId ?? this.currentThreadId;
|
||||
this.currentThreadId = null;
|
||||
hasThread = false;
|
||||
}
|
||||
}
|
||||
|
||||
if (msgType === 'agent_message') {
|
||||
@@ -314,7 +327,9 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
messageBuffer.addMessage('Turn aborted', 'status');
|
||||
} else if (msgType === 'task_failed') {
|
||||
const error = asString(msg.error);
|
||||
messageBuffer.addMessage(error ? `Task failed: ${error}` : 'Task failed', 'status');
|
||||
const message = error ? `Task failed: ${error}` : 'Task failed';
|
||||
messageBuffer.addMessage(message, 'status');
|
||||
session.sendSessionEvent({ type: 'message', message });
|
||||
}
|
||||
|
||||
if (msgType === 'task_started') {
|
||||
@@ -627,7 +642,9 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
cliOverrides: session.codexCliOverrides
|
||||
});
|
||||
|
||||
const resumeCandidate = session.sessionId;
|
||||
const resumeCandidate = session.sessionId && session.sessionId !== invalidThreadId
|
||||
? session.sessionId
|
||||
: null;
|
||||
let threadId: string | null = null;
|
||||
|
||||
if (resumeCandidate) {
|
||||
@@ -695,11 +712,13 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
const turnRecord = asRecord(turnResponse);
|
||||
const turn = turnRecord ? asRecord(turnRecord.turn) : null;
|
||||
const turnId = asString(turn?.id);
|
||||
if (turnInFlight) {
|
||||
if (turnId) {
|
||||
this.currentTurnId = turnId;
|
||||
} else if (!this.currentTurnId) {
|
||||
allowAnonymousTerminalEvent = true;
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
logger.warn('Error in codex session:', error);
|
||||
const isAbortError = error instanceof Error && error.name === 'AbortError';
|
||||
|
||||
@@ -16,6 +16,21 @@ describe('AppServerEventConverter', () => {
|
||||
expect(events).toEqual([{ type: 'thread_started', thread_id: 'thread-2' }]);
|
||||
});
|
||||
|
||||
it('maps thread systemError to a task failure', () => {
|
||||
const converter = new AppServerEventConverter();
|
||||
const events = converter.handleNotification('thread/status/changed', {
|
||||
thread: { id: 'thread-1' },
|
||||
status: { type: 'systemError' }
|
||||
});
|
||||
|
||||
expect(events).toEqual([{
|
||||
type: 'task_failed',
|
||||
thread_id: 'thread-1',
|
||||
terminal_source: 'thread_status',
|
||||
error: 'Codex thread entered systemError'
|
||||
}]);
|
||||
});
|
||||
|
||||
it('maps turn/started and completed statuses', () => {
|
||||
const converter = new AppServerEventConverter();
|
||||
|
||||
|
||||
@@ -266,6 +266,24 @@ export class AppServerEventConverter {
|
||||
return events;
|
||||
}
|
||||
|
||||
if (method === 'thread/status/changed') {
|
||||
const thread = asRecord(paramsRecord.thread) ?? paramsRecord;
|
||||
const threadId = asString(thread.threadId ?? thread.thread_id ?? thread.id);
|
||||
const status = asRecord(paramsRecord.status ?? thread.status);
|
||||
const statusType = asString(status?.type ?? paramsRecord.statusType ?? paramsRecord.status_type);
|
||||
if (statusType === 'systemError') {
|
||||
const error = asString(status?.message ?? status?.error ?? paramsRecord.message ?? paramsRecord.error)
|
||||
?? 'Codex thread entered systemError';
|
||||
events.push({
|
||||
type: 'task_failed',
|
||||
...(threadId ? { thread_id: threadId } : {}),
|
||||
terminal_source: 'thread_status',
|
||||
error
|
||||
});
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
if (method === 'turn/started') {
|
||||
const turn = asRecord(paramsRecord.turn) ?? paramsRecord;
|
||||
const turnId = asString(turn.turnId ?? turn.turn_id ?? turn.id);
|
||||
|
||||
@@ -44,6 +44,32 @@ describe('shouldIgnoreTerminalEvent', () => {
|
||||
expect(ignored).toBe(true);
|
||||
});
|
||||
|
||||
it('accepts thread-level terminal events for the current thread when explicitly allowed', () => {
|
||||
const ignored = shouldIgnoreTerminalEvent({
|
||||
eventTurnId: null,
|
||||
currentTurnId: 'turn-1',
|
||||
turnInFlight: true,
|
||||
eventThreadId: 'thread-1',
|
||||
currentThreadId: 'thread-1',
|
||||
allowMatchingThreadIdTerminalEvent: true
|
||||
});
|
||||
|
||||
expect(ignored).toBe(false);
|
||||
});
|
||||
|
||||
it('ignores thread-level terminal events for a different thread', () => {
|
||||
const ignored = shouldIgnoreTerminalEvent({
|
||||
eventTurnId: null,
|
||||
currentTurnId: 'turn-1',
|
||||
turnInFlight: true,
|
||||
eventThreadId: 'thread-old',
|
||||
currentThreadId: 'thread-1',
|
||||
allowMatchingThreadIdTerminalEvent: true
|
||||
});
|
||||
|
||||
expect(ignored).toBe(true);
|
||||
});
|
||||
|
||||
it('ignores stale terminal events from another turn', () => {
|
||||
const ignored = shouldIgnoreTerminalEvent({
|
||||
eventTurnId: 'turn-old',
|
||||
|
||||
@@ -3,6 +3,9 @@ export type TerminalEventGuardInput = {
|
||||
currentTurnId: string | null;
|
||||
turnInFlight: boolean;
|
||||
allowAnonymousTerminalEvent?: boolean;
|
||||
eventThreadId?: string | null;
|
||||
currentThreadId?: string | null;
|
||||
allowMatchingThreadIdTerminalEvent?: boolean;
|
||||
};
|
||||
|
||||
export function shouldIgnoreTerminalEvent(input: TerminalEventGuardInput): boolean {
|
||||
@@ -13,6 +16,15 @@ export function shouldIgnoreTerminalEvent(input: TerminalEventGuardInput): boole
|
||||
}
|
||||
|
||||
if (input.currentTurnId) {
|
||||
const allowMatchingThreadIdTerminalEvent = input.allowMatchingThreadIdTerminalEvent === true;
|
||||
if (
|
||||
allowMatchingThreadIdTerminalEvent &&
|
||||
input.eventThreadId &&
|
||||
input.currentThreadId &&
|
||||
input.eventThreadId === input.currentThreadId
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user