From 6df84df75666bfebf4df04e893df37c29ae02d94 Mon Sep 17 00:00:00 2001 From: xiaobaifly7 Date: Wed, 6 May 2026 20:02:46 +0800 Subject: [PATCH] fix(hapi): consolidate approved web and Codex recovery fixes (#578) --- cli/src/codex/codexRemoteLauncher.test.ts | 199 +++++++++++++-- cli/src/codex/codexRemoteLauncher.ts | 240 ++++++++++++++++-- .../utils/appServerEventConverter.test.ts | 37 +++ .../codex/utils/appServerEventConverter.ts | 30 ++- cli/src/utils/spawnHappyCLI.test.ts | 89 ++++++- cli/src/utils/spawnHappyCLI.ts | 37 ++- hub/src/sync/messageService.ts | 11 + hub/src/sync/rpcGateway.test.ts | 64 +++++ hub/src/sync/rpcGateway.ts | 29 ++- hub/src/sync/sessionModel.test.ts | 193 ++++++++++++++ hub/src/sync/syncEngine.ts | 76 +++++- hub/src/web/routes/sessions.test.ts | 69 ++++- hub/src/web/routes/sessions.ts | 52 ++++ web/src/lib/codexSlashCommands.test.ts | 18 ++ 14 files changed, 1091 insertions(+), 53 deletions(-) create mode 100644 hub/src/sync/rpcGateway.test.ts diff --git a/cli/src/codex/codexRemoteLauncher.test.ts b/cli/src/codex/codexRemoteLauncher.test.ts index 7e1f6194..290a9990 100644 --- a/cli/src/codex/codexRemoteLauncher.test.ts +++ b/cli/src/codex/codexRemoteLauncher.test.ts @@ -12,7 +12,12 @@ const harness = vi.hoisted(() => ({ interruptedTurns: [] as Array<{ threadId: string; turnId: string }>, compactThreadIds: [] as string[], suppressTurnCompletion: false, - remainingThreadSystemErrors: 0 + remainingThreadSystemErrors: 0, + startTurnMessages: [] as string[], + failResumeThreadIds: [] as string[], + nextThreadSystemErrorMessage: null as string | null, + failNextCompact: false, + deferThreadStatusNotifications: false })); vi.mock('./codexAppServerClient', () => { @@ -43,12 +48,29 @@ vi.mock('./codexAppServerClient', () => { async resumeThread(params?: { threadId?: string }): Promise<{ thread: { id: string }; model: string }> { const id = params?.threadId ?? 'thread-resumed'; harness.resumeThreadIds.push(id); + if (harness.failResumeThreadIds.includes(id)) { + throw new Error('resume failed'); + } return { thread: { id }, model: 'gpt-5.4' }; } - async startTurn(params?: { threadId?: string }): Promise<{ turn: { id?: string } }> { + async compactThread(params?: { threadId?: string }): Promise> { + const threadId = params?.threadId ?? 'thread-unknown'; + harness.compactThreadIds.push(threadId); + if (harness.failNextCompact) { + harness.failNextCompact = false; + throw new Error('compact failed'); + } + const compacted = { threadId, turnId: `compact-${harness.compactThreadIds.length}` }; + harness.notifications.push({ method: 'thread/compacted', params: compacted }); + this.notificationHandler?.('thread/compacted', compacted); + return {}; + } + + async startTurn(params?: { threadId?: string; input?: Array<{ text?: string }>; message?: string; userMessage?: string }): Promise<{ turn: { id?: string } }> { const threadId = params?.threadId ?? 'thread-unknown'; harness.startTurnThreadIds.push(threadId); + harness.startTurnMessages.push(params?.input?.[0]?.text ?? params?.message ?? params?.userMessage ?? ''); const turnId = `turn-${harness.startTurnThreadIds.length}`; const started = { turn: { id: turnId } }; harness.notifications.push({ method: 'turn/started', params: started }); @@ -58,10 +80,15 @@ vi.mock('./codexAppServerClient', () => { harness.remainingThreadSystemErrors -= 1; const failed = { thread: { id: threadId }, - status: { type: 'systemError' } + status: { type: 'systemError', ...(harness.nextThreadSystemErrorMessage ? { message: harness.nextThreadSystemErrorMessage } : {}) } }; harness.notifications.push({ method: 'thread/status/changed', params: failed }); - this.notificationHandler?.('thread/status/changed', failed); + const notify = () => this.notificationHandler?.('thread/status/changed', failed); + if (harness.deferThreadStatusNotifications) { + setTimeout(notify, 0); + } else { + notify(); + } return { turn: { id: turnId } }; } @@ -110,11 +137,6 @@ vi.mock('./codexAppServerClient', () => { return {}; } - async compactThread(params?: { threadId?: string }): Promise> { - harness.compactThreadIds.push(params?.threadId ?? 'thread-unknown'); - return {}; - } - async disconnect(): Promise {} } @@ -250,7 +272,12 @@ describe('codexRemoteLauncher', () => { harness.interruptedTurns = []; harness.compactThreadIds = []; harness.suppressTurnCompletion = false; + harness.startTurnMessages = []; + harness.failResumeThreadIds = []; harness.remainingThreadSystemErrors = 0; + harness.nextThreadSystemErrorMessage = null; + harness.failNextCompact = false; + harness.deferThreadStatusNotifications = false; }); it('finishes a turn and emits ready when task lifecycle events include turn_id', async () => { @@ -287,14 +314,17 @@ describe('codexRemoteLauncher', () => { 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(); + it('surfaces thread-level systemError only after same-thread retries are exhausted', async () => { + harness.remainingThreadSystemErrors = 4; + const { session, sessionEvents } = createSessionStub(['first message']); 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(harness.startThreadIds).toEqual(['thread-1']); + expect(harness.resumeThreadIds).toEqual([]); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-1', 'thread-1', 'thread-1']); + expect(harness.startTurnMessages).toEqual(['first message', 'first message', 'first message', 'first message']); expect(sessionEvents).toContainEqual({ type: 'message', message: 'Task failed: Codex thread entered systemError' @@ -303,17 +333,152 @@ describe('codexRemoteLauncher', () => { expect(session.thinking).toBe(false); }); - it('starts a fresh thread for the next queued message after thread-level systemError', async () => { + it('retries a thread-level systemError on the same thread without starting a fresh thread', async () => { + harness.remainingThreadSystemErrors = 1; + const { session, sessionEvents } = createSessionStub(['first message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startThreadIds).toEqual(['thread-1']); + expect(harness.resumeThreadIds).toEqual([]); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-1']); + expect(harness.startTurnMessages).toEqual(['first message', 'first message']); + expect(session.sessionId).toBe('thread-1'); + expect(sessionEvents).not.toContainEqual({ + type: 'message', + message: 'Task failed: Codex thread entered systemError' + }); + expect(session.thinking).toBe(false); + }); + + it('compacts the same thread before retrying context-window overflow', async () => { + harness.remainingThreadSystemErrors = 1; + harness.nextThreadSystemErrorMessage = "Codex ran out of room in the model's context window. Start a new thread or clear earlier history before retrying."; + const { session, sessionEvents } = createSessionStub(['first message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startThreadIds).toEqual(['thread-1']); + expect(harness.compactThreadIds).toEqual(['thread-1']); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-1']); + expect(harness.startTurnMessages).toEqual(['first message', 'first message']); + expect(session.sessionId).toBe('thread-1'); + expect(sessionEvents).not.toContainEqual({ + type: 'message', + message: "Task failed: Codex ran out of room in the model's context window. Start a new thread or clear earlier history before retrying." + }); + expect(session.thinking).toBe(false); + }); + + it('retries asynchronous thread-level systemError notifications on the same thread', async () => { + harness.remainingThreadSystemErrors = 1; + harness.deferThreadStatusNotifications = true; + const { session, sessionEvents } = createSessionStub(['first message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startThreadIds).toEqual(['thread-1']); + expect(harness.resumeThreadIds).toEqual([]); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-1']); + expect(harness.startTurnMessages).toEqual(['first message', 'first message']); + expect(session.sessionId).toBe('thread-1'); + expect(sessionEvents).not.toContainEqual({ + type: 'message', + message: 'Task failed: Codex thread entered systemError' + }); + expect(session.thinking).toBe(false); + }); + + it('compacts before retrying asynchronous context-window overflow notifications', async () => { + harness.remainingThreadSystemErrors = 1; + harness.deferThreadStatusNotifications = true; + harness.nextThreadSystemErrorMessage = "Codex ran out of room in the model's context window. Start a new thread or clear earlier history before retrying."; + const { session, sessionEvents } = createSessionStub(['first message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startThreadIds).toEqual(['thread-1']); + expect(harness.compactThreadIds).toEqual(['thread-1']); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-1']); + expect(harness.startTurnMessages).toEqual(['first message', 'first message']); + expect(session.sessionId).toBe('thread-1'); + expect(sessionEvents).not.toContainEqual({ + type: 'message', + message: "Task failed: Codex ran out of room in the model's context window. Start a new thread or clear earlier history before retrying." + }); + expect(session.thinking).toBe(false); + }); + + it('does not create a new thread when same-conversation compact fails', async () => { + harness.remainingThreadSystemErrors = 1; + harness.nextThreadSystemErrorMessage = "Codex ran out of room in the model's context window. Start a new thread or clear earlier history before retrying."; + harness.failNextCompact = true; + const { session, sessionEvents } = createSessionStub(['first message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startThreadIds).toEqual(['thread-1']); + expect(harness.compactThreadIds).toEqual(['thread-1']); + expect(harness.startTurnThreadIds).toEqual(['thread-1']); + expect(session.sessionId).toBe('thread-1'); + expect(sessionEvents).toContainEqual({ + type: 'message', + message: 'Task failed: context window overflow and same-conversation compact failed' + }); + expect(session.thinking).toBe(false); + }); + + it('keeps using the old thread for later messages after same-thread retries are exhausted', async () => { + harness.remainingThreadSystemErrors = 4; + const { session } = createSessionStub(['first message', 'second message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startThreadIds).toEqual(['thread-1']); + expect(harness.resumeThreadIds).toEqual([]); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-1', 'thread-1', 'thread-1', 'thread-1']); + expect(harness.startTurnMessages).toEqual(['first message', 'first message', 'first message', 'first message', 'second message']); + expect(session.sessionId).toBe('thread-1'); + expect(session.thinking).toBe(false); + }); + + it('does not create a new thread when an existing conversation cannot be resumed', async () => { + harness.failResumeThreadIds = ['thread-old']; + const { session, sessionEvents } = createSessionStub(['first message']); + session.sessionId = 'thread-old'; + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.resumeThreadIds).toEqual(['thread-old']); + expect(harness.startThreadIds).toEqual([]); + expect(harness.startTurnThreadIds).toEqual([]); + expect(session.sessionId).toBe('thread-old'); + expect(sessionEvents).toContainEqual({ + type: 'message', + message: 'Task failed: Codex conversation thread-old could not be resumed; no new conversation was created' + }); + expect(session.thinking).toBe(false); + }); + + it('does not start 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.startThreadIds).toEqual(['thread-1']); expect(harness.resumeThreadIds).toEqual([]); - expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-2']); - expect(session.sessionId).toBe('thread-2'); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-1', 'thread-1']); + expect(harness.startTurnMessages).toEqual(['first message', 'first message', 'second message']); + expect(session.sessionId).toBe('thread-1'); expect(session.thinking).toBe(false); }); diff --git a/cli/src/codex/codexRemoteLauncher.ts b/cli/src/codex/codexRemoteLauncher.ts index 64153060..cdfd1e7f 100644 --- a/cli/src/codex/codexRemoteLauncher.ts +++ b/cli/src/codex/codexRemoteLauncher.ts @@ -26,6 +26,35 @@ import { type HappyServer = Awaited>['server']; type QueuedMessage = { message: string; mode: EnhancedMode; isolate: boolean; hash: string }; +const SAME_THREAD_RETRYABLE_ERROR_PATTERNS = [ + 'selected model is at capacity', + 'codex thread entered systemerror' +]; +const CONTEXT_COMPACT_RETRYABLE_ERROR_PATTERNS = [ + 'ran out of room in the model', + 'context window', + 'clear earlier history' +]; +const SAME_THREAD_MAX_RETRIES = 3; +const SAME_THREAD_MAX_COMPACT_RETRIES = 1; +const SAME_THREAD_COMPACT_TIMEOUT_MS = 10 * 60 * 1000; + +function isSameThreadRetryableCodexError(error: string | null): boolean { + if (!error) { + return false; + } + const normalized = error.toLowerCase(); + return SAME_THREAD_RETRYABLE_ERROR_PATTERNS.some((pattern) => normalized.includes(pattern)); +} + +function isContextCompactRetryableCodexError(error: string | null): boolean { + if (!error) { + return false; + } + const normalized = error.toLowerCase(); + return CONTEXT_COMPACT_RETRYABLE_ERROR_PATTERNS.some((pattern) => normalized.includes(pattern)); +} + class CodexRemoteLauncher extends RemoteLauncherBase { private readonly session: CodexSession; private readonly appServerClient: CodexAppServerClient; @@ -243,6 +272,117 @@ class CodexRemoteLauncher extends RemoteLauncherBase { let turnInFlight = false; let allowAnonymousTerminalEvent = false; let invalidThreadId: string | null = null; + let activeMessage: QueuedMessage | null = null; + let sameThreadRetryAttempt = 0; + let sameThreadCompactAttempt = 0; + let recoveryInFlight = false; + let compactRecovery: { + threadId: string; + message: QueuedMessage; + timeout: ReturnType | null; + } | null = null; + let loopWakeWaiter: (() => void) | null = null; + + const wakeLoop = () => { + const waiter = loopWakeWaiter; + if (!waiter) { + return; + } + loopWakeWaiter = null; + waiter(); + }; + + const waitForTurnOrRecovery = (signal: AbortSignal): Promise => new Promise((resolve) => { + if (!turnInFlight && !recoveryInFlight) { + resolve(); + return; + } + + const finish = () => { + if (loopWakeWaiter === finish) { + loopWakeWaiter = null; + } + signal.removeEventListener('abort', finish); + resolve(); + }; + + loopWakeWaiter = finish; + signal.addEventListener('abort', finish, { once: true }); + }); + + const clearCompactRecovery = (recovery: typeof compactRecovery) => { + if (!recovery) { + return; + } + if (recovery.timeout) { + clearTimeout(recovery.timeout); + } + if (compactRecovery === recovery) { + compactRecovery = null; + } + recoveryInFlight = false; + wakeLoop(); + }; + + const failCompactRecovery = (recovery: typeof compactRecovery, message: string) => { + if (!recovery || compactRecovery !== recovery) { + return; + } + logger.warn(`[Codex] ${message}`); + messageBuffer.addMessage(message, 'status'); + session.sendSessionEvent({ type: 'message', message }); + activeMessage = null; + clearCompactRecovery(recovery); + }; + + const completeCompactRecovery = (threadId: string | null) => { + const recovery = compactRecovery; + if (!recovery) { + return false; + } + if (!threadId || threadId !== recovery.threadId) { + return false; + } + if (!this.shouldExit && this.currentThreadId === recovery.threadId) { + pending = recovery.message; + const message = 'Context compacted; retrying same conversation'; + messageBuffer.addMessage(message, 'status'); + session.sendSessionEvent({ type: 'message', message }); + } + clearCompactRecovery(recovery); + return true; + }; + + const beginCompactRecovery = (threadId: string, messageToRetry: QueuedMessage, error: string | null) => { + sameThreadCompactAttempt += 1; + recoveryInFlight = true; + const recovery = { + threadId, + message: messageToRetry, + timeout: null as ReturnType | null + }; + compactRecovery = recovery; + recovery.timeout = setTimeout(() => { + failCompactRecovery( + recovery, + 'Task failed: context window overflow and same-conversation compact timed out' + ); + }, SAME_THREAD_COMPACT_TIMEOUT_MS); + recovery.timeout.unref?.(); + + logger.debug( + `[Codex] Compacting retryable context failure on same thread ` + + `(attempt ${sameThreadCompactAttempt}/${SAME_THREAD_MAX_COMPACT_RETRIES}): ${error ?? 'unknown error'}` + ); + void appServerClient.compactThread({ threadId }, { signal: this.abortController.signal }) + .catch((compactError) => { + logger.warn('[Codex] Failed to start app-server thread compact before retry:', compactError); + failCompactRecovery( + recovery, + 'Task failed: context window overflow and same-conversation compact failed' + ); + }); + }; const handleCodexEvent = (msg: Record) => { const msgType = asString(msg.type); @@ -260,6 +400,11 @@ class CodexRemoteLauncher extends RemoteLauncherBase { return; } + if (msgType === 'thread_compacted') { + completeCompactRecovery(eventThreadId); + return; + } + if (msgType === 'task_started') { const turnId = eventTurnId; if (turnId) { @@ -271,6 +416,18 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } const isThreadStatusFailure = msgType === 'task_failed' && msg.terminal_source === 'thread_status'; + const error = msgType === 'task_failed' ? asString(msg.error) : null; + const shouldCompactAndRetrySameThread = msgType === 'task_failed' + && isContextCompactRetryableCodexError(error) + && Boolean(activeMessage) + && Boolean(this.currentThreadId) + && sameThreadCompactAttempt < SAME_THREAD_MAX_COMPACT_RETRIES; + const shouldRetrySameThread = msgType === 'task_failed' + && !shouldCompactAndRetrySameThread + && isSameThreadRetryableCodexError(error) + && Boolean(activeMessage) + && Boolean(this.currentThreadId) + && sameThreadRetryAttempt < SAME_THREAD_MAX_RETRIES; if (isTerminalEvent) { if (shouldIgnoreTerminalEvent({ @@ -290,12 +447,24 @@ class CodexRemoteLauncher extends RemoteLauncherBase { ); return; } + if (shouldCompactAndRetrySameThread) { + const threadId = this.currentThreadId; + const messageToRetry = activeMessage; + if (threadId && messageToRetry) { + beginCompactRecovery(threadId, messageToRetry, error); + } + } else if (shouldRetrySameThread) { + sameThreadRetryAttempt += 1; + pending = activeMessage; + logger.debug( + `[Codex] Retrying retryable failure on same thread ` + + `(attempt ${sameThreadRetryAttempt}/${SAME_THREAD_MAX_RETRIES}): ${error ?? 'unknown error'}` + ); + } this.currentTurnId = null; allowAnonymousTerminalEvent = false; - if (isThreadStatusFailure) { - invalidThreadId = eventThreadId ?? this.currentThreadId; - this.currentThreadId = null; - hasThread = false; + if (isThreadStatusFailure && !shouldRetrySameThread && !shouldCompactAndRetrySameThread) { + logger.warn(`[Codex] Thread-level failure on ${eventThreadId ?? this.currentThreadId ?? 'unknown thread'}; preserving same conversation`); } } @@ -327,10 +496,23 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } else if (msgType === 'turn_aborted') { messageBuffer.addMessage('Turn aborted', 'status'); } else if (msgType === 'task_failed') { - const error = asString(msg.error); - const message = error ? `Task failed: ${error}` : 'Task failed'; - messageBuffer.addMessage(message, 'status'); - session.sendSessionEvent({ type: 'message', message }); + if (shouldCompactAndRetrySameThread) { + const retryMessage = error + ? `Task failed: ${error}; compacting same conversation before retry (${sameThreadCompactAttempt}/${SAME_THREAD_MAX_COMPACT_RETRIES})` + : `Task failed; compacting same conversation before retry (${sameThreadCompactAttempt}/${SAME_THREAD_MAX_COMPACT_RETRIES})`; + messageBuffer.addMessage(retryMessage, 'status'); + session.sendSessionEvent({ type: 'message', message: retryMessage }); + } else if (shouldRetrySameThread) { + const retryMessage = error + ? `Task failed: ${error}; retrying same conversation (${sameThreadRetryAttempt}/${SAME_THREAD_MAX_RETRIES})` + : `Task failed; retrying same conversation (${sameThreadRetryAttempt}/${SAME_THREAD_MAX_RETRIES})`; + messageBuffer.addMessage(retryMessage, 'status'); + session.sendSessionEvent({ type: 'message', message: retryMessage }); + } else { + const message = error ? `Task failed: ${error}` : 'Task failed'; + messageBuffer.addMessage(message, 'status'); + session.sendSessionEvent({ type: 'message', message }); + } } if (msgType === 'task_started') { @@ -353,6 +535,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } diffProcessor.reset(); appServerEventConverter.reset(); + wakeLoop(); } if (isTerminalEvent && !turnInFlight) { @@ -361,6 +544,14 @@ class CodexRemoteLauncher extends RemoteLauncherBase { scheduleReadyAfterTurn?.(); } + if (msgType === 'task_complete') { + sameThreadRetryAttempt = 0; + sameThreadCompactAttempt = 0; + recoveryInFlight = false; + clearCompactRecovery(compactRecovery); + activeMessage = null; + } + if (msgType === 'agent_reasoning_section_break') { reasoningProcessor.handleSectionBreak(); } @@ -625,7 +816,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { readyAfterTurnTimer = setTimeout(() => { readyAfterTurnTimer = null; emitReadyIfIdle({ - pending, + pending: pending ?? (recoveryInFlight ? activeMessage : null), queueSize: () => session.queue.size(), shouldExit: this.shouldExit, sendReady @@ -753,9 +944,22 @@ class CodexRemoteLauncher extends RemoteLauncherBase { while (!this.shouldExit) { logActiveHandles('loop-top'); + if (!pending && (turnInFlight || recoveryInFlight) && session.queue.size() === 0) { + await waitForTurnOrRecovery(this.abortController.signal); + if (this.abortController.signal.aborted && !this.shouldExit) { + logger.debug('[codex]: Internal wait aborted while turn/recovery was active; continuing'); + continue; + } + continue; + } + let message: QueuedMessage | null = pending; + const isRetryMessage = Boolean(message); pending = null; if (!message) { + sameThreadRetryAttempt = 0; + sameThreadCompactAttempt = 0; + activeMessage = null; const waitSignal = this.abortController.signal; const batch = await session.queue.waitForMessagesAndGetAsString(waitSignal); if (!batch) { @@ -773,7 +977,10 @@ class CodexRemoteLauncher extends RemoteLauncherBase { break; } - messageBuffer.addMessage(message.message, 'user'); + if (!isRetryMessage) { + messageBuffer.addMessage(message.message, 'user'); + } + activeMessage = message; try { if (await handleSpecialCommand(message)) { @@ -788,9 +995,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { cliOverrides: session.codexCliOverrides }); - const resumeCandidate = session.sessionId && session.sessionId !== invalidThreadId - ? session.sessionId - : null; + const resumeCandidate = session.sessionId ?? null; let threadId: string | null = null; if (resumeCandidate) { @@ -807,7 +1012,12 @@ class CodexRemoteLauncher extends RemoteLauncherBase { applyResolvedModel(resumeRecord?.model); logger.debug(`[Codex] Resumed app-server thread ${threadId}`); } catch (error) { - logger.warn(`[Codex] Failed to resume app-server thread ${resumeCandidate}, starting new thread`, error); + logger.warn(`[Codex] Failed to resume app-server thread ${resumeCandidate}; preserving old conversation boundary`, error); + const failureMessage = `Task failed: Codex conversation ${resumeCandidate} could not be resumed; no new conversation was created`; + messageBuffer.addMessage(failureMessage, 'status'); + session.sendSessionEvent({ type: 'message', message: failureMessage }); + pending = null; + continue; } } @@ -891,7 +1101,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { session.onThinkingChange(false); clearReadyAfterTurnTimer?.(); emitReadyIfIdle({ - pending, + pending: pending ?? (recoveryInFlight ? activeMessage : null), queueSize: () => session.queue.size(), shouldExit: this.shouldExit, sendReady diff --git a/cli/src/codex/utils/appServerEventConverter.test.ts b/cli/src/codex/utils/appServerEventConverter.test.ts index a3f11756..369a9927 100644 --- a/cli/src/codex/utils/appServerEventConverter.test.ts +++ b/cli/src/codex/utils/appServerEventConverter.test.ts @@ -389,4 +389,41 @@ describe('AppServerEventConverter', () => { expect(events).toEqual([{ type: 'task_failed', error: 'fatal' }]); }); + + it('maps thread/compacted notifications', () => { + const converter = new AppServerEventConverter(); + const events = converter.handleNotification('thread/compacted', { + threadId: 'thread-1', + turnId: 'turn-compact' + }); + + expect(events).toEqual([{ + type: 'thread_compacted', + thread_id: 'thread-1', + turn_id: 'turn-compact' + }]); + }); + + it('ignores compacted notifications without thread ids', () => { + const converter = new AppServerEventConverter(); + + expect(converter.handleNotification('thread/compacted', { turnId: 'turn-compact' })).toEqual([]); + expect(converter.handleNotification('codex/event/context_compacted', { + msg: { type: 'context_compacted', turn_id: 'turn-compact' } + })).toEqual([]); + }); + + it('unwraps context_compacted events', () => { + const converter = new AppServerEventConverter(); + const events = converter.handleNotification('codex/event/context_compacted', { + msg: { type: 'context_compacted', thread_id: 'thread-1', turn_id: 'turn-compact' } + }); + + expect(events).toEqual([{ + type: 'thread_compacted', + thread_id: 'thread-1', + turn_id: 'turn-compact' + }]); + }); + }); diff --git a/cli/src/codex/utils/appServerEventConverter.ts b/cli/src/codex/utils/appServerEventConverter.ts index 79038233..ecb5a835 100644 --- a/cli/src/codex/utils/appServerEventConverter.ts +++ b/cli/src/codex/utils/appServerEventConverter.ts @@ -276,13 +276,25 @@ export class AppServerEventConverter { return extractPlanUpdate(msg); } + if (msgType === 'context_compacted') { + const threadId = asString(msg.thread_id ?? msg.threadId); + if (!threadId) { + return []; + } + const turnId = asString(msg.turn_id ?? msg.turnId); + return [{ + type: 'thread_compacted', + thread_id: threadId, + ...(turnId ? { turn_id: turnId } : {}) + }]; + } + if ( msgType === 'mcp_startup_update' || msgType === 'mcp_startup_complete' || msgType === 'skills_update_available' || msgType === 'stream_error' || msgType === 'warning' || - msgType === 'context_compacted' || msgType === 'terminal_interaction' || msgType === 'user_message' ) { @@ -304,7 +316,21 @@ export class AppServerEventConverter { return extractPlanUpdate(paramsRecord); } - if (method === 'account/rateLimits/updated' || method === 'thread/compacted') { + if (method === 'account/rateLimits/updated') { + return events; + } + + if (method === 'thread/compacted') { + const threadId = asString(paramsRecord.threadId ?? paramsRecord.thread_id); + if (!threadId) { + return events; + } + const turnId = asString(paramsRecord.turnId ?? paramsRecord.turn_id); + events.push({ + type: 'thread_compacted', + thread_id: threadId, + ...(turnId ? { turn_id: turnId } : {}) + }); return events; } diff --git a/cli/src/utils/spawnHappyCLI.test.ts b/cli/src/utils/spawnHappyCLI.test.ts index 69c2b7a2..bfc097e8 100644 --- a/cli/src/utils/spawnHappyCLI.test.ts +++ b/cli/src/utils/spawnHappyCLI.test.ts @@ -1,7 +1,17 @@ import { beforeAll, afterAll, beforeEach, describe, expect, it, vi } from 'vitest'; import type { SpawnOptions } from 'child_process'; -const spawnMock = vi.fn((..._args: any[]) => ({ pid: 12345 } as any)); +const { + spawnMock, + existsSyncMock, + isBunCompiledMock, + projectPathMock +} = vi.hoisted(() => ({ + spawnMock: vi.fn((..._args: any[]) => ({ pid: 12345 }) as any), + existsSyncMock: vi.fn((path: string) => !path.includes('missing-hapi.exe')), + isBunCompiledMock: vi.fn(() => false), + projectPathMock: vi.fn(() => process.cwd()) +})); vi.mock('child_process', async () => { const actual = await vi.importActual('child_process'); @@ -11,8 +21,22 @@ vi.mock('child_process', async () => { }; }); +vi.mock('node:fs', async () => { + const actual = await vi.importActual('node:fs'); + return { + ...actual, + existsSync: existsSyncMock + }; +}); + +vi.mock('@/projectPath', () => ({ + isBunCompiled: isBunCompiledMock, + projectPath: projectPathMock +})); + const originalPlatformDescriptor = Object.getOwnPropertyDescriptor(process, 'platform'); const originalInvokedCwd = process.env.HAPI_INVOKED_CWD; +const originalCliExecutable = process.env.HAPI_CLI_EXECUTABLE; function setPlatform(value: string) { Object.defineProperty(process, 'platform', { @@ -40,11 +64,20 @@ describe('spawnHappyCLI windowsHide behavior', () => { beforeEach(() => { vi.clearAllMocks(); + vi.resetModules(); + existsSyncMock.mockImplementation((path: string) => !path.includes('missing-hapi.exe')); + isBunCompiledMock.mockReturnValue(false); + projectPathMock.mockReturnValue(process.cwd()); if (originalInvokedCwd === undefined) { delete process.env.HAPI_INVOKED_CWD; } else { process.env.HAPI_INVOKED_CWD = originalInvokedCwd; } + if (originalCliExecutable === undefined) { + delete process.env.HAPI_CLI_EXECUTABLE; + } else { + process.env.HAPI_CLI_EXECUTABLE = originalCliExecutable; + } }); afterAll(() => { @@ -104,13 +137,61 @@ describe('spawnHappyCLI windowsHide behavior', () => { expect(command.command).toBe(process.execPath); if (isBunRuntime) { expect(command.args[0]).toBe('--cwd'); - expect(command.args[1].replace(/\\/g, '/')).toMatch(/\/hapi\/cli$/); - expect(command.args[2].replace(/\\/g, '/')).toMatch(/\/hapi\/cli\/src\/index\.ts$/); + expect(command.args[1].replace(/\\/g, '/')).toMatch(/\/cli$/); + expect(command.args[2].replace(/\\/g, '/')).toMatch(/\/cli\/src\/index\.ts$/); } else { - expect(command.args.some((arg) => arg.replace(/\\/g, '/').endsWith('/hapi/cli/src/index.ts'))).toBe(true); + expect(command.args.some((arg) => arg.replace(/\\/g, '/').endsWith('/cli/src/index.ts'))).toBe(true); } }); + it('uses an inherited compiled CLI executable override when it points to an existing binary', async () => { + isBunCompiledMock.mockReturnValue(true); + process.env.HAPI_CLI_EXECUTABLE = 'C:\\Users\\Administrator\\.hapi\\patched\\hapi.exe'; + const { getHappyCliCommand, resolveHappyCliExecutable } = await import('./spawnHappyCLI'); + + const command = getHappyCliCommand(['mcp', '--url', 'http://127.0.0.1:1234/']); + + expect(resolveHappyCliExecutable()).toBe(process.env.HAPI_CLI_EXECUTABLE); + expect(command.command).toBe(process.env.HAPI_CLI_EXECUTABLE); + }); + + it('falls back to a real argv0 executable before process.execPath in compiled mode', async () => { + isBunCompiledMock.mockReturnValue(true); + const previousArgv0 = process.argv[0]; + process.argv[0] = 'C:\\Users\\Administrator\\.hapi\\patched\\resume-recovery-0.17.2\\hapi.exe'; + const { resolveHappyCliExecutable } = await import('./spawnHappyCLI'); + + try { + expect(resolveHappyCliExecutable()).toBe(process.argv[0]); + } finally { + process.argv[0] = previousArgv0; + } + }); + + it('ignores an inherited compiled CLI executable override when the binary is missing', async () => { + isBunCompiledMock.mockReturnValue(true); + process.env.HAPI_CLI_EXECUTABLE = 'C:\\Users\\Administrator\\.hapi\\patched\\missing-hapi.exe'; + const { getHappyCliCommand } = await import('./spawnHappyCLI'); + + const command = getHappyCliCommand(['mcp', '--url', 'http://127.0.0.1:1234/']); + + expect(command.command).toBe(process.execPath); + }); + + it('passes the resolved compiled executable to child HAPI processes', async () => { + isBunCompiledMock.mockReturnValue(true); + process.env.HAPI_CLI_EXECUTABLE = 'C:\\Users\\Administrator\\.hapi\\patched\\hapi.exe'; + const { spawnHappyCLI } = await import('./spawnHappyCLI'); + + spawnHappyCLI(['mcp', '--url', 'http://127.0.0.1:1234/'], { + stdio: 'ignore' + }); + + const [command, _args, options] = spawnMock.mock.calls[0] as unknown[] | undefined ?? []; + expect(command).toBe(process.env.HAPI_CLI_EXECUTABLE); + expect((options as SpawnOptions | undefined)?.env?.HAPI_CLI_EXECUTABLE).toBe(process.env.HAPI_CLI_EXECUTABLE); + }); + it('passes invoked workspace cwd to child processes when cwd is provided', async () => { const { spawnHappyCLI } = await import('./spawnHappyCLI'); const childCwd = 'C:\\workspace\\project'; diff --git a/cli/src/utils/spawnHappyCLI.ts b/cli/src/utils/spawnHappyCLI.ts index 1f26d831..0f207fd7 100644 --- a/cli/src/utils/spawnHappyCLI.ts +++ b/cli/src/utils/spawnHappyCLI.ts @@ -32,6 +32,8 @@ import { isBunCompiled, projectPath } from '@/projectPath'; import { logger } from '@/ui/logger'; import { existsSync } from 'node:fs'; +const HAPI_CLI_EXECUTABLE_ENV = 'HAPI_CLI_EXECUTABLE'; + /** * Resolve the TypeScript entrypoint for development mode. */ @@ -71,11 +73,30 @@ function resolveInvokedCwd(cwd: SpawnOptions['cwd']): string { return process.cwd(); } +export function resolveHappyCliExecutable(): string { + const override = process.env[HAPI_CLI_EXECUTABLE_ENV]?.trim(); + if (override && isCrossPlatformAbsolutePath(override) && existsSync(override)) { + return override; + } + + const argv0 = process.argv[0]?.trim(); + if (argv0 && isCrossPlatformAbsolutePath(argv0) && existsSync(argv0)) { + return argv0; + } + + const bunArgv0 = globalThis.Bun?.argv?.[0]?.trim(); + if (bunArgv0 && isCrossPlatformAbsolutePath(bunArgv0) && existsSync(bunArgv0)) { + return bunArgv0; + } + + return process.execPath; +} + export function getHappyCliCommand(args: string[]): HappyCliCommand { // Compiled binary mode: just use the executable directly if (isBunCompiled()) { return { - command: process.execPath, + command: resolveHappyCliExecutable(), args }; } @@ -118,10 +139,11 @@ export function spawnHappyCLI(args: string[], options: SpawnOptions = {}): Child const fullCommand = `hapi ${args.join(' ')}`; logger.debug(`[SPAWN HAPI CLI] Spawning: ${fullCommand} in ${directory}`); + const compiledMode = isBunCompiled(); const { command: spawnCommand, args: spawnArgs } = getHappyCliCommand(args); // Sanity check that the entrypoint path exists - if (!isBunCompiled()) { + if (!compiledMode) { const entrypoint = spawnArgs.find((arg) => arg.endsWith('index.ts')); if (entrypoint && !existsSync(entrypoint)) { const errorMessage = `Entrypoint ${entrypoint} does not exist`; @@ -133,8 +155,12 @@ export function spawnHappyCLI(args: string[], options: SpawnOptions = {}): Child // On Windows, detached processes allocate a new console window by default. // windowsHide: true suppresses this to prevent cmd windows from accumulating. const finalOptions: SpawnOptions = { ...options }; - if (!isBunCompiled()) { - const finalEnv = { ...process.env, ...options.env }; + const finalEnv = { ...process.env, ...options.env }; + let shouldSetEnv = false; + if (compiledMode) { + finalEnv[HAPI_CLI_EXECUTABLE_ENV] = spawnCommand; + shouldSetEnv = true; + } else { const invokedCwd = finalEnv.HAPI_INVOKED_CWD?.trim(); const hasExplicitCwd = 'cwd' in options && options.cwd !== undefined; finalEnv.HAPI_INVOKED_CWD = hasExplicitCwd @@ -142,6 +168,9 @@ export function spawnHappyCLI(args: string[], options: SpawnOptions = {}): Child : invokedCwd && isCrossPlatformAbsolutePath(invokedCwd) ? invokedCwd : resolveInvokedCwd(options.cwd); + shouldSetEnv = true; + } + if (shouldSetEnv) { finalOptions.env = finalEnv; } if (process.platform === 'win32' && options.detached) { diff --git a/hub/src/sync/messageService.ts b/hub/src/sync/messageService.ts index 07425861..82a76550 100644 --- a/hub/src/sync/messageService.ts +++ b/hub/src/sync/messageService.ts @@ -13,6 +13,17 @@ export class MessageService { ) { } + getMessages(sessionId: string, limit: number = 200): DecryptedMessage[] { + const stored = this.store.messages.getMessages(sessionId, limit) + return stored.map((message) => ({ + id: message.id, + seq: message.seq, + localId: message.localId, + content: message.content, + createdAt: message.createdAt + })) + } + getMessagesPage(sessionId: string, options: { limit: number; beforeSeq: number | null }): { messages: DecryptedMessage[] page: { diff --git a/hub/src/sync/rpcGateway.test.ts b/hub/src/sync/rpcGateway.test.ts new file mode 100644 index 00000000..705d35a1 --- /dev/null +++ b/hub/src/sync/rpcGateway.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, it } from 'bun:test' +import type { Server } from 'socket.io' +import type { RpcRegistry } from '../socket/rpcRegistry' +import { RpcGateway } from './rpcGateway' + +function createGateway() { + const timeouts: number[] = [] + const socket = { + timeout(timeoutMs: number) { + timeouts.push(timeoutMs) + return { + async emitWithAck(_event: string, payload: { method: string; params: string }) { + return JSON.stringify({ + success: true, + method: payload.method, + params: JSON.parse(payload.params) as unknown + }) + } + } + } + } + + const io = { + of() { + return { + sockets: { + get() { + return socket + } + } + } + } + } as unknown as Server + + const rpcRegistry = { + getSocketIdForMethod() { + return 'socket-1' + } + } as unknown as RpcRegistry + + return { + gateway: new RpcGateway(io, rpcRegistry), + timeouts + } +} + +describe('RpcGateway RPC timeouts', () => { + it('uses the default RPC timeout for regular machine RPCs', async () => { + const { gateway, timeouts } = createGateway() + + await gateway.listMachineDirectory('machine-1', 'C:\\workspace') + + expect(timeouts).toEqual([30_000]) + }) + + it('uses an extended RPC timeout when listing Codex models', async () => { + const { gateway, timeouts } = createGateway() + + await gateway.listCodexModelsForMachine('machine-1') + + expect(timeouts).toEqual([120_000]) + }) +}) + diff --git a/hub/src/sync/rpcGateway.ts b/hub/src/sync/rpcGateway.ts index 977b4ead..34856e33 100644 --- a/hub/src/sync/rpcGateway.ts +++ b/hub/src/sync/rpcGateway.ts @@ -2,6 +2,9 @@ import type { CodexCollaborationMode, PermissionMode } from '@hapi/protocol/type import type { Server } from 'socket.io' import type { RpcRegistry } from '../socket/rpcRegistry' +const DEFAULT_RPC_TIMEOUT_MS = 30_000 +const MODEL_LIST_RPC_TIMEOUT_MS = 120_000 + export type RpcCommandResponse = { success: boolean stdout?: string @@ -267,11 +270,11 @@ export class RpcGateway { } async listCodexModelsForSession(sessionId: string): Promise { - return await this.sessionRpc(sessionId, 'listCodexModels', {}) as RpcListCodexModelsResponse + return await this.sessionRpc(sessionId, 'listCodexModels', {}, MODEL_LIST_RPC_TIMEOUT_MS) as RpcListCodexModelsResponse } async listCodexModelsForMachine(machineId: string): Promise { - return await this.machineRpc(machineId, 'listCodexModels', {}) as RpcListCodexModelsResponse + return await this.machineRpc(machineId, 'listCodexModels', {}, MODEL_LIST_RPC_TIMEOUT_MS) as RpcListCodexModelsResponse } async listOpencodeModelsForSession(sessionId: string): Promise { @@ -282,15 +285,25 @@ export class RpcGateway { return await this.machineRpc(machineId, 'listOpencodeModelsForCwd', { cwd }) as RpcListOpencodeModelsResponse } - private async sessionRpc(sessionId: string, method: string, params: unknown): Promise { - return await this.rpcCall(`${sessionId}:${method}`, params) + private async sessionRpc( + sessionId: string, + method: string, + params: unknown, + timeoutMs: number = DEFAULT_RPC_TIMEOUT_MS + ): Promise { + return await this.rpcCall(`${sessionId}:${method}`, params, timeoutMs) } - private async machineRpc(machineId: string, method: string, params: unknown): Promise { - return await this.rpcCall(`${machineId}:${method}`, params) + private async machineRpc( + machineId: string, + method: string, + params: unknown, + timeoutMs: number = DEFAULT_RPC_TIMEOUT_MS + ): Promise { + return await this.rpcCall(`${machineId}:${method}`, params, timeoutMs) } - private async rpcCall(method: string, params: unknown): Promise { + private async rpcCall(method: string, params: unknown, timeoutMs: number = DEFAULT_RPC_TIMEOUT_MS): Promise { const socketId = this.rpcRegistry.getSocketIdForMethod(method) if (!socketId) { throw new Error(`RPC handler not registered: ${method}`) @@ -301,7 +314,7 @@ export class RpcGateway { throw new Error(`RPC socket disconnected: ${method}`) } - const response = await socket.timeout(30_000).emitWithAck('rpc-request', { + const response = await socket.timeout(timeoutMs).emitWithAck('rpc-request', { method, params: JSON.stringify(params) }) as unknown diff --git a/hub/src/sync/sessionModel.test.ts b/hub/src/sync/sessionModel.test.ts index f743fe65..cfb4bb44 100644 --- a/hub/src/sync/sessionModel.test.ts +++ b/hub/src/sync/sessionModel.test.ts @@ -605,6 +605,199 @@ describe('session model', () => { } }) + it('recovers claude resume session ID from stored messages when metadata is missing it', async () => { + const store = new Store(':memory:') + const engine = new SyncEngine( + store, + {} as never, + new RpcRegistry(), + { broadcast() {} } as never + ) + + try { + const session = engine.getOrCreateSession( + 'session-claude-resume-from-message', + { + path: '/tmp/project', + host: 'localhost', + machineId: 'machine-1', + flavor: 'claude' + }, + null, + 'default', + 'sonnet' + ) + store.messages.addMessage(session.id, { + role: 'agent', + content: { + type: 'output', + data: { + type: 'assistant', + sessionId: '7f5cd4ee-3a76-4601-a7b4-f9eb976bf515' + } + } + }) + engine.getOrCreateMachine( + 'machine-1', + { host: 'localhost', platform: 'linux', happyCliVersion: '0.1.0' }, + null, + 'default' + ) + engine.handleMachineAlive({ machineId: 'machine-1', time: Date.now() }) + + let capturedResumeSessionId: string | undefined + ;(engine as any).rpcGateway.spawnSession = async ( + _machineId: string, + _directory: string, + _agent: string, + _model?: string, + _modelReasoningEffort?: string, + _yolo?: boolean, + _sessionType?: 'simple' | 'worktree', + _worktreeName?: string, + resumeSessionId?: string + ) => { + capturedResumeSessionId = resumeSessionId + return { type: 'success', sessionId: session.id } + } + ;(engine as any).waitForSessionActive = async () => true + + const result = await engine.resumeSession(session.id, 'default') + + expect(result).toEqual({ type: 'success', sessionId: session.id }) + expect(capturedResumeSessionId).toBe('7f5cd4ee-3a76-4601-a7b4-f9eb976bf515') + expect(store.sessions.getSession(session.id)?.metadata).toMatchObject({ + claudeSessionId: '7f5cd4ee-3a76-4601-a7b4-f9eb976bf515' + }) + } finally { + engine.stop() + } + }) + + + it('recovers the newest claude session ID when stored messages contain multiple IDs', async () => { + const store = new Store(':memory:') + const engine = new SyncEngine( + store, + {} as never, + new RpcRegistry(), + { broadcast() {} } as never + ) + + try { + const session = engine.getOrCreateSession( + 'session-claude-resume-newest-message', + { + path: '/tmp/project', + host: 'localhost', + machineId: 'machine-1', + flavor: 'claude' + }, + null, + 'default', + 'sonnet' + ) + store.messages.addMessage(session.id, { + role: 'agent', + content: { + type: 'output', + data: { + type: 'assistant', + sessionId: '11111111-1111-4111-8111-111111111111' + } + } + }) + store.messages.addMessage(session.id, { + role: 'agent', + content: { + type: 'output', + data: { + type: 'assistant', + sessionId: '22222222-2222-4222-8222-222222222222' + } + } + }) + engine.getOrCreateMachine( + 'machine-1', + { host: 'localhost', platform: 'linux', happyCliVersion: '0.1.0' }, + null, + 'default' + ) + engine.handleMachineAlive({ machineId: 'machine-1', time: Date.now() }) + + let capturedResumeSessionId: string | undefined + ;(engine as any).rpcGateway.spawnSession = async ( + _machineId: string, + _directory: string, + _agent: string, + _model?: string, + _modelReasoningEffort?: string, + _yolo?: boolean, + _sessionType?: 'simple' | 'worktree', + _worktreeName?: string, + resumeSessionId?: string + ) => { + capturedResumeSessionId = resumeSessionId + return { type: 'success', sessionId: session.id } + } + ;(engine as any).waitForSessionActive = async () => true + + const result = await engine.resumeSession(session.id, 'default') + + expect(result).toEqual({ type: 'success', sessionId: session.id }) + expect(capturedResumeSessionId).toBe('22222222-2222-4222-8222-222222222222') + expect(store.sessions.getSession(session.id)?.metadata).toMatchObject({ + claudeSessionId: '22222222-2222-4222-8222-222222222222' + }) + } finally { + engine.stop() + } + }) + + it('does not recover a non-UUID sessionId from stored messages', async () => { + const store = new Store(':memory:') + const engine = new SyncEngine( + store, + {} as never, + new RpcRegistry(), + { broadcast() {} } as never + ) + + try { + const session = engine.getOrCreateSession( + 'session-claude-resume-no-token', + { + path: '/tmp/project', + host: 'localhost', + machineId: 'machine-1', + flavor: 'claude' + }, + null, + 'default', + 'sonnet' + ) + store.messages.addMessage(session.id, { + role: 'agent', + content: { + type: 'output', + data: { + sessionId: 'hapi-session-id-not-claude-uuid' + } + } + }) + + const result = await engine.resumeSession(session.id, 'default') + + expect(result).toEqual({ + type: 'error', + message: 'Resume session ID unavailable', + code: 'resume_unavailable' + }) + } finally { + engine.stop() + } + }) + it('passes the cached permissionMode when respawning a resumed session', async () => { const store = new Store(':memory:') const engine = new SyncEngine( diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index 8fffc0b6..bc883607 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -59,7 +59,7 @@ export class SyncEngine { private inactivityTimer: NodeJS.Timeout | null = null constructor( - store: Store, + private readonly store: Store, io: Server, rpcRegistry: RpcRegistry, sseManager: SSEManager @@ -458,7 +458,7 @@ export class SyncEngine { ? metadata.opencodeSessionId : flavor === 'cursor' ? metadata.cursorSessionId - : metadata.claudeSessionId + : (metadata.claudeSessionId ?? this.recoverClaudeSessionIdFromMessages(access.sessionId, namespace)) if (!resumeToken) { return { type: 'error', message: 'Resume session ID unavailable', code: 'resume_unavailable' } @@ -527,6 +527,78 @@ export class SyncEngine { return { type: 'success', sessionId: spawnResult.sessionId } } + private recoverClaudeSessionIdFromMessages(sessionId: string, namespace: string): string | null { + const messages = this.messageService.getMessages(sessionId, 200) + for (let i = messages.length - 1; i >= 0; i -= 1) { + const found = this.extractClaudeSessionId(messages[i].content) + if (!found) continue + + this.persistRecoveredClaudeSessionId(sessionId, namespace, found) + return found + } + return null + } + + private extractClaudeSessionId(value: unknown): string | null { + if (!value || typeof value !== 'object') { + return null + } + + const obj = value as Record + const direct = this.normalizeClaudeSessionId(obj.session_id) ?? this.normalizeClaudeSessionId(obj.sessionId) + if (direct) { + return direct + } + + const content = obj.content + if (content && typeof content === 'object') { + const found = this.extractClaudeSessionId(content) + if (found) return found + } + + const data = obj.data + if (data && typeof data === 'object') { + const found = this.extractClaudeSessionId(data) + if (found) return found + } + + return null + } + + private normalizeClaudeSessionId(value: unknown): string | null { + if (typeof value !== 'string') { + return null + } + const trimmed = value.trim() + return /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(trimmed) + ? trimmed + : null + } + + private persistRecoveredClaudeSessionId(sessionId: string, namespace: string, claudeSessionId: string): void { + for (let attempt = 0; attempt < 2; attempt += 1) { + const latest = this.sessionCache.getSessionByNamespace(sessionId, namespace) + ?? this.sessionCache.refreshSession(sessionId) + if (!latest?.metadata) return + if (latest.metadata.claudeSessionId === claudeSessionId) return + + const result = this.store.sessions.updateSessionMetadata( + sessionId, + { ...latest.metadata, claudeSessionId }, + latest.metadataVersion, + namespace, + { touchUpdatedAt: false } + ) + if (result.result === 'success') { + this.sessionCache.refreshSession(sessionId) + return + } + if (result.result !== 'version-mismatch') { + return + } + } + } + private hasSameAgentSessionIds( prev: Session['metadata'] | null, next: NonNullable diff --git a/hub/src/web/routes/sessions.test.ts b/hub/src/web/routes/sessions.test.ts index fc29e293..84c670ae 100644 --- a/hub/src/web/routes/sessions.test.ts +++ b/hub/src/web/routes/sessions.test.ts @@ -52,6 +52,7 @@ function createSession(overrides?: Partial): Session { function createApp(session: Session, opts?: { resumeSession?: (sessionId: string, namespace: string, resumeOpts?: { permissionMode?: string }) => Promise<{ type: string; sessionId?: string; message?: string; code?: string }> + listSlashCommands?: SyncEngine['listSlashCommands'] }) { const applySessionConfigCalls: Array<[string, Record]> = [] const applySessionConfig = async (sessionId: string, config: Record) => { @@ -77,7 +78,11 @@ function createApp(session: Session, opts?: { applySessionConfig, listCodexModelsForSession, listOpencodeModelsForSession, - resumeSession + resumeSession, + listSlashCommands: opts?.listSlashCommands ?? (async () => ({ + success: true, + commands: [] + })) } as Partial const app = new Hono() @@ -457,4 +462,66 @@ describe('sessions routes', () => { expect(response.status).toBe(200) expect(capturedResumeOpts).toEqual({ permissionMode: 'bypassPermissions' }) }) + + it('falls back to metadata slash commands when RPC listing fails', async () => { + const session = createSession({ + metadata: { + path: '/tmp/project', + host: 'localhost', + flavor: 'claude', + slashCommands: ['help', 'memory', 'status'] + } + }) + const { app } = createApp(session, { + listSlashCommands: async () => { + throw new Error('RPC unavailable') + } + }) + + const response = await app.request('/api/sessions/session-1/slash-commands') + + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ + success: true, + commands: [ + { name: 'help', source: 'builtin' }, + { name: 'memory', source: 'builtin' }, + { name: 'status', source: 'builtin' } + ] + }) + }) + + it('merges RPC and metadata slash commands without hiding built-ins', async () => { + const session = createSession({ + metadata: { + path: '/tmp/project', + host: 'localhost', + flavor: 'claude', + slashCommands: ['help', 'memory'] + } + }) + const { app } = createApp(session, { + listSlashCommands: async () => ({ + success: true, + commands: [ + { name: 'clear', source: 'builtin' }, + { name: 'project-only', source: 'project', content: 'Project prompt' } + ] + }) + }) + + const response = await app.request('/api/sessions/session-1/slash-commands') + + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ + success: true, + commands: [ + { name: 'help', source: 'builtin' }, + { name: 'memory', source: 'builtin' }, + { name: 'clear', source: 'builtin' }, + { name: 'project-only', source: 'project', content: 'Project prompt' } + ] + }) + }) + }) diff --git a/hub/src/web/routes/sessions.ts b/hub/src/web/routes/sessions.ts index 96148e38..d8f489a6 100644 --- a/hub/src/web/routes/sessions.ts +++ b/hub/src/web/routes/sessions.ts @@ -46,6 +46,39 @@ const uploadDeleteSchema = z.object({ const MAX_UPLOAD_BYTES = 50 * 1024 * 1024 + +type SlashCommand = { + name: string + description?: string + source: 'builtin' | 'user' | 'plugin' | 'project' + content?: string + pluginName?: string +} + +function commandsFromMetadataSlashCommands(names: readonly string[] | undefined): SlashCommand[] { + if (!names?.length) { + return [] + } + + return names + .filter((name): name is string => typeof name === 'string' && name.trim().length > 0) + .map((name) => ({ + name, + source: 'builtin' + })) +} + +function mergeSlashCommands( + primary: readonly SlashCommand[], + fallback: readonly SlashCommand[] +): SlashCommand[] { + const commandMap = new Map() + for (const command of [...fallback, ...primary]) { + commandMap.set(command.name, command) + } + return Array.from(commandMap.values()) +} + function estimateBase64Bytes(base64: string): number { const len = base64.length if (len === 0) return 0 @@ -498,10 +531,29 @@ export function createSessionsRoutes(getSyncEngine: () => SyncEngine | null): Ho // Get agent type from session metadata, default to 'claude' const agent = sessionResult.session.metadata?.flavor ?? 'claude' + const metadataCommands = commandsFromMetadataSlashCommands( + sessionResult.session.metadata?.slashCommands + ) + try { const result = await engine.listSlashCommands(sessionResult.sessionId, agent) + if (result.success && result.commands) { + return c.json({ + ...result, + commands: mergeSlashCommands(result.commands, metadataCommands) + }) + } + + if (metadataCommands.length > 0) { + return c.json({ success: true, commands: metadataCommands }) + } + return c.json(result) } catch (error) { + if (metadataCommands.length > 0) { + return c.json({ success: true, commands: metadataCommands }) + } + return c.json({ success: false, error: error instanceof Error ? error.message : 'Failed to list slash commands' diff --git a/web/src/lib/codexSlashCommands.test.ts b/web/src/lib/codexSlashCommands.test.ts index 15b3c3d4..ff2f7a49 100644 --- a/web/src/lib/codexSlashCommands.test.ts +++ b/web/src/lib/codexSlashCommands.test.ts @@ -33,6 +33,24 @@ describe('mergeSlashCommands', () => { { name: 'clear', source: 'project', content: 'project clear prompt' } ]) }) + + it('keeps API-provided built-ins while de-duplicating by name', () => { + const commands = mergeSlashCommands([ + { name: 'clear', source: 'builtin' }, + { name: 'status', source: 'builtin' }, + { name: 'help', source: 'builtin' }, + { name: 'status', source: 'builtin', description: 'Captured status' }, + { name: 'project-only', source: 'project', content: 'Project prompt' } + ]) + + expect(commands).toEqual([ + { name: 'clear', source: 'builtin' }, + { name: 'help', source: 'builtin' }, + { name: 'status', source: 'builtin', description: 'Captured status' }, + { name: 'project-only', source: 'project', content: 'Project prompt' } + ]) + }) + }) describe('findCodexCustomPromptExpansion', () => {