From bcc7dd5aba4b7aa05c15a490bae1ff5ddfeb4d6c Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 2 Nov 2026 03:40:05 -0800 Subject: [PATCH] Retire a Claude input on its own endpoint wherever it is owed A follow-up whose own result arrives after the turn it joined ended can no longer anchor the next message's turn: every endpoint Claude reports for an input (its named result, a terminal receipt, a refusal, or the end of the run it ran in) now retires it in one step, wherever it is waiting. Inputs are looked up by the ids Claude names them with instead of scanning every message a long turn has folded in, so a turn with many follow-ups stays cheap. Tracked in STA-9923. --- .../claude/claude-fold-steer-receipt.test.ts | 37 ++++ src/main/claude/claude-input-runs.ts | 182 ++++++++++++------ src/main/claude/claude-open-turn.ts | 11 +- src/main/claude/claude-turn-opening.ts | 3 +- ...structured-dispatch-queue-boundary.test.ts | 88 +++++++++ ...ed-agent-session-claude-precompact.test.ts | 40 ++++ 6 files changed, 294 insertions(+), 67 deletions(-) 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) + } +)