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.
This commit is contained in:
Brennan Benson
2026-11-02 03:40:05 -08:00
parent 43df8fce90
commit bcc7dd5aba
6 changed files with 294 additions and 67 deletions
@@ -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<string, unknown>): CapturedFoldFrame {
return { at, frame: { session_id: PROVIDER_SESSION_ID, parent_tool_use_id: null, ...frame } }
}
+118 -64
View File
@@ -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<string, ClaudeRunInput>()
private readonly bySubmission = new Map<string, ClaudeRunInput>()
private members = new Set<ClaudeRunInput>()
private owed = new Set<ClaudeRunInput>()
/** 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<ClaudeRunInput>()
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<string, unknown>): 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<T>(items: ReadonlySet<T>): T | undefined {
return items.values().next().value
}
+9 -2
View File
@@ -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
}
+2 -1
View File
@@ -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
@@ -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<string, unknown>) => {
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)
})
@@ -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)
}
)