diff --git a/src/main/claude/claude-structured-dispatch.test.ts b/src/main/claude/claude-structured-dispatch.test.ts index 67f7e1d97aa..31924f44b41 100644 --- a/src/main/claude/claude-structured-dispatch.test.ts +++ b/src/main/claude/claude-structured-dispatch.test.ts @@ -293,6 +293,55 @@ describe('Claude structured dispatch image limits', () => { }) }) + // The CLI round-trips a client-supplied uuid on the replay (measured), so a + // dispatch can be matched to its OWN echo instead of to whatever is at the + // head of the queue. + it('stamps a uuid on the outgoing frame', async () => { + const session = sessionFor() + void dispatchClaudeTurn( + session, + { clientMessageId: 'client-1', body: userMessage([{ type: 'text', text: 'hello' }]) }, + ACK_BUDGET_MS + ) + await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1)) + + expect(session.connection.send).toHaveBeenCalledWith( + expect.objectContaining({ uuid: session.dispatchWaiters[0]!.sentUuid }) + ) + }) + + it('settles the dispatch whose uuid was echoed, not the queue 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 second = dispatchClaudeTurn( + session, + { clientMessageId: 'client-2', body: userMessage([{ type: 'text', text: 'two' }]) }, + ACK_BUDGET_MS + ) + await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(2)) + const firstUuid = session.dispatchWaiters[0]!.sentUuid + const secondUuid = session.dispatchWaiters[1]!.sentUuid + + // Out of order: the second send's echo arrives while the first is still head. + resolveClaudeReplayWaiter(session, userReplayFrame(secondUuid, 'two')) + await expect(second).resolves.toMatchObject({ + state: 'accepted', + providerIdentity: { uuid: secondUuid } + }) + expect(session.dispatchWaiters.map((waiter) => waiter.sentUuid)).toEqual([firstUuid]) + + resolveClaudeReplayWaiter(session, userReplayFrame(firstUuid, 'one')) + await expect(first).resolves.toMatchObject({ + state: 'accepted', + providerIdentity: { uuid: firstUuid } + }) + }) + // 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. diff --git a/src/main/claude/claude-structured-dispatch.ts b/src/main/claude/claude-structured-dispatch.ts index 2e748116d54..7788f2ce708 100644 --- a/src/main/claude/claude-structured-dispatch.ts +++ b/src/main/claude/claude-structured-dispatch.ts @@ -1,3 +1,4 @@ +import { randomUUID } from 'node:crypto' import { extname } from 'node:path' import { open } from 'node:fs/promises' import type { AgentJournalMessageItem } from '../../shared/agent-session-journal-types' @@ -101,28 +102,44 @@ function isClaudeUserMessageReplay(message: Record): boolean { return !isKnownHarnessInjectedUserTurnText(claudeUserFrameText(message)) } +function settleWaiter(session: ClaudeSession, waiter: ClaudeDispatchWaiter, uuid: string): void { + dropWaiter(session, waiter) + clearTimeout(waiter.timer) + waiter.settledUuid = uuid + waiter.resolve(uuid) +} + export function resolveClaudeReplayWaiter( session: ClaudeSession, message: Record ): void { - const isUserReplay = isClaudeUserMessageReplay(message) - const isCompletedCommand = message.type === 'result' - if ( - (!isUserReplay && !isCompletedCommand) || - readClaudeFrameString(message, 'session_id') !== session.providerSessionId - ) { + if (readClaudeFrameString(message, 'session_id') !== session.providerSessionId) { return } const uuid = readClaudeFrameString(message, 'uuid') - const current = session.dispatchWaiters[0] - if (isCompletedCommand && !current?.acceptsResult) { + if (!uuid) { return } - const waiter = uuid ? session.dispatchWaiters.shift() : undefined - if (waiter && uuid) { - clearTimeout(waiter.timer) - waiter.resolve(uuid) + // Exact correlation: the CLI echoes the uuid we stamped on the outgoing frame, + // so this replay belongs to that dispatch whatever its queue position - which + // is what keeps a late replay from settling somebody else's send. + const owner = session.dispatchWaiters.find((candidate) => candidate.sentUuid === uuid) + if (owner) { + settleWaiter(session, owner, uuid) + return } + // Compat path for a CLI that mints its own uuid instead of echoing ours. The + // frame then carries no correlation at all, so the head is the only candidate + // and the shape gates have to carry the weight. + const isCompletedCommand = message.type === 'result' + if (!isClaudeUserMessageReplay(message) && !isCompletedCommand) { + return + } + const head = session.dispatchWaiters[0] + if (!head || (isCompletedCommand && !head.acceptsResult)) { + return + } + settleWaiter(session, head, uuid) } async function imageContent( @@ -190,12 +207,14 @@ function dropWaiter(session: ClaudeSession, waiter: ClaudeDispatchWaiter): void function waitForReplay( session: ClaudeSession, timeoutMs: number, - acceptsResult: boolean + acceptsResult: boolean, + sentUuid: string ): { waiter: ClaudeDispatchWaiter; replayed: Promise } { let waiter!: ClaudeDispatchWaiter const replayed = new Promise((resolve) => { waiter = { acceptsResult, + sentUuid, resolve, timer: setTimeout(() => { dropWaiter(session, waiter) @@ -253,10 +272,15 @@ export async function dispatchClaudeTurn( const acceptsResult = input.body.blocks.some( (block) => block.type === 'text' && isSlashCommandText(block.text) ) - const { waiter, replayed } = waitForReplay(session, timeoutMs, acceptsResult) + // Stamped so the echo can be matched to THIS dispatch: the CLI round-trips a + // client-supplied uuid verbatim (measured), and it also dedupes on it, so a + // resend of the same frame cannot land twice. + const sentUuid = randomUUID() + const { waiter, replayed } = waitForReplay(session, timeoutMs, acceptsResult, sentUuid) try { await session.connection.send({ type: 'user', + uuid: sentUuid, message: { role: 'user', content }, parent_tool_use_id: null, session_id: session.providerSessionId @@ -265,7 +289,7 @@ export async function dispatchClaudeTurn( // A write can reach Claude and still reject on the flush that follows, so the // replay may already have claimed this waiter. Reporting `unknown` then // strands an identity nothing else can adopt, and the echo duplicates. - if (!session.dispatchWaiters.includes(waiter)) { + if (waiter.settledUuid) { const settled = await replayed if (settled) { return { diff --git a/src/main/claude/claude-structured-session-state.ts b/src/main/claude/claude-structured-session-state.ts index 9a8462f6ac2..eb1686d7c4f 100644 --- a/src/main/claude/claude-structured-session-state.ts +++ b/src/main/claude/claude-structured-session-state.ts @@ -56,6 +56,13 @@ export type ClaudeDispatchWaiter = { resolve: (uuid: string | null) => void timer: ReturnType acceptsResult: boolean + /** The uuid stamped on this dispatch's outgoing frame. The CLI echoes a + * client-supplied uuid verbatim, so a replay carrying it belongs to THIS + * send - no queue-position guessing. */ + sentUuid: string + /** Set when this waiter was actually settled, so a caller can tell "my replay + * arrived" from "something removed me from the queue". */ + settledUuid?: string } export type ClaudeSession = {