fix(native-chat): drop the replay of a dispatch that already gave up

The uuid index only covered LIVE waiters, so once a dispatch left the queue -
timed out, or its send failed - its uuid was forgotten. Its replay is still in
flight, and it then matched no owner, passed the shape gate as the genuine
stamped replay it is, and settled whoever was now at the head: that send
resolved `accepted` carrying a dead send's uuid, so the dead send's text upserted
over a live bubble and the live send's own echo appended a second row.

The previous commit's description claimed this case was closed. It was not - the
correlation only helped while both parties were still queued.

Remember the uuid of a dispatch that leaves without its echo (bounded to the
last 64, and it only grows when a dispatch gives up) and drop a replay carrying
one before the compat path can hand it to the head.

Found by the review loop on #15657.
This commit is contained in:
Brennan Benson
2026-08-23 12:08:09 -07:00
parent c53dba195c
commit 1ba8554af4
3 changed files with 59 additions and 2 deletions
@@ -342,6 +342,40 @@ describe('Claude structured dispatch image limits', () => {
})
})
// A dispatch that gave up still has a replay in flight. Once its waiter is
// gone, that frame matches no owner and would otherwise fall to the compat
// path and settle whoever is now at the head.
it('drops the replay of a dispatch that already timed out', async () => {
const session = sessionFor()
const first = dispatchClaudeTurn(
session,
{ clientMessageId: 'client-1', body: userMessage([{ type: 'text', text: 'one' }]) },
60
)
await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1))
const firstUuid = session.dispatchWaiters[0]!.sentUuid
await expect(first).resolves.toMatchObject({ state: 'unknown' })
expect(session.dispatchWaiters).toHaveLength(0)
const second = dispatchClaudeTurn(
session,
{ clientMessageId: 'client-2', body: userMessage([{ type: 'text', text: 'two' }]) },
ACK_BUDGET_MS
)
await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1))
const secondUuid = session.dispatchWaiters[0]!.sentUuid
// The dead send's echo lands while the second is waiting.
resolveClaudeReplayWaiter(session, userReplayFrame(firstUuid, 'one'))
expect(session.dispatchWaiters).toHaveLength(1)
resolveClaudeReplayWaiter(session, userReplayFrame(secondUuid, 'two'))
await expect(second).resolves.toMatchObject({
state: 'accepted',
providerIdentity: { uuid: secondUuid }
})
})
// 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.
+21 -2
View File
@@ -128,6 +128,12 @@ export function resolveClaudeReplayWaiter(
settleWaiter(session, owner, uuid)
return
}
// A replay for a dispatch that already gave up belongs to nobody. Dropping it
// here is what keeps the compat path below from handing it to the head, which
// would upsert the dead send's text over a live one's bubble.
if (session.retiredSentUuids?.includes(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.
@@ -194,6 +200,10 @@ async function messageContent(body: AgentJournalMessageItem): Promise<unknown[]>
return content
}
/** Enough to cover the replays still in flight for sends that gave up; the list
* only grows when a dispatch leaves without its echo. */
const RETIRED_UUID_MEMORY = 64
function dropWaiter(session: ClaudeSession, waiter: ClaudeDispatchWaiter): void {
const index = session.dispatchWaiters.indexOf(waiter)
if (index !== -1) {
@@ -201,6 +211,15 @@ function dropWaiter(session: ClaudeSession, waiter: ClaudeDispatchWaiter): void
}
}
/** Remember a dispatch that left without its echo, so the echo cannot later be
* mistaken for somebody else's on the compat path. */
function retireWaiter(session: ClaudeSession, waiter: ClaudeDispatchWaiter): void {
dropWaiter(session, waiter)
const retired = session.retiredSentUuids ?? []
retired.push(waiter.sentUuid)
session.retiredSentUuids = retired.slice(-RETIRED_UUID_MEMORY)
}
/** 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. */
@@ -217,7 +236,7 @@ function waitForReplay(
sentUuid,
resolve,
timer: setTimeout(() => {
dropWaiter(session, waiter)
retireWaiter(session, waiter)
resolve(null)
}, timeoutMs)
}
@@ -302,7 +321,7 @@ export async function dispatchClaudeTurn(
}
}
}
dropWaiter(session, waiter)
retireWaiter(session, waiter)
clearTimeout(waiter.timer)
waiter.resolve(null)
return { state: 'unknown', reason: (error as Error).message }
@@ -72,6 +72,10 @@ export type ClaudeSession = {
fence: number
prompts: ClaudePromptRegistry
dispatchWaiters: ClaudeDispatchWaiter[]
/** Uuids of dispatches that left the queue without their echo (timed out, or
* the send failed). Their replay can still arrive, and it must be dropped
* rather than fall through to the head - it belongs to nobody now. */
retiredSentUuids?: string[]
options: Map<string, string>
reportedOptions: { model?: string; effort?: string }
translator: ClaudeJournalTranslator | null