diff --git a/src/main/claude/claude-structured-dispatch.test.ts b/src/main/claude/claude-structured-dispatch.test.ts index f863dfb07a4..cf47b43ce44 100644 --- a/src/main/claude/claude-structured-dispatch.test.ts +++ b/src/main/claude/claude-structured-dispatch.test.ts @@ -6,6 +6,10 @@ import type { AgentJournalMessageItem } from '../../shared/agent-session-journal import { dispatchClaudeTurn, resolveClaudeReplayWaiter } from './claude-structured-dispatch' import type { ClaudeSession } from './claude-structured-session-state' +// vi.waitFor's first successful poll lands ~50ms after the waiter is armed, so a +// 100ms ack budget leaves almost none for the frame under parallel load. +const ACK_BUDGET_MS = 5_000 + function sessionFor(send = vi.fn().mockResolvedValue(undefined)): ClaudeSession { return { connection: { send } as unknown as ClaudeSession['connection'], @@ -43,7 +47,7 @@ describe('Claude structured dispatch image limits', () => { const dispatched = dispatchClaudeTurn( session, { clientMessageId: 'client-1', body: userMessage([{ type: 'text', text: '/permissions' }]) }, - 100 + ACK_BUDGET_MS ) await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1)) @@ -69,7 +73,7 @@ describe('Claude structured dispatch image limits', () => { const dispatched = dispatchClaudeTurn( session, { clientMessageId: 'client-1', body: userMessage([{ type: 'text', text: 'hello' }]) }, - 100 + ACK_BUDGET_MS ) await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1)) @@ -96,7 +100,7 @@ describe('Claude structured dispatch image limits', () => { const dispatched = dispatchClaudeTurn( session, { clientMessageId: 'client-1', body: userMessage([{ type: 'text', text: 'queued' }]) }, - 100 + ACK_BUDGET_MS ) await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1)) @@ -136,7 +140,7 @@ describe('Claude structured dispatch image limits', () => { const dispatched = dispatchClaudeTurn( session, { clientMessageId: 'client-1', body: userMessage([{ type: 'text', text: 'queued' }]) }, - 100 + ACK_BUDGET_MS ) await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1)) @@ -165,7 +169,7 @@ describe('Claude structured dispatch image limits', () => { const first = dispatchClaudeTurn( session, { clientMessageId: 'client-1', body: userMessage([{ type: 'text', text: 'one' }]) }, - 100 + ACK_BUDGET_MS ) await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1)) resolveClaudeReplayWaiter(session, userReplayFrame('replay-one', 'one')) @@ -174,7 +178,7 @@ describe('Claude structured dispatch image limits', () => { const second = dispatchClaudeTurn( session, { clientMessageId: 'client-2', body: userMessage([{ type: 'text', text: 'two' }]) }, - 100 + ACK_BUDGET_MS ) await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1)) @@ -202,7 +206,7 @@ describe('Claude structured dispatch image limits', () => { const dispatched = dispatchClaudeTurn( session, { clientMessageId: 'client-1', body: userMessage([{ type: 'text', text: 'hello' }]) }, - 100 + ACK_BUDGET_MS ) await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1)) @@ -230,12 +234,10 @@ describe('Claude structured dispatch image limits', () => { const path = join(directory, 'shot.png') await writeFile(path, Buffer.alloc(64)) const session = sessionFor() - // Real file I/O runs before the ack timer starts, so give this one room: - // a 100ms budget can expire under parallel load before the frame arrives. const dispatched = dispatchClaudeTurn( session, { clientMessageId: 'client-1', body: userMessage([{ type: 'image-ref', path }]) }, - 5_000 + ACK_BUDGET_MS ) await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1)) @@ -263,6 +265,36 @@ describe('Claude structured dispatch image limits', () => { } }) + // The waiter queue is positional, so shifting the head on a send failure + // resolves whichever dispatch happens to be first - reporting a delivered + // message as unconfirmed while the send that actually failed keeps waiting. + it('retires its own waiter when the send fails, not the one at the head', async () => { + const session = sessionFor() + const first = dispatchClaudeTurn( + session, + { clientMessageId: 'client-1', body: userMessage([{ type: 'text', text: 'one' }]) }, + ACK_BUDGET_MS + ) + await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1)) + const firstWaiter = session.dispatchWaiters[0] + + session.connection.send = vi.fn().mockRejectedValue(new Error('broken pipe')) + await expect( + dispatchClaudeTurn( + session, + { clientMessageId: 'client-2', body: userMessage([{ type: 'text', text: 'two' }]) }, + ACK_BUDGET_MS + ) + ).resolves.toEqual({ state: 'unknown', reason: 'broken pipe' }) + + expect(session.dispatchWaiters).toEqual([firstWaiter]) + resolveClaudeReplayWaiter(session, userReplayFrame('replay-one', 'one')) + await expect(first).resolves.toMatchObject({ + state: 'accepted', + providerIdentity: { uuid: 'replay-one' } + }) + }) + it('rejects more than twenty URL images before sending', async () => { const session = sessionFor() const body = userMessage( diff --git a/src/main/claude/claude-structured-dispatch.ts b/src/main/claude/claude-structured-dispatch.ts index 7ea922106c0..daab1be3279 100644 --- a/src/main/claude/claude-structured-dispatch.ts +++ b/src/main/claude/claude-structured-dispatch.ts @@ -3,7 +3,7 @@ import { open } from 'node:fs/promises' import type { AgentJournalMessageItem } from '../../shared/agent-session-journal-types' import type { NativeChatBlock } from '../../shared/native-chat-types' import type { AgentSessionDispatchOutcome } from '../native-chat/agent-session-wire/structured-agent-session-adapter' -import type { ClaudeSession } from './claude-structured-session-state' +import type { ClaudeDispatchWaiter, ClaudeSession } from './claude-structured-session-state' import { readClaudeFrameString } from './claude-structured-init-proof' import { isKnownHarnessInjectedUserTurnText } from '../../shared/harness-injected-user-turns' import { claudeRecord } from './claude-structured-item-translation' @@ -189,26 +189,35 @@ async function messageContent(body: AgentJournalMessageItem): Promise return content } +function dropWaiter(session: ClaudeSession, waiter: ClaudeDispatchWaiter): void { + const index = session.dispatchWaiters.indexOf(waiter) + if (index !== -1) { + session.dispatchWaiters.splice(index, 1) + } +} + +/** Arms a waiter and hands it back, so a failure can retire ITS OWN waiter — the + * queue is positional, and shifting the head resolves whichever send happens to + * be first instead. */ function waitForReplay( session: ClaudeSession, timeoutMs: number, acceptsResult: boolean -): Promise { - return new Promise((resolve) => { - const waiter = { +): { waiter: ClaudeDispatchWaiter; replayed: Promise } { + let waiter!: ClaudeDispatchWaiter + const replayed = new Promise((resolve) => { + waiter = { acceptsResult, resolve, timer: setTimeout(() => { - const index = session.dispatchWaiters.indexOf(waiter) - if (index !== -1) { - session.dispatchWaiters.splice(index, 1) - } + dropWaiter(session, waiter) resolve(null) }, timeoutMs) } waiter.timer.unref?.() session.dispatchWaiters.push(waiter) }) + return { waiter, replayed } } export async function dispatchClaudeTurn( @@ -225,7 +234,7 @@ export async function dispatchClaudeTurn( const acceptsResult = input.body.blocks.some( (block) => block.type === 'text' && block.text.trimStart().startsWith('/') ) - const replayed = waitForReplay(session, timeoutMs, acceptsResult) + const { waiter, replayed } = waitForReplay(session, timeoutMs, acceptsResult) try { await session.connection.send({ type: 'user', @@ -234,11 +243,9 @@ export async function dispatchClaudeTurn( session_id: session.providerSessionId }) } catch (error) { - const waiter = session.dispatchWaiters.shift() - if (waiter) { - clearTimeout(waiter.timer) - waiter.resolve(null) - } + dropWaiter(session, waiter) + clearTimeout(waiter.timer) + waiter.resolve(null) return { state: 'unknown', reason: (error as Error).message } } const uuid = await replayed