mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
fix(native-chat): correlate a Claude replay to its dispatch by uuid
The CLI round-trips a client-supplied uuid verbatim. Measured:
stdin : {"type":"user","uuid":"11111111-2222-4333-8444-555555555555",...}
stdout: replay uuid = 11111111-2222-4333-8444-555555555555
So stamp one on the outgoing frame and match the echo on it. A replay now
settles the dispatch it belongs to, whatever its queue position - which closes
the limit this PR had been documenting as a redesign:
- a late replay no longer settles the NEXT dispatch after its own waiter timed
out, handing it the previous send's identity;
- the send-failure path now checks `waiter.settledUuid` - positive evidence that
THIS waiter was settled - instead of inferring adoption from queue membership,
which could return `accepted` carrying another dispatch's uuid.
The CLI also dedupes on the input uuid, so a resent frame cannot land twice.
The shape gates stay as the compat path for a CLI that mints its own uuid rather
than echoing ours; there the frame carries no correlation and the head is the
only candidate. Every pre-existing test exercises that path, since their frames
carry uuids that match no waiter.
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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<string, unknown>): 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<string, unknown>
|
||||
): 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<string | null> } {
|
||||
let waiter!: ClaudeDispatchWaiter
|
||||
const replayed = new Promise<string | null>((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 {
|
||||
|
||||
@@ -56,6 +56,13 @@ export type ClaudeDispatchWaiter = {
|
||||
resolve: (uuid: string | null) => void
|
||||
timer: ReturnType<typeof setTimeout>
|
||||
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 = {
|
||||
|
||||
Reference in New Issue
Block a user