From 67785d2a08b015ffdba0f515da930adcb4888a41 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 20 Aug 2026 18:01:23 -0700 Subject: [PATCH] fix(native-chat): retire the failing dispatch's own waiter, not the queue head `dispatchClaudeTurn`'s error path shifted `dispatchWaiters[0]`, which is only this call's waiter when nothing else is in flight. With two sends queued, a broken pipe on the SECOND resolved the FIRST with null - reporting a delivered message as "delivery unconfirmed" while the send that actually failed stayed queued for its full ack budget, and the first send's replay then settled a waiter nobody awaited. Same mis-adoption class this PR is hardening: settle a dispatch only on evidence that belongs to it. `waitForReplay` now hands its waiter back so the error path can splice that one, mirroring what the timeout path already did. Also raises the tests' ack budget to 5s. vi.waitFor's first successful poll lands ~50ms after the waiter is armed, so a 100ms budget left almost nothing for the frame and could expire under parallel load. Found by the review loop on #15657. --- .../claude/claude-structured-dispatch.test.ts | 52 +++++++++++++++---- src/main/claude/claude-structured-dispatch.ts | 35 ++++++++----- 2 files changed, 63 insertions(+), 24 deletions(-) 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