diff --git a/src/main/claude/claude-fold-steer-receipt.test.ts b/src/main/claude/claude-fold-steer-receipt.test.ts index 7fa475c1070..0c6bf2e96e0 100644 --- a/src/main/claude/claude-fold-steer-receipt.test.ts +++ b/src/main/claude/claude-fold-steer-receipt.test.ts @@ -642,6 +642,43 @@ describe('Claude fold receipt boundaries (synthetic orders)', () => { ) }) +describe('Claude input endpoints (synthetic orders)', () => { + it.each([ + ['plural', (uuid: string) => ({ user_message_uuids: [uuid] })], + ['singular', (uuid: string) => ({ user_message_uuid: uuid })] + ])('an owed input its own %s result names never owns a later run', async (_label, names) => { + const rig = await riggedAdapter() + const uuidA = await rig.dispatchAt(10, 'client-a', 'first prompt') + rig.deliver(userReplay(100, PROVIDER_SESSION_ID, uuidA, 'first prompt')) + await rig.dispatchAt(200, 'client-b', STEER_PROMPT) + rig.deliver( + resultFrame(300, PROVIDER_SESSION_ID, 'result-a', { + userMessageUuids: [uuidA], + durationMs: 290, + numTurns: 1 + }) + ) + const uuidB = String(rig.claude.connections[0]!.sent.at(-1)?.uuid) + rig.deliver( + untagged(400, { + type: 'result', + subtype: 'success', + is_error: false, + uuid: 'result-b', + ...names(uuidB) + }) + ) + const uuidC = await rig.dispatchAt(500, 'client-c', SECOND_STEER_PROMPT) + rig.deliver(userReplay(600, PROVIDER_SESSION_ID, uuidC, SECOND_STEER_PROMPT)) + + expect(rig.turns().at(-1)).toMatchObject({ + turnId: uuidC, + state: 'running', + userItemId: agentJournalSubmissionKey('client-c') + }) + }) +}) + function untagged(at: number, frame: Record): CapturedFoldFrame { return { at, frame: { session_id: PROVIDER_SESSION_ID, parent_tool_use_id: null, ...frame } } } diff --git a/src/main/claude/claude-input-runs.ts b/src/main/claude/claude-input-runs.ts index 212d44b3769..10e1cde7438 100644 --- a/src/main/claude/claude-input-runs.ts +++ b/src/main/claude/claude-input-runs.ts @@ -11,18 +11,23 @@ export type ClaudeOfferedInput = { requestedAt?: number } -type ClaudeRunInput = ClaudeOfferedInput & { - /** Its own receipt, echo, result correlation, or the run's first execution proved Claude ran it here. */ - participated: boolean +type ClaudeRunInput = { + readonly offered: ClaudeOfferedInput receipt?: 'queued' | 'started' | 'ended' /** The uuid Claude gave its echo; results may name the input by it. */ echoUuid?: string } -/** Orca inputs behind the open run, and those a finished run never consumed, owed to a later cycle. */ +/** Orca inputs behind the open run, and those a finished run never consumed, owed to a later cycle. + * Each input is reachable by every identity Claude names it with, so attribution never scans the run. */ export class ClaudeInputRuns { - private members: ClaudeRunInput[] = [] - private successors: ClaudeRunInput[] = [] + private readonly byUuid = new Map() + private readonly bySubmission = new Map() + private members = new Set() + private owed = new Set() + /** Members not yet proven to have run here (by their own receipt, echo, named result, or the run's + * first execution); only these move at a cycle boundary or run end. */ + private readonly provisional = new Set() private joinable = false /** A provider cycle began and has not reached its result or idle. */ private cycleOpen = false @@ -34,38 +39,45 @@ export class ClaudeInputRuns { private worked = false get tracked(): boolean { - return this.members.length > 0 + return this.members.size > 0 } get anchor(): ClaudeOfferedInput | null { - return this.members[0] ?? null + return first(this.members)?.offered ?? null } /** Root work outside a run belongs to an owed input Claude has not reported as merely queued. */ get successor(): ClaudeOfferedInput | null { - return ( - this.successors.find((input) => input.receipt === 'started') ?? - this.successors.find((input) => input.receipt !== 'queued') ?? - null - ) + let unreported: ClaudeRunInput | undefined + for (const input of this.owed) { + if (input.receipt === 'started') { + return input.offered + } + unreported ??= input.receipt === 'queued' ? undefined : input + } + return unreported?.offered ?? null } /** Claude runs owed inputs in order, so the oldest anchors the next run. */ get oldestOwed(): ClaudeOfferedInput | null { - return this.successors[0] ?? null + return first(this.owed)?.offered ?? null } owns(clientMessageId: string): boolean { - return this.members.some((input) => input.clientMessageId === clientMessageId) + const input = this.bySubmission.get(clientMessageId) + return input !== undefined && this.members.has(input) } /** True when the offer needs a new run; a live run gains it as a provisional member, a command owes it. */ - offer(input: ClaudeOfferedInput, runOpen: boolean): boolean { - const offered = { ...input, participated: false } + offer(offered: ClaudeOfferedInput, runOpen: boolean): boolean { + const input: ClaudeRunInput = { offered } + this.byUuid.set(offered.sentUuid, input) + this.bySubmission.set(offered.clientMessageId, input) if (runOpen && this.joinable) { - this.members.push(offered) + this.members.add(input) + this.provisional.add(input) } else { - this.successors.push(offered) + this.owed.add(input) } return !runOpen } @@ -77,25 +89,31 @@ export class ClaudeInputRuns { this.awaiting = offered this.predecessorOpen = offered && this.cycleOpen if (!joinable) { - this.members = [] return } - const anchor = this.successors.find((input) => input.clientMessageId === anchorClientMessageId) - this.members = anchor - ? [anchor, ...this.successors.filter((i) => i !== anchor)] - : this.successors - this.successors = [] + const owedAnchor = anchorClientMessageId && this.bySubmission.get(anchorClientMessageId) + const anchor = owedAnchor && this.owed.has(owedAnchor) ? owedAnchor : undefined + this.members = new Set(anchor ? [anchor, ...this.owed] : this.owed) + this.owed = new Set() + for (const input of this.members) { + this.provisional.add(input) + } if (anchor && (!offered || anchor.receipt === 'started')) { this.participate(anchor) } } - /** Inputs the ended run never ran stay owed; delivery waiters keep their own recovery. */ + /** Inputs the ended run consumed or Claude ended leave for good; only unconsumed offers stay owed. */ end(): void { - this.successors.push( - ...this.members.filter((input) => !input.participated && input.receipt !== 'ended') - ) - this.members = [] + for (const input of this.members) { + if (this.provisional.has(input) && input.receipt !== 'ended') { + this.owed.add(input) + } else { + this.retire(input) + } + } + this.members.clear() + this.provisional.clear() this.joinable = false this.awaiting = false this.predecessorOpen = false @@ -104,23 +122,27 @@ export class ClaudeInputRuns { /** The child is gone and can run nothing it was given. */ reset(): void { this.end() - this.successors = [] + this.owed.clear() + this.byUuid.clear() + this.bySubmission.clear() this.cycleOpen = false } /** Removes only the input Claude proved it never ran. */ withdraw(clientMessageId: string): 'kept' | 'reanchor' | 'retract' { - const matches = (input: ClaudeRunInput) => input.clientMessageId === clientMessageId - this.successors = this.successors.filter((input) => !matches(input)) - const index = this.members.findIndex(matches) - if (index === -1) { + const input = this.bySubmission.get(clientMessageId) + if (!input || !this.members.has(input)) { + if (input) { + this.retire(input) + } return 'kept' } - this.members.splice(index, 1) - if (this.members.length === 0) { + const anchored = first(this.members) === input + this.retire(input) + if (this.members.size === 0) { return this.worked ? 'kept' : 'retract' } - return index === 0 ? 'reanchor' : 'kept' + return anchored ? 'reanchor' : 'kept' } /** Attribute a root frame to this run's inputs and cycle before any of its effects apply. */ @@ -135,8 +157,11 @@ export class ClaudeInputRuns { } else if (type === 'system' && message.subtype === 'init') { // A new cycle after this run already answered is the missing-result boundary. if (responded) { - this.successors.push(...this.members.filter((input) => !input.participated)) - this.members = this.members.filter((input) => input.participated) + for (const input of this.provisional) { + this.members.delete(input) + this.owed.add(input) + } + this.provisional.clear() } this.predecessorOpen = false this.startCycle() @@ -145,16 +170,15 @@ export class ClaudeInputRuns { this.startCycle() } } else if (type === 'user') { - const input = this.members.find( - (member) => - member.clientMessageId === clientMessageId || - member.sentUuid === claudeText(message.uuid) || - member.sentUuid === claudeText(message.user_message_uuid) - ) - if (input) { - const echo = claudeText(message.uuid) + const input = + (clientMessageId === undefined ? undefined : this.bySubmission.get(clientMessageId)) ?? + this.named(message.uuid) ?? + this.named(message.user_message_uuid) + const echo = claudeText(message.uuid) + if (input && this.members.has(input)) { if (echo) { input.echoUuid = echo + this.byUuid.set(echo, input) } this.participate(input) } @@ -168,18 +192,25 @@ export class ClaudeInputRuns { } } - /** Correlated results end the run they name; an unnamed one only an executing run. */ + /** A named result is each named input's own endpoint wherever it is; it ends this run only if it names a member. + * An unnamed result ends only an executing run. */ resultEndsRun(message: Record): boolean { - const names = new Set(claudeResultUserMessageUuids(message)) - const named = claudeText(message.user_message_uuid) - if (named) { - names.add(named) + const names = claudeResultUserMessageUuids(message) + const singular = claudeText(message.user_message_uuid) + if (singular) { + names.push(singular) } - const hits = this.members.filter( - (input) => names.has(input.sentUuid) || (input.echoUuid && names.has(input.echoUuid)) - ) - hits.forEach((input) => this.participate(input)) - const ends = !this.tracked || (names.size > 0 ? hits.length > 0 : !this.awaiting) + let namesMember = false + for (const name of names) { + const input = this.byUuid.get(name) + if (input && this.members.has(input)) { + namesMember = true + this.participate(input) + } else if (input) { + this.retire(input) + } + } + const ends = !this.tracked || (names.length > 0 ? namesMember : !this.awaiting) this.cycleOpen = false this.predecessorOpen = false return ends @@ -193,8 +224,13 @@ export class ClaudeInputRuns { return ends } + private named(uuid: unknown): ClaudeRunInput | undefined { + const name = claudeText(uuid) + return name === null ? undefined : this.byUuid.get(name) + } + private observeReceipt(commandUuid: string | null, state: string | null): void { - const input = [...this.members, ...this.successors].find((i) => i.sentUuid === commandUuid) + const input = commandUuid === null ? undefined : this.byUuid.get(commandUuid) if (!input) { return } @@ -202,19 +238,33 @@ export class ClaudeInputRuns { input.receipt = 'started' // Claude runs one command at a time, so its start also ends any earlier cycle. this.predecessorOpen = false - if (this.members.includes(input)) { + if (this.members.has(input)) { this.participate(input) } } else if (state === 'queued') { input.receipt = input.receipt === 'started' ? 'started' : 'queued' } else if (state !== null) { input.receipt = 'ended' - this.successors = this.successors.filter((owed) => owed !== input) + if (!this.members.has(input)) { + this.retire(input) + } } } + /** The one way an input stops being owed: every identity and collection drops it together. */ + private retire(input: ClaudeRunInput): void { + this.byUuid.delete(input.offered.sentUuid) + if (input.echoUuid) { + this.byUuid.delete(input.echoUuid) + } + this.bySubmission.delete(input.offered.clientMessageId) + this.members.delete(input) + this.owed.delete(input) + this.provisional.delete(input) + } + private participate(input: ClaudeRunInput): void { - input.participated = true + this.provisional.delete(input) this.awaiting = false this.predecessorOpen = false this.cycleOpen = true @@ -225,10 +275,14 @@ export class ClaudeInputRuns { this.cycleOpen = true if (this.awaiting && !this.predecessorOpen) { this.awaiting = false - const anchor = this.members[0] + const anchor = first(this.members) if (anchor) { - anchor.participated = true + this.provisional.delete(anchor) } } } } + +function first(items: ReadonlySet): T | undefined { + return items.values().next().value +} diff --git a/src/main/claude/claude-open-turn.ts b/src/main/claude/claude-open-turn.ts index 0ff198aa237..2a4f4ab2663 100644 --- a/src/main/claude/claude-open-turn.ts +++ b/src/main/claude/claude-open-turn.ts @@ -15,11 +15,14 @@ import { } from './claude-turn-lifecycle-item' import { writeAgentJournalTurnRow } from '../native-chat/agent-session-timeline/agent-journal-turn-row-revision' import type { ClaudeCommandTurn } from './claude-command-turn' -import { createClaudeTurnOpener, type ClaudeTurnSource } from './claude-turn-opening' +import { + createClaudeTurnOpener, + isRootClaudeFrame, + type ClaudeTurnSource +} from './claude-turn-opening' import { refineClaudeQuotaTurnEnd } from './claude-terminal-failure' import type { ClaudeUsageLimitEvidence } from './claude-usage-limits' import { ClaudeInputRuns, type ClaudeOfferedInput } from './claude-input-runs' -import { isRootClaudeFrame } from './claude-turn-opening' export type ClaudeOpenTurnDeps = { sink: StructuredAgentSessionEventSink @@ -130,6 +133,10 @@ export class ClaudeOpenTurn { /** Orca's handoff owns the run it starts; a live run gains the input as a member. */ offerInput(input: ClaudeOfferedInput, observedAt: number): void { + // A host command's own input belongs to its command row. + if (this.current?.command?.sentUuid === input.sentUuid) { + return + } if (!this.runs.offer(input, this.current !== null)) { return } diff --git a/src/main/claude/claude-turn-opening.ts b/src/main/claude/claude-turn-opening.ts index e94d1ff6376..347054d4dfe 100644 --- a/src/main/claude/claude-turn-opening.ts +++ b/src/main/claude/claude-turn-opening.ts @@ -87,7 +87,8 @@ export function createClaudeTurnOpener(deps: { deps.open( { sessionId: source.sessionId, - turnId: owed?.sentUuid ?? source.uuid, + // The frame's own id: the owed input's later echo can then never collide with this row. + turnId: source.uuid, startedAt: observedAt, ...(owed?.requestedAt === undefined ? {} : { requestedAt: owed.requestedAt }), userItemId: owed diff --git a/src/main/claude/structured/claude-structured-dispatch-queue-boundary.test.ts b/src/main/claude/structured/claude-structured-dispatch-queue-boundary.test.ts index 965caafa132..c41ee447315 100644 --- a/src/main/claude/structured/claude-structured-dispatch-queue-boundary.test.ts +++ b/src/main/claude/structured/claude-structured-dispatch-queue-boundary.test.ts @@ -11,6 +11,7 @@ import { AgentSessionPreDispatchError } from '../../native-chat/agent-session-wi import { AgentSessionOptionRejectedError } from '../../native-chat/agent-session-wire/structured-agent-session-option-error' import { createClaudeUserMessageQueue } from '../claude-agent-sdk-user-message-queue' import { claudeTurnLifecycleIdentity } from '../claude-turn-lifecycle-item' +import { resolveClaudeReplayTurn } from '../claude-replay-turn-resolution' import { dispatchClaudeTurn, rejectClaudeDispatchWaiters } from './claude-structured-dispatch' import { createClaudeJournalTranslator } from './journal/claude-structured-journal-translation' import { sessionFor } from './claude-structured-dispatch-test-support' @@ -225,3 +226,90 @@ it('keeps the run for the input that survives when Claude refuses the one that o await rig.pump.next() await expect(joined).resolves.toMatchObject({ state: 'admitted' }) }) + +/** Offers and frames take acquisition's path: delivery resolution first, then the translator. */ +function attributionRig() { + const session = sessionFor() + session.connection.send = async (_message, beforeDispatch) => { + await beforeDispatch?.() + } + const translator = createClaudeJournalTranslator({ + sink: { appendItem: () => {}, appendTombstone: () => {}, publish: () => {} } + }) + session.translator = translator + let identityReads = 0 + const offerInput = translator.offerInput + // Count every read of an offered input's identity, wherever the run model keeps it. + translator.offerInput = (input) => { + const { sessionId, sentUuid, clientMessageId } = input + offerInput({ + sessionId, + get sentUuid() { + identityReads += 1 + return sentUuid + }, + get clientMessageId() { + identityReads += 1 + return clientMessageId + } + }) + } + const frame = (message: Record) => { + const full = { session_id: 'provider-session', parent_tool_use_id: null, ...message } + const origin = resolveClaudeReplayTurn(session, full) + translator.handle({ + type: 'message', + sessionId: 'session-1', + message: full, + ...(origin ? { startsTurn: true as const } : {}), + ...(origin?.clientMessageId ? { clientMessageId: origin.clientMessageId } : {}) + }) + } + const offer = (index: number) => + dispatchClaudeTurn( + session, + { body: USER_MESSAGE, clientMessageId: `client-${index}`, sentUuid: `sent-${index}` }, + async () => {} + ) + const runsFolded = async (count: number) => { + for (let index = 0; index <= count; index += 1) { + await offer(index) + frame({ type: 'command_lifecycle', state: 'started', command_uuid: `sent-${index}` }) + frame({ + type: 'user', + uuid: `sent-${index}`, + message: { role: 'user', content: [{ type: 'text', text: 'hi' }] } + }) + } + } + return { session, translator, frame, offer, runsFolded, reads: () => identityReads } +} + +it('attributes receipts, echoes and named results by identity, not by the size of the run', async () => { + const measure = async (count: number) => { + const rig = attributionRig() + await rig.runsFolded(count) + // Every input folded into one live run, and no delivery waiter is left to bound it. + expect(rig.translator.currentTurnId).toBe('sent-0') + expect(rig.session.dispatchWaiters).toHaveLength(0) + const beforeReceipt = rig.reads() + rig.frame({ type: 'command_lifecycle', state: 'completed', command_uuid: 'sent-1' }) + const receipt = rig.reads() - beforeReceipt + rig.frame({ + type: 'result', + subtype: 'success', + is_error: false, + user_message_uuids: ['sent-0'] + }) + expect(rig.translator.currentTurnId).toBeNull() + const total = rig.reads() + // The run released every input: the next send anchors its own run. + await rig.offer(count + 1) + expect(rig.translator.currentTurnId).toBe(`sent-${count + 1}`) + return { receipt, total } + } + const small = await measure(40) + const large = await measure(400) + expect(large.receipt).toBe(small.receipt) + expect(large.total).toBeLessThanOrEqual(10 * small.total + 20) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-precompact.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-precompact.test.ts index 88d89cfde38..c8a869b8473 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-precompact.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-precompact.test.ts @@ -323,3 +323,43 @@ it.each([ expect(completions[1]).not.toBe(prior) } ) + +it.each([ + ['plural', (uuid: unknown) => ({ user_message_uuids: [uuid] })], + ['singular', (uuid: unknown) => ({ user_message_uuid: uuid })] +])( + 'retires a follow-up its own %s result names, so it cannot anchor the next send', + async (_label, names) => { + const connection = fixture.claude.connections[0]! + await openTurn(connection, 'Say the word A.') + const uuidA = connection.sent.at(-1)!.uuid + const followUp = await sendUnechoed(connection) + const uuidB = connection.sent.at(-1)!.uuid + frame(connection, { + type: 'result', + subtype: 'success', + is_error: false, + uuid: 'result-a', + user_message_uuids: [uuidA] + }) + // B's own result arrives with no echo, init or output of B. + frame(connection, { + type: 'result', + subtype: 'success', + is_error: false, + uuid: 'result-b', + ...names(uuidB) + }) + expect(await dispatch(followUp)).toMatchObject({ state: 'accepted' }) + + const next = await send('Say the word C.') + await eventually(() => expect(connection.sent).toHaveLength(3)) + const written = connection.sent.at(-1)! + frame(connection, { ...written, parent_tool_use_id: null }) + const running = (await turns()).filter((turn) => turn.state === 'running') + expect(running).toEqual([ + expect.objectContaining({ turnId: written.uuid, userItemId: agentJournalSubmissionKey(next) }) + ]) + expect(connection.sent).toHaveLength(3) + } +)