diff --git a/cli/src/claude/claudeRemoteLauncher.launchFailure.test.ts b/cli/src/claude/claudeRemoteLauncher.launchFailure.test.ts new file mode 100644 index 00000000..2a9ad79f --- /dev/null +++ b/cli/src/claude/claudeRemoteLauncher.launchFailure.test.ts @@ -0,0 +1,403 @@ +import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest'; +import { RPC_METHODS } from '@hapi/protocol/rpcMethods'; +import { MessageQueue2 } from '@/utils/MessageQueue2'; +import { Session } from './session'; +import type { EnhancedMode } from './loop'; + +// claudeRemote() wraps the actual Claude Agent SDK subprocess spawn (the +// external-process boundary for this launcher) -- mock it the same way +// cursorLegacyRemoteLauncher.test.ts mocks `spawn`, rather than mocking any +// internal collaborator. +const claudeRemoteMock = vi.fn(); + +vi.mock('./claudeRemote', () => ({ + claudeRemote: (opts: unknown) => claudeRemoteMock(opts) +})); + +vi.mock('@/ui/logger', () => ({ + logger: { + debug: vi.fn(), + debugLargeJson: vi.fn(), + warn: vi.fn(), + error: vi.fn() + } +})); + +type RpcHandler = (params: unknown) => Promise | unknown; + +function makeClient() { + const handlers = new Map(); + let agentState: Record = {}; + return { + handlers, + rpcHandlerManager: { + registerHandler: vi.fn((method: string, handler: RpcHandler) => { + handlers.set(method, handler); + }) + }, + updateMetadata: vi.fn(), + updateAgentState: vi.fn((handler: (state: any) => any) => { + agentState = handler(agentState); + }), + sendSessionEvent: vi.fn(), + sendClaudeSessionMessage: vi.fn(), + sendAgentMessage: vi.fn(), + keepAlive: vi.fn(), + emitMessagesConsumed: vi.fn() + }; +} + +function makeSession(queue: MessageQueue2, client: ReturnType): Session { + return new Session({ + api: {} as never, + client: client as never, + path: '/tmp/project', + logPath: '/tmp/log', + sessionId: null, + mcpServers: {}, + messageQueue: queue, + onModeChange: vi.fn(), + mode: 'remote', + startedBy: 'runner', + startingMode: 'remote', + hookSettingsPath: '/tmp/hooks.json' + }); +} + +// Fires the RPC handler registered for `switch` (mirrors a UI-triggered +// switch-to-local request). ClaudeRemoteLauncher.requestExit() sets +// `exitReason` synchronously before awaiting the abort, so calling this +// without awaiting it lets a test deterministically terminate the launcher's +// respawn loop from inside a claudeRemote() mock implementation. +function triggerSwitch(client: ReturnType): void { + const handler = client.handlers.get(RPC_METHODS.Switch); + void handler?.(undefined); +} + +// True once the launcher has sent the "give up on this message, drop it" +// banner (i.e. the immediate-failure cap fired at least once). Used as an +// *outcome-based* stop condition below, instead of a hardcoded call number -- +// if a mutation changes when/whether the cap fires, the number of mock +// invocations before this becomes true changes too, so tests asserting an +// exact call count actually fail under that mutation rather than happening to +// reach the same hardcoded checkpoint regardless of production behavior. +function hasDropBanner(client: ReturnType): boolean { + return client.sendSessionEvent.mock.calls.some( + ([event]: any[]) => typeof event?.message === 'string' && event.message.includes('Dropping the queued message') + ); +} + +describe('claudeRemoteLauncher launch-failure recovery', () => { + beforeEach(() => { + claudeRemoteMock.mockReset(); + process.stdin.isTTY = false; + process.stdout.isTTY = false; + process.env.CLAUDE_REMOTE_RESPAWN_BACKOFF_MS = '0'; + }); + + afterEach(() => { + delete process.env.CLAUDE_REMOTE_RESPAWN_BACKOFF_MS; + }); + + it('restores the dequeued message into the queue when claudeRemote throws before onReady', async () => { + const queue = new MessageQueue2((mode) => JSON.stringify(mode)); + queue.push('hello', { permissionMode: 'default' }); + + const client = makeClient(); + const session = makeSession(queue, client); + + let callCount = 0; + let queueSizeOnSecondAttempt: number | undefined; + let restoredMessageText: string | undefined; + + claudeRemoteMock.mockImplementation(async (opts: any) => { + callCount += 1; + if (callCount === 1) { + const msg = await opts.nextMessage(); + expect(msg?.message).toBe('hello'); + throw new Error('spawn failed'); + } + + // Second attempt: inspect queue state directly rather than + // dequeuing again -- calling nextMessage() here would hang + // forever under the pre-fix behavior, where the message was + // silently dropped and nothing will ever push a new one. + queueSizeOnSecondAttempt = session.queue.size(); + restoredMessageText = session.queue.queue[0]?.message; + triggerSwitch(client); + throw new Error('spawn failed again'); + }); + + const { claudeRemoteLauncher } = await import('./claudeRemoteLauncher'); + await claudeRemoteLauncher(session); + + expect(callCount).toBe(2); + // The message that was already dequeued+acked on attempt 1 must be + // restored to the queue before attempt 2, not silently dropped. + expect(queueSizeOnSecondAttempt).toBe(1); + expect(restoredMessageText).toBe('hello'); + }); + + it('restores a joined batch as separate items with each original localId and order preserved', async () => { + // MessageQueue2.collectBatch() joins same-mode messages into a single + // `message` string for the SDK (`sameModeMessages.join('\n')`), but + // still tracks each original item's localId for the ack it fires. + // Restoring the joined string as one new queue item would silently + // drop both original localIds, orphaning the retried prompt from its + // hub row (and from cancel-by-localId). The restore must unshift each + // original item individually, in reverse order, so the queue ends up + // with the same two items in the same order with their own localIds + // intact. + const queue = new MessageQueue2((mode) => JSON.stringify(mode)); + queue.push('first', { permissionMode: 'default' }, 'local-id-1'); + queue.push('second', { permissionMode: 'default' }, 'local-id-2'); + + const client = makeClient(); + const session = makeSession(queue, client); + + let callCount = 0; + let queueItemsOnSecondAttempt: Array<{ message: string; localId: string | undefined }> | undefined; + + claudeRemoteMock.mockImplementation(async (opts: any) => { + callCount += 1; + if (callCount === 1) { + const msg = await opts.nextMessage(); + // Joined for the SDK, as before -- this part of the contract + // does not change. + expect(msg?.message).toBe('first\nsecond'); + throw new Error('spawn failed'); + } + + queueItemsOnSecondAttempt = session.queue.queue.map((item) => ({ + message: item.message, + localId: item.localId + })); + triggerSwitch(client); + throw new Error('spawn failed again'); + }); + + const { claudeRemoteLauncher } = await import('./claudeRemoteLauncher'); + await claudeRemoteLauncher(session); + + expect(callCount).toBe(2); + expect(queueItemsOnSecondAttempt).toEqual([ + { message: 'first', localId: 'local-id-1' }, + { message: 'second', localId: 'local-id-2' } + ]); + }); + + it('applies backoff between immediate failures and drops the message after repeated immediate failures instead of spinning forever', async () => { + const queue = new MessageQueue2((mode) => JSON.stringify(mode)); + const client = makeClient(); + const session = makeSession(queue, client); + + const BACKOFF_MS = 50; + process.env.CLAUDE_REMOTE_RESPAWN_BACKOFF_MS = String(BACKOFF_MS); + + let callCount = 0; + const SAFETY_CUTOFF = 20; + + claudeRemoteMock.mockImplementation(async () => { + callCount += 1; + if (hasDropBanner(client) || callCount > SAFETY_CUTOFF) { + // Outcome-based stop: the cap already fired on a previous + // attempt (proven by the drop banner), or the safety valve + // tripped because it never did. + triggerSwitch(client); + } + throw new Error('deterministic launch failure'); + }); + + const start = Date.now(); + const { claudeRemoteLauncher } = await import('./claudeRemoteLauncher'); + await claudeRemoteLauncher(session); + const elapsed = Date.now() - start; + + // 3 consecutive immediate failures hit the cap and drop the message + // on call 3 (2 backoff waits happen first, between calls 1->2 and + // 2->3; the cap-triggered drop on call 3 does not itself back off). + // Call 4 observes the drop banner and ends the test. This is an + // exact count, not a loose upper bound -- if the cap value or the + // reachedReadyThisAttempt gating regresses, this number changes. + expect(callCount).toBe(4); + expect(elapsed).toBeGreaterThanOrEqual(2 * BACKOFF_MS); + expect(elapsed).toBeLessThan(5_000); + + const messages = client.sendSessionEvent.mock.calls.map(([event]: any[]) => event); + const dropBanner = messages.find( + (event: any) => typeof event.message === 'string' && event.message.includes('Dropping the queued message') + ); + expect(dropBanner).toBeDefined(); + expect(dropBanner!.message).toContain('3 times in a row'); + }); + + it('resets the immediate-failure streak once onReady fires even if that same attempt later throws', async () => { + const queue = new MessageQueue2((mode) => JSON.stringify(mode)); + const client = makeClient(); + const session = makeSession(queue, client); + + let callCount = 0; + const SAFETY_CUTOFF = 20; + + claudeRemoteMock.mockImplementation(async (opts: any) => { + callCount += 1; + + if (hasDropBanner(client) || callCount > SAFETY_CUTOFF) { + triggerSwitch(client); + throw new Error('ending test'); + } + + if (callCount === 1) { + // Immediate failure, never reaches ready. + throw new Error('first attempt failure'); + } + if (callCount === 2) { + // Reaches onReady this attempt (a real turn happened), then + // still throws -- this must reset the immediate-failure + // streak, because the failure is no longer "immediate". + opts.onReady(); + throw new Error('second attempt failure post-ready'); + } + + // A brand new streak of deterministic immediate failures begins + // here. If call 2's reset did not happen, the streak would + // already be at 2 after call 2 and the cap would fire one call + // sooner. + throw new Error('post-reset streak failure'); + }); + + const { claudeRemoteLauncher } = await import('./claudeRemoteLauncher'); + await claudeRemoteLauncher(session); + + // call 1 = 1 immediate failure (streak -> 1) + // call 2 = reaches onReady, then throws -> streak resets to 0 (NOT + // counted as an immediate failure despite throwing) + // calls 3-5 = a fresh streak of 3 immediate failures -> cap fires on + // call 5, drop banner sent + // call 6 = observes the drop banner and ends the test + // This is an exact count: if the reset in call 2 did not happen, the + // cap would fire on call 4 instead (5 total calls, not 6). + expect(callCount).toBe(6); + }); + + it('resets the immediate-failure streak on a successful attempt that delivers a message but never reaches onReady', async () => { + // claudeRemote.ts's /clear handling delivers the queued message to + // the SDK (a real nextMessage() call, not a no-op park), then calls + // onSessionReset()/onCompletionEvent() and returns successfully -- + // WITHOUT ever calling onReady(). onReady alone can't tell that apart + // from a trivial no-op completion (the park-then-return-null case the + // livelock fix guards against), so the reset must also fire on + // "a message was actually delivered this attempt", not only on + // "onReady fired this attempt". + const queue = new MessageQueue2((mode) => JSON.stringify(mode)); + queue.push('/clear', { permissionMode: 'default' }); + + const client = makeClient(); + const session = makeSession(queue, client); + + let callCount = 0; + const SAFETY_CUTOFF = 20; + + claudeRemoteMock.mockImplementation(async (opts: any) => { + callCount += 1; + + if (hasDropBanner(client) || callCount > SAFETY_CUTOFF) { + triggerSwitch(client); + throw new Error('ending test'); + } + + if (callCount === 1) { + // Immediate failure, never reaches ready. + throw new Error('first attempt failure'); + } + if (callCount === 2) { + // Immediate failure again -- streak is now 2, one away from + // the cap. + throw new Error('second attempt failure'); + } + if (callCount === 3) { + // Mirrors claudeRemote.ts's /clear handling: the message is + // delivered (a real nextMessage() call), then the attempt + // returns successfully without calling onReady. + await opts.nextMessage(); + return; + } + + // A brand new streak of deterministic immediate failures begins + // here. If call 3's successful /clear did not reset the streak, + // it would still be at 2 and this single failure would already + // hit the cap. + throw new Error('post-clear streak failure'); + }); + + const { claudeRemoteLauncher } = await import('./claudeRemoteLauncher'); + await claudeRemoteLauncher(session); + + // call 1, 2 = 2 immediate failures (streak -> 2) + // call 3 = delivers /clear, completes successfully without onReady -> + // streak resets to 0 + // calls 4-6 = a fresh streak of 3 immediate failures -> cap fires on + // call 6, drop banner sent + // call 7 = observes the drop banner and ends the test + // This is an exact count: if call 3 did not reset the streak, the cap + // would fire on call 4 instead (5 total calls, not 7), and the drop + // banner would falsely claim "3 times in a row" after just 1 failure + // past the successful /clear. + expect(callCount).toBe(7); + }); + + it('does not livelock when an isolated message repeatedly hits a deterministic launch failure', async () => { + // Reproduces the hostile-review probe: an isolated message (e.g. + // /compact, /clear) that keeps hitting a deterministic launch + // failure oscillates between two nextMessage() shapes every other + // attempt -- parked into `pending` (returns null, attempt ends + // without throwing) vs. handed to the SDK from `pending` (attempt + // throws). If the immediate-failure streak were reset on every + // non-throwing attempt (rather than gated on reachedReadyThisAttempt), + // this oscillation would reset the streak every other attempt and the + // cap would never fire. + const queue = new MessageQueue2((mode) => JSON.stringify(mode)); + queue.pushIsolateAndClear('/compact', { permissionMode: 'default' }); + + const client = makeClient(); + const session = makeSession(queue, client); + + let callCount = 0; + const SAFETY_CUTOFF = 20; + + claudeRemoteMock.mockImplementation(async (opts: any) => { + callCount += 1; + + if (hasDropBanner(client) || callCount > SAFETY_CUTOFF) { + // Outcome-based stop: cap already fired (or the safety valve + // tripped because it never did) -- stop here instead of + // hanging on session.queue.waitForMessagesAndGetAsString() + // blocking forever on the now-empty, non-closed queue. + triggerSwitch(client); + throw new Error('ending test'); + } + + // Mirror claudeRemote.ts's real "get initial message" step + // (`initial = await opts.nextMessage(); if (!initial) return;`) + // instead of a scripted per-call script, so the actual + // pending/isolate parking logic in claudeRemoteLauncher.ts's + // nextMessage() drives the oscillation, exactly as it does live. + const initial = await opts.nextMessage(); + if (!initial) { + return; + } + throw new Error('deterministic launch failure'); + }); + + const { claudeRemoteLauncher } = await import('./claudeRemoteLauncher'); + await claudeRemoteLauncher(session); + + // Must terminate via the cap+drop path well under the safety cutoff + // -- if the reset gating regresses, this livelocks the + // immediate-failure counter at 1<->0 forever and hits the cutoff + // instead (the probe that found this bug observed 25 attempts with + // the cap never firing). + expect(callCount).toBeLessThan(SAFETY_CUTOFF); + expect(hasDropBanner(client)).toBe(true); + }); +}); diff --git a/cli/src/claude/claudeRemoteLauncher.ts b/cli/src/claude/claudeRemoteLauncher.ts index b3f39ef1..7703c2d9 100644 --- a/cli/src/claude/claudeRemoteLauncher.ts +++ b/cli/src/claude/claudeRemoteLauncher.ts @@ -26,6 +26,25 @@ interface PermissionsField { allowedTools?: string[]; } +// If claudeRemote() throws before it ever reaches onReady (spawn failure, +// invalid model/args, auth failure, resume-anchor rejection, ...), the failure +// is almost certainly deterministic: respawning immediately just repeats the +// same failure. MAX_IMMEDIATE_RESPAWN_FAILURES caps consecutive such failures +// before this loop drops the message that keeps triggering them instead of +// respawning forever, and getRespawnBackoffMs() paces the retries in between. +// The streak resets whenever an attempt reaches onReady, or once the message +// is dropped, so a later unrelated failure gets its own fresh budget. Only +// the one message is given up on -- the session/process itself is not ended. +const MAX_IMMEDIATE_RESPAWN_FAILURES = 3; + +function getRespawnBackoffMs(): number { + const raw = process.env.CLAUDE_REMOTE_RESPAWN_BACKOFF_MS; + if (raw === undefined) return 1000; + const parsed = Number.parseInt(raw, 10); + if (Number.isNaN(parsed) || parsed < 0) return 1000; + return parsed; +} + class ClaudeRemoteLauncher extends RemoteLauncherBase { private readonly session: Session; private abortController: AbortController | null = null; @@ -49,6 +68,28 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase { await this.abortFuture?.promise; } + // Waits for `ms`, but resolves early if `signal` aborts -- mirrors + // cursorLegacyRemoteLauncher.transientBackoff (single completion path so + // the abort listener is always removed, whether the timer or the abort + // wins) so a user-initiated switch/exit during the respawn backoff isn't + // stuck waiting out the full delay. + private async respawnBackoff(ms: number, signal: AbortSignal): Promise { + if (ms <= 0 || signal.aborted) return; + await new Promise((resolve) => { + let timer: ReturnType | null = null; + const finish = () => { + if (timer !== null) { + clearTimeout(timer); + timer = null; + } + signal.removeEventListener('abort', finish); + resolve(); + }; + timer = setTimeout(finish, ms); + signal.addEventListener('abort', finish, { once: true }); + }); + } + private async handleAbortRequest(): Promise { logger.debug('[remote]: doAbort'); await this.abort(); @@ -272,9 +313,12 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase { let pending: { message: string; mode: EnhancedMode; + isolate: boolean; + items: Array<{ message: string; localId?: string }>; } | null = null; let previousSessionId: string | null = null; + let immediateFailureCount = 0; while (!this.exitReason) { logger.debug('[remote]: launch'); messageBuffer.addMessage('═'.repeat(40), 'status'); @@ -296,6 +340,44 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase { this.abortFuture = new Future(); let modeHash: string | null = null; let mode: EnhancedMode | null = null; + // True once onReady() has fired at least once during this + // attempt. Used to distinguish an immediate/deterministic + // failure (never reached ready) from a failure after real + // progress was made, for the respawn-storm guard below. + let reachedReadyThisAttempt = false; + // True once nextMessage() has actually handed a message to the + // SDK during this attempt (as opposed to parking it into + // `pending` and returning null). Some commands complete and + // return successfully without ever calling onReady -- e.g. + // claudeRemote.ts's /clear handling calls onSessionReset() and + // returns directly -- so onReady alone cannot tell "a real + // message was processed" apart from "nothing was delivered + // this attempt" (e.g. the livelock-prone park-then-return-null + // case). See the success-path reset below. + let deliveredMessageThisAttempt = false; + // Tracks the most recent message batch handed to the SDK via + // nextMessage() that has not yet been confirmed complete by a + // following onReady(). If claudeRemote() throws while a + // message is in flight, it is dequeued+acked already (see + // MessageQueue2.collectBatch) but never delivered -- the catch + // block below restores it with queue.unshift()/unshiftIsolated() + // (per original item, preserving each item's localId) so it is + // not silently dropped. + type InFlightMessage = { + items: Array<{ message: string; localId?: string }>; + mode: EnhancedMode; + isolate: boolean; + }; + // The `as InFlightMessage | null` (rather than plain `= null`) + // is required, not decorative: the only assignments of a + // non-null value happen inside the nextMessage()/onReady() + // closures below, which TS's control-flow narrowing does not + // see from this function body. Without the cast, TS narrows + // this declaration to the `null` literal type, and the + // `if (inFlightMessage)` checks further down then fail + // typecheck with "Property 'isolate' does not exist on type + // 'never'" (verified against `bun run typecheck`). + let inFlightMessage: InFlightMessage | null = null as InFlightMessage | null; try { await claudeRemote({ sessionId: session.sessionId, @@ -326,6 +408,8 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase { // mid-session model switch), so a construction-time snapshot // would go stale. See SDKToLogConverter.updateSelectedModel. sdkToLogConverter.updateSelectedModel(p.mode.model ?? null); + inFlightMessage = { items: p.items, mode: p.mode, isolate: p.isolate }; + deliveredMessageThisAttempt = true; return p; } @@ -333,6 +417,19 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase { if (msg) { if ((modeHash && msg.hash !== modeHash) || msg.isolate) { + // Parked into `pending`, not handed to the + // SDK -- deliberately NOT tracked as + // inFlightMessage. `pending` is declared + // outside the while loop and is only ever + // cleared when actually consumed (the + // `if (pending)` branch above), so it + // already survives a throw in this or any + // later attempt without help from the + // restore-on-catch logic below. Tracking + // it here too would restore a second copy + // via queue.unshift() on top of the one + // still safely held in `pending`, + // delivering it twice. logger.debug('[remote]: mode has changed, pending message'); pending = msg; return null; @@ -341,6 +438,8 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase { mode = msg.mode; permissionHandler.handleModeChange(mode.permissionMode); sdkToLogConverter.updateSelectedModel(mode.model ?? null); + inFlightMessage = { items: msg.items, mode: msg.mode, isolate: msg.isolate }; + deliveredMessageThisAttempt = true; return { message: msg.message, mode: msg.mode @@ -382,6 +481,13 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase { session.consumeOneTimeFlags(); }, onReady: () => { + // Reaching ready at all means this attempt is not an + // immediate/deterministic failure -- reset the + // respawn-storm guard. The turn that led here is no + // longer "in flight" either. + reachedReadyThisAttempt = true; + inFlightMessage = null; + logger.debug( `[claudeRemoteLauncher][async-debug] onReady callback ` + `(hasPending=${Boolean(pending)}, queueSize=${session.queue.size()})` @@ -399,12 +505,98 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase { if (!this.exitReason && controller.signal.aborted) { session.client.sendSessionEvent({ type: 'message', message: 'Aborted by user' }); } + + // A full attempt completed without throwing. Clear the + // immediate-failure streak if this attempt either reached + // onReady, or actually delivered a message to the SDK + // (deliveredMessageThisAttempt) -- some commands complete + // successfully without ever calling onReady (e.g. + // claudeRemote.ts's /clear handling calls onSessionReset() + // and returns directly), and that is still real progress, + // not an immediate/deterministic failure. Without the + // deliveredMessageThisAttempt half of this condition, a + // successful /clear between two unrelated launch failures + // would not reset the streak and the cap could fire after + // just one more failure instead of a fresh budget of 3. + // + // Neither flag is set for a trivial no-op completion (e.g. + // the initial nextMessage() returning null because the + // only queued message was parked into `pending` for a + // later attempt, per claudeRemote.ts's "initial + // nextMessage returned null; exiting") -- resetting on + // that case would clear the streak every other attempt + // while the underlying deterministic failure keeps + // recurring on alternating attempts, and the cap would + // never fire (livelock). + if (reachedReadyThisAttempt || deliveredMessageThisAttempt) { + immediateFailureCount = 0; + } } catch (e) { logger.debug('[remote]: launch error', e); + + // Restores a message batch that was already + // dequeued+acked from the queue (see + // MessageQueue2.collectBatch: the ack fires at dequeue + // time, before the SDK ever sees the message) but never + // got a chance to be processed, so it is retried instead + // of lost. Each original item is unshifted individually, + // in reverse order, so per-item localId is preserved + // (reconnecting the retried prompt to its hub row instead + // of a localId-less orphan) and the original relative + // order is restored -- collectBatch() joins same-mode + // items into a single `message` string for the SDK, but + // still returns the pre-join `items` breakdown for this. + const restoreInFlightMessage = () => { + if (!inFlightMessage) return; + const { items, mode, isolate } = inFlightMessage; + for (const item of [...items].reverse()) { + if (isolate) { + session.queue.unshiftIsolated(item.message, mode, item.localId); + } else { + session.queue.unshift(item.message, mode, item.localId); + } + } + inFlightMessage = null; + }; + if (!this.exitReason) { const detail = e instanceof Error ? e.message : String(e); - session.client.sendSessionEvent({ type: 'message', message: `Process exited unexpectedly: ${detail}` }); - continue; + + if (reachedReadyThisAttempt) { + immediateFailureCount = 0; + } else { + immediateFailureCount += 1; + } + + if (immediateFailureCount >= MAX_IMMEDIATE_RESPAWN_FAILURES) { + // Give up on retrying *this message*, not the + // whole session. Restoring it here would feed it + // straight back into another immediate failure on + // the very next attempt (unshift -> re-dequeue -> + // re-throw), storming again -- matches + // cursorLegacyRemoteLauncher's drop-and-reset + // policy on its own consecutive-failure cap. + // Reset the streak and keep the loop (and this OS + // process) alive so an unrelated later message + // gets its own fresh budget. + inFlightMessage = null; + session.client.sendSessionEvent({ + type: 'message', + message: `Process exited unexpectedly ${MAX_IMMEDIATE_RESPAWN_FAILURES} times in a row: ${detail}. Dropping the queued message; resolve the issue and resend it.` + }); + immediateFailureCount = 0; + } else { + restoreInFlightMessage(); + session.client.sendSessionEvent({ type: 'message', message: `Process exited unexpectedly: ${detail}` }); + await this.respawnBackoff(getRespawnBackoffMs(), controller.signal); + continue; + } + } else { + // exitReason already set by something else (e.g. a + // user-initiated switch/exit racing this throw) -- + // still restore any in-flight message so it isn't + // silently dropped by that unrelated shutdown. + restoreInFlightMessage(); } } finally { logger.debug('[remote]: launch finally'); diff --git a/cli/src/utils/MessageQueue2.ts b/cli/src/utils/MessageQueue2.ts index 54d0e5a3..a596b0d7 100644 --- a/cli/src/utils/MessageQueue2.ts +++ b/cli/src/utils/MessageQueue2.ts @@ -314,7 +314,7 @@ export class MessageQueue2 { * Wait for messages and return all messages with the same mode as a single string * Returns { message: string, mode: T } or null if aborted/closed */ - async waitForMessagesAndGetAsString(abortSignal?: AbortSignal): Promise<{ message: string, mode: T, isolate: boolean, hash: string } | null> { + async waitForMessagesAndGetAsString(abortSignal?: AbortSignal): Promise<{ message: string, mode: T, isolate: boolean, hash: string, items: Array<{ message: string, localId?: string }> } | null> { // If we have messages, return them immediately if (this.queue.length > 0) { return this.collectBatch(); @@ -338,7 +338,7 @@ export class MessageQueue2 { /** * Collect a batch of messages with the same mode, respecting isolation requirements */ - private collectBatch(): { message: string, mode: T, hash: string, isolate: boolean } | null { + private collectBatch(): { message: string, mode: T, hash: string, isolate: boolean, items: Array<{ message: string, localId?: string }> } | null { if (this.queue.length === 0) { return null; } @@ -346,6 +346,11 @@ export class MessageQueue2 { const firstItem = this.queue[0]; const sameModeMessages: string[] = []; const consumedLocalIds: string[] = []; + // Per-item breakdown of this batch, preserved alongside the joined + // `message` string below so callers that need to requeue individual + // messages (e.g. restoring a failed batch with each item's own + // localId intact) don't have to re-split an already-joined string. + const items: Array<{ message: string, localId?: string }> = []; let mode = firstItem.mode; let isolate = firstItem.isolate ?? false; const targetModeHash = firstItem.modeHash; @@ -354,6 +359,7 @@ export class MessageQueue2 { if (firstItem.isolate) { const item = this.queue.shift()!; sameModeMessages.push(item.message); + items.push({ message: item.message, localId: item.localId }); if (item.localId) consumedLocalIds.push(item.localId); logger.debug(`[MessageQueue2] Collected isolated message with mode hash: ${targetModeHash}`); } else { @@ -363,6 +369,7 @@ export class MessageQueue2 { !this.queue[0].isolate) { const item = this.queue.shift()!; sameModeMessages.push(item.message); + items.push({ message: item.message, localId: item.localId }); if (item.localId) consumedLocalIds.push(item.localId); } logger.debug(`[MessageQueue2] Collected batch of ${sameModeMessages.length} messages with mode hash: ${targetModeHash}`); @@ -379,7 +386,8 @@ export class MessageQueue2 { message: combinedMessage, mode, hash: targetModeHash, - isolate + isolate, + items }; }