diff --git a/src/main/claude/claude-stream-json-connection-close.test.ts b/src/main/claude/claude-stream-json-connection-close.test.ts index 92c3316fab2..5bb0a375474 100644 --- a/src/main/claude/claude-stream-json-connection-close.test.ts +++ b/src/main/claude/claude-stream-json-connection-close.test.ts @@ -1,23 +1,13 @@ import { EventEmitter } from 'node:events' import { PassThrough } from 'node:stream' import type { ChildProcessWithoutNullStreams } from 'node:child_process' -import { afterEach, describe, expect, it, vi } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import type { query } from '@anthropic-ai/claude-agent-sdk' import { CLAUDE_READER_DRAIN_AFTER_EXIT_MS, openClaudeStreamJsonConnection, type ClaudeStreamJsonLaunch } from './claude-stream-json-connection' -import { createClaudeJournalTranslator } from './claude-structured-journal-translation' -import { createDeferredStructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' -import { testEventSinkLogging } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support' -import { - closeProviderTimelineRigs, - openProviderTimelineRig, - SESSION -} from '../native-chat/agent-session-timeline/provider-timeline-assembler-test-support' - -afterEach(closeProviderTimelineRigs) const mocks = vi.hoisted(() => { const refresh = vi.fn() @@ -106,16 +96,12 @@ describe('Claude stream-json close ordering', () => { await expect(connection.close()).resolves.toBe(true) }) - it.each([false, true])('retains an SDK result resolved just before root exit (reserved=%s)', async (reserved) => { + it('releases a pulled frame when provider exit is reported', async () => { mocks.refresh.mockReset() mocks.proveClaudeChildExit.mockReset() mocks.refresh.mockResolvedValue(undefined) mocks.proveClaudeChildExit.mockResolvedValue(true) const child = fakeChild() - const rig = await openProviderTimelineRig() - const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging()) - deferred.bind({ journal: rig.journal, fence: 1, publish: () => {} }) - const translator = createClaudeJournalTranslator({ sink: deferred.sink }) const first = Promise.withResolvers>() const next = vi .fn<() => Promise>>>() @@ -137,49 +123,21 @@ describe('Claude stream-json close ordering', () => { const connection = await openClaudeStreamJsonConnection( { pathToClaudeCodeExecutable: 'claude', options: {}, cwd: '/work/repo' }, { - ...(reserved ? { reserveInput: deferred.sink.reserveInput } : {}), - onMessage: (message) => { - events.push(`message:${String(message.type)}`) - translator.handle({ type: 'message', sessionId: SESSION, message }) - }, - onExit: () => { - events.push('exit') - deferred.sink.sealInput?.(() => { - translator.handle({ type: 'ended', sessionId: SESSION, reason: 'root exit' }) - translator.dispose() - }) - } + onMessage: (message) => events.push(`message:${String(message.type)}`), + onExit: () => events.push('exit') }, () => child, queryImpl ) connection.pauseReading?.() + first.resolve({ type: 'task_notification' }) await vi.waitFor(() => expect(next).toHaveBeenCalledOnce()) expect(events).toEqual([]) - // Resolve and exit in one stack, before the awaiting reader can consume the complete result. - first.resolve({ - type: 'user', - uuid: 'resolved-before-exit', - session_id: 'provider-session', - message: { - role: 'user', - content: [{ type: 'tool_result', tool_use_id: 'received', content: 'MIDDLE:TAIL' }] - } - }) + child.emit('exit', 1, null) + await vi.waitFor(() => expect(events).toEqual(['exit', 'message:task_notification'])) await expect(connection.close()).resolves.toBe(true) - await deferred.drained() - const reopened = await rig.reopenJournal() - expect(reopened.snapshot().items.map((row) => row.body)).toContainEqual( - expect.objectContaining({ - kind: 'tool-call', - callId: 'received', - output: expect.objectContaining({ head: 'MIDDLE:TAIL', truncated: false }) - }) - ) - expect(events).toEqual(['exit', 'message:user']) - deferred.close() }) // Stdout can end before the exit is seen, as when the provider supervisor's own exit closes it. diff --git a/src/main/claude/claude-stream-json-connection.ts b/src/main/claude/claude-stream-json-connection.ts index 8ad9512a175..828a58e043a 100644 --- a/src/main/claude/claude-stream-json-connection.ts +++ b/src/main/claude/claude-stream-json-connection.ts @@ -21,7 +21,6 @@ import { import type { ClaudeStructuredSdkOptions } from './claude-structured-launch-resolution' import { providerStderrForDisplay } from '../provider-process/provider-spawn-failure-report' import { withMissingProviderExecutable } from '../provider-process/provider-executable-missing' -import type { StructuredAgentSessionInputPermit } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' export { ClaudeControlRequestError } @@ -67,7 +66,6 @@ function withoutKeys(env: Record, keys: readonly string[]): Reco } export type ClaudeStreamJsonConnectionHandlers = { - reserveInput?: () => StructuredAgentSessionInputPermit | null onMessage?: (message: Record) => void /** * The SDK owns inbound permission control: it hands `can_use_tool` to this callback with @@ -240,38 +238,16 @@ export async function openClaudeStreamJsonConnection( try { for (;;) { await waitUntilReadable() - const permit = handlers.reserveInput?.() - if (handlers.reserveInput && !permit) { - if (closing || rootSettled() || terminalError) { - break - } - await waitUntilReadable() - continue + const next = await iterator.next() + if (next.done) { + completed = true + break } - try { - const next = await iterator.next() - if (next.done) { - completed = true - break - } - if (!permit) { - await waitUntilReadable() - } - if (!isRecord(next.value)) { - throw new Error('claude stream-json yielded a non-object message') - } - const message = next.value - const receive = (): void => { - handlers.onMessage?.(message) - } - if (permit) { - permit.run(receive) - } else { - receive() - } - } finally { - permit?.cancel() + await waitUntilReadable() + if (!isRecord(next.value)) { + throw new Error('claude stream-json yielded a non-object message') } + handlers.onMessage?.(next.value) } } finally { if (!completed) { diff --git a/src/main/claude/claude-structured-event-delivery.ts b/src/main/claude/claude-structured-event-delivery.ts index 9e3dd8c639a..5430eb82f8c 100644 --- a/src/main/claude/claude-structured-event-delivery.ts +++ b/src/main/claude/claude-structured-event-delivery.ts @@ -40,17 +40,7 @@ export function emitClaudeStructuredSessionEvent({ if (event.type === 'message' && session?.commands.observe(event.message)) { session.events?.publish() } - const translate = (): void => { - session?.translator?.handle(event) - if (event.type === 'ended') { - session?.translator?.dispose() - } - } - if (event.type === 'ended' && session?.events?.sealInput) { - session.events.sealInput(translate) - } else { - translate() - } + session?.translator?.handle(event) deps.onEvent?.(event) const retired = event.type === 'message' && session ? retireClaudeLaunchedModel(session, event.message) : null diff --git a/src/main/claude/claude-structured-session-acquisition.ts b/src/main/claude/claude-structured-session-acquisition.ts index 14bf0d31931..749bc825849 100644 --- a/src/main/claude/claude-structured-session-acquisition.ts +++ b/src/main/claude/claude-structured-session-acquisition.ts @@ -187,7 +187,6 @@ export async function acquireClaudeSession({ }, { onMessage, - ...(input.events?.reserveInput ? { reserveInput: input.events.reserveInput } : {}), canUseTool, onUserDialog, ...(input.onOutput ? { onOutput: input.onOutput } : {}), diff --git a/src/main/claude/claude-structured-session-close.ts b/src/main/claude/claude-structured-session-close.ts index 1640028b765..316d90b0a50 100644 --- a/src/main/claude/claude-structured-session-close.ts +++ b/src/main/claude/claude-structured-session-close.ts @@ -75,11 +75,7 @@ export function settleClaudeExitedSession(session: ClaudeSession): void { for (const prompt of session.prompts.clear()) { prompt.settle(null) } - if (session.events?.sealInput) { - session.events.sealInput(() => session.translator?.dispose()) - } else { - session.translator?.dispose() - } + session.translator?.dispose() } type CloseClaudePublishedSessionInput = { @@ -153,24 +149,17 @@ async function finalizeClaudePublishedSession( acquisitionGeneration: session.acquisitionGeneration, observedAt: Date.now() } - const finish = (): void => { + try { try { session.translator?.handle(ended) - } finally { - session.translator?.dispose() + } catch (error) { + callbackThrew = true + callbackError ??= error } + deliver(ended) + } finally { + session.translator?.dispose() } - try { - if (session.events?.sealInput) { - session.events.sealInput(finish) - } else { - finish() - } - } catch (error) { - callbackThrew = true - callbackError ??= error - } - deliver(ended) } session.closePersistence ??= persistClosedClaudeSession(input, session) if (rootExitVerdict) { diff --git a/src/main/claude/claude-structured-session-test-support.ts b/src/main/claude/claude-structured-session-test-support.ts index b5674412a12..f1df2d51fe4 100644 --- a/src/main/claude/claude-structured-session-test-support.ts +++ b/src/main/claude/claude-structured-session-test-support.ts @@ -331,18 +331,7 @@ export const tick = (): Promise => new Promise((resolve) => setImmediate(r /** Delivers one frame from Claude on `connection`, under the provider session it runs. */ export function claudeFrame(connection: FakeConnection, message: Record): void { - const receive = (): void => { - connection.handlers.onMessage?.({ session_id: PROVIDER_SESSION_ID, ...message }) - } - if (connection.handlers.reserveInput) { - const permit = connection.handlers.reserveInput() - if (!permit) { - throw new Error('test delivered a Claude frame while its reader was paused') - } - permit.run(receive) - } else { - receive() - } + connection.handlers.onMessage?.({ session_id: PROVIDER_SESSION_ID, ...message }) } /** Whether anything sent to Claude on `connection` carries `text`. */ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts index 21f031e087a..2adf4328944 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts @@ -9,7 +9,6 @@ import type { StructuredAgentSessionSinkWatermarks } from './structured-agent-session-event-sink' import type { StructuredAgentSessionTransitionJournal } from './structured-agent-session-transition' -import { StructuredAgentSessionInputCapture } from './structured-agent-session-input-capture' import { StructuredAgentSessionFinalTailReservation } from './structured-agent-session-final-tail-reservation' export type StructuredAgentSessionSinkOperation = { @@ -29,14 +28,11 @@ export type StructuredAgentSessionSinkOperation = { run: (target: StructuredAgentSessionEventTarget) => Promise | void } -type Runnable = Omit & { - sequence?: number +type Admitted = StructuredAgentSessionSinkOperation & { superseded?: boolean releaseFinalTail?: () => void } -type Admitted = Runnable & { sequence: number } - export type StructuredAgentSessionDrainWaiter = { through: number resolve: (result: StructuredAgentSessionSinkBarrier) => void @@ -56,22 +52,6 @@ export class StructuredAgentSessionSinkQueue { private failure: { error: unknown } | null = null private queuedBytes = 0 private queuedOperations = 0 - private ordinaryOperations = 0 - private readonly input = new StructuredAgentSessionInputCapture({ - open: () => !this.closed && this.failure === null, - canReserve: () => { - if (this.ordinaryOperations >= this.deps.watermarks.maxQueuedOperations) { - this.setBackpressure(true) - } - return !this.backpressured - }, - submit: (operation) => this.submit(operation, {}, true), - charge: (bytes) => { - this.queuedBytes += bytes - }, - run: (operation, bound) => this.runOperation(operation, bound), - pressure: () => this.updateBackpressure() - }) private lifecycleQueuedBytes = 0 private lifecycleQueuedOperations = 0 private backpressured = false @@ -82,7 +62,7 @@ export class StructuredAgentSessionSinkQueue { /** Settles once every operation handed over so far has, in handover order. */ private handedOverSettled: Promise = Promise.resolve() private readonly buffered: Admitted[] = [] - private readonly waitingPublications = new Map() + private readonly waitingPublications = new Map() private readonly waiters: StructuredAgentSessionDrainWaiter[] = [] private readonly finalTail = new StructuredAgentSessionFinalTailReservation() @@ -146,7 +126,6 @@ export class StructuredAgentSessionSinkQueue { /** Operations already handed over are the journal's and still land; only buffered ones drop. */ close(): void { this.closed = true - this.input.cancel() this.dropBuffered() this.finalTail.close() this.updateBackpressure() @@ -169,13 +148,9 @@ export class StructuredAgentSessionSinkQueue { : settled } - reserveInput = this.input.reserve - sealInput = this.input.seal - submit( operation: Omit, - options: StructuredAgentSessionAppendOptions = {}, - reserved = false + options: StructuredAgentSessionAppendOptions = {} ): StructuredAgentSessionSinkAdmission { if (this.closed) { return { accepted: false, reason: 'closed' } @@ -183,9 +158,6 @@ export class StructuredAgentSessionSinkQueue { if (this.failure !== null) { return { accepted: false, reason: 'failed' } } - if (!reserved && this.input.append(operation)) { - return { accepted: true } - } const key = operation.publicationKey const replaceAt = key === undefined ? -1 : this.buffered.findIndex((queued) => queued.publicationKey === key) @@ -195,27 +167,21 @@ export class StructuredAgentSessionSinkQueue { : key === undefined ? undefined : this.waitingPublications.get(key) - const charged = replaced?.sequence === undefined ? undefined : replaced const finalTail = operation.finalTail === true const lifecycle = !finalTail && (operation.lifecycle ?? options.lifecycle === true) const lifecycleBytes = lifecycle ? (operation.lifecycleBytes ?? operation.bytes) : 0 - const nextBytes = this.queuedBytes - (charged?.bytes ?? 0) + operation.bytes - const nextOperations = this.queuedOperations + (charged ? 0 : 1) - const nextOrdinaryOperations = - this.ordinaryOperations - (charged && !charged.lifecycle && !charged.finalTail ? 1 : 0) + - (!lifecycle && !finalTail ? 1 : 0) + const nextBytes = this.queuedBytes - (replaced?.bytes ?? 0) + operation.bytes + const nextOperations = this.queuedOperations + (replaced ? 0 : 1) const nextLifecycleBytes = - this.lifecycleQueuedBytes - (charged ? lifecycleCost(charged) : 0) + lifecycleBytes + this.lifecycleQueuedBytes - (replaced ? lifecycleCost(replaced) : 0) + lifecycleBytes const nextLifecycleOperations = - this.lifecycleQueuedOperations - (charged?.lifecycle ? 1 : 0) + (lifecycle ? 1 : 0) + this.lifecycleQueuedOperations - (replaced?.lifecycle ? 1 : 0) + (lifecycle ? 1 : 0) // A byte watermark paces waiting work; an empty budget can own one complete operation. const exceedsOrdinary = - !reserved && !finalTail && !lifecycle && ((this.queuedOperations > 0 && nextBytes > this.deps.watermarks.maxQueuedBytes) || - nextOrdinaryOperations + (this.input.reserved ? 1 : 0) > - this.deps.watermarks.maxQueuedOperations) + nextOperations > this.deps.watermarks.maxQueuedOperations) const exceedsLifecycle = lifecycle && ((this.lifecycleQueuedOperations > 0 && @@ -231,19 +197,19 @@ export class StructuredAgentSessionSinkQueue { this.fail(new Error('structured agent-session final tail exceeded its reserved capacity')) return { accepted: false, reason: 'failed' } } - const accepted: Admitted = Object.assign(reserved ? operation : { ...operation }, { + const accepted: Admitted = { + ...operation, sequence: ++this.acceptedSequence, lifecycle, lifecycleBytes, releaseFinalTail: releaseFinalTail ?? undefined - }) + } if (replaceAt >= 0) { this.buffered.splice(replaceAt, 1) } else if (replaced) { // Its turn still comes, and settles as nothing: the publication after it covers it. replaced.superseded = true } - this.ordinaryOperations = nextOrdinaryOperations this.queuedBytes = nextBytes this.queuedOperations = nextOperations this.lifecycleQueuedBytes = nextLifecycleBytes @@ -260,31 +226,15 @@ export class StructuredAgentSessionSinkQueue { private handOver(operation: Admitted, bound: StructuredAgentSessionEventTarget): void { const earlier = this.handedOverSettled this.handedOverSequence = operation.sequence - const outcome = this.runOperation(operation, bound) - // Handled at once: a write can fail while an earlier one is still landing. - const landed = outcome.then(() => undefined, this.fail) - this.handedOverSettled = Promise.all([earlier, landed]).then(() => this.settle(operation)) - } - - private runOperation( - operation: Runnable, - bound: StructuredAgentSessionEventTarget - ): Promise { const key = operation.publicationKey + let outcome: Promise if (key === undefined) { - return runNow(() => operation.run(bound)) + outcome = runNow(() => operation.run(bound)) } else { - const replaced = this.waitingPublications.get(key) - if (replaced && !replaced.superseded) { - replaced.superseded = true - if (replaced.sequence !== undefined) { - this.release(replaced) - } - } this.waitingPublications.set(key, operation) // At handover, unless a write is running or writes wait in line; then at its place in it, so // it never announces ahead of the writes issued before it. - return runNow(() => + outcome = runNow(() => bound.journal.readInOrder(() => { if (this.waitingPublications.get(key) === operation) { this.waitingPublications.delete(key) @@ -296,6 +246,9 @@ export class StructuredAgentSessionSinkQueue { }) ) } + // Handled at once: a write can fail while an earlier one is still landing. + const landed = outcome.then(() => undefined, this.fail) + this.handedOverSettled = Promise.all([earlier, landed]).then(() => this.settle(operation)) } private settle(operation: Admitted): void { @@ -307,11 +260,8 @@ export class StructuredAgentSessionSinkQueue { this.settleWaiters() } - private release(operation: Runnable): void { + private release(operation: Admitted): void { operation.releaseFinalTail?.() - if (!operation.lifecycle && !operation.finalTail) { - this.ordinaryOperations = Math.max(0, this.ordinaryOperations - 1) - } this.queuedBytes = Math.max(0, this.queuedBytes - operation.bytes) this.queuedOperations = Math.max(0, this.queuedOperations - 1) if (operation.lifecycle) { @@ -378,7 +328,6 @@ export class StructuredAgentSessionSinkQueue { /** Writes handed over before this runs still land; nothing submitted after it is admitted. */ private fail = (error: unknown): void => { if (this.failure === null) { - this.input.cancel() this.failure = { error } this.deps.onFailed?.(error) } @@ -388,7 +337,7 @@ export class StructuredAgentSessionSinkQueue { } } -function lifecycleCost(operation: Runnable): number { +function lifecycleCost(operation: StructuredAgentSessionSinkOperation): number { return operation.lifecycle ? (operation.lifecycleBytes ?? operation.bytes) : 0 } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts index d414ca8b43c..1f48f068058 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts @@ -29,11 +29,6 @@ export type StructuredAgentSessionSinkState = { failed: boolean } -export type StructuredAgentSessionInputPermit = { - run: (receive: () => void) => void - cancel: () => void -} - export type StructuredAgentSessionSinkBarrier = { ok: true } | { ok: false; error: unknown } /** Linkage a producer stamps on the rows it writes. Absent on every append the @@ -90,10 +85,6 @@ export type StructuredAgentSessionRevisionResolver = ( export type StructuredAgentSessionRevisionOptions = StructuredAgentSessionItemAppendOptions export type StructuredAgentSessionEventSink = StructuredAgentSessionTransitionSink & { - /** Reserve before consuming a source batch; an unused permit owns no journal sequence. */ - reserveInput?(): StructuredAgentSessionInputPermit | null - /** Transfer the finite stopped producer's remaining writes before its lifecycle notification. */ - sealInput?(receive: () => void): void appendItem( identity: AgentJournalItemIdentity, body: AgentJournalItemBody, @@ -295,8 +286,6 @@ export function createDeferredStructuredAgentSessionEventSink(deps: { return { sink: { - reserveInput: queue.reserveInput, - sealInput: queue.sealInput, appendItem: (identity, body, options) => { appendItem(identity, body, options) }, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-input-capture.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-input-capture.test.ts deleted file mode 100644 index e7055788171..00000000000 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-input-capture.test.ts +++ /dev/null @@ -1,110 +0,0 @@ -import { afterEach, expect, it, vi } from 'vitest' -import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' -import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' -import { adapterFor, claudeFrame, fakeClaude, identityFor } from '../../claude/claude-structured-session-test-support' -import { closeProviderTimelineRigs, openProviderTimelineRig, SESSION } from '../agent-session-timeline/provider-timeline-assembler-test-support' -import { createDeferredStructuredAgentSessionEventSink } from './structured-agent-session-event-sink' -import { estimateStructuredAgentSessionItemBytes } from './structured-agent-session-event-sink-estimate' -import { testEventSinkLogging } from './structured-agent-session-logger-test-support' - -const options = { turnScope: AGENT_JOURNAL_THREAD_SCOPE } -const id = (recordId: string) => ({ provider: 'legacy' as const, agent: 'grok', sessionId: SESSION, recordId }) -const body = (text: string) => ({ kind: 'status' as const, text, tone: 'info' as const }) -const releases: (() => void)[] = [] -afterEach(async () => { releases.splice(0).forEach((release) => release()); await closeProviderTimelineRigs() }) - -it.each([false, true])('owns a real Claude envelope before cleanup (failed=%s)', async (failed) => { - for (const count of [1, 1025]) { - const rig = await openProviderTimelineRig() - const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging()) - deferred.bind({ journal: rig.journal, fence: 1, publish: () => {} }) - const claude = fakeClaude(), adapter = adapterFor(claude) - await adapter.acquire({ identity: identityFor(SESSION), fence: 1, spawnToken: 'capture-test', events: deferred.sink }) - await deferred.drained() - const gate = Promise.withResolvers() - releases.push(() => gate.resolve()) - const blocked = rig.journal.readInOrder(() => gate.promise) - deferred.sink.appendItem(id('before'), body('before'), options) - const before = deferred.state() - const append = vi.spyOn(rig.journal, 'appendItem') - const text = count === 1 ? `${'\u0001'.repeat(6 * 1024 * 1024)}MIDDLE:TAIL` : 'MIDDLE:TAIL' - const content = Array.from({ length: count }, (_, index) => ({ type: 'tool_result', tool_use_id: `result-${index}`, content: `${text}:${index}`, is_error: failed })) - claudeFrame(claude.connections[0], { type: 'user', uuid: 'capture-envelope', message: { role: 'user', content } }) - expect(append).toHaveBeenCalledTimes(count) - const charged = append.mock.calls.reduce((total, [identity, value]) => total + estimateStructuredAgentSessionItemBytes(identity, value), 0) - expect(deferred.state().queuedOperations - before.queuedOperations).toBe(1) - expect(deferred.state().queuedBytes - before.queuedBytes).toBeGreaterThanOrEqual(charged) - let landed = false - if (failed) { - await adapter.releaseAcquisition({ sessionId: SESSION }) - } else { - claude.connections[0].handlers.onExit?.(new Error('provider exited')) - await adapter.drainObservedExits() - } - const barrier = deferred.sink.written?.().then((result) => { landed = true; return result }) - await Promise.resolve() - expect(landed).toBe(false) - expect(append).toHaveBeenCalledTimes(count) - gate.resolve() - await blocked - expect(await barrier).toEqual({ ok: true }) - await deferred.drained() - const reopened = await rig.reopenJournal() - expect(reopened.snapshot().items.flatMap(({ body: value }) => value.kind === 'tool-call' ? [value] : [])).toEqual(content.map((result) => expect.objectContaining({ callId: result.tool_use_id, state: failed ? 'failed' : 'completed', output: expect.objectContaining({ head: result.content, truncated: false }) }))) - deferred.close() - } -}) - -it.each([false, true])('preserves journal planning, publication positions and barriers (blocked=%s)', async (blocked) => { - const results = [] - for (const captured of [false, true]) { - const rig = await openProviderTimelineRig() - const publications: unknown[] = [] - const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging()) - deferred.bind({ journal: rig.journal, fence: 1, publish: () => publications.push(rig.journal.snapshot().items.map((row) => row.body)) }) - const unused = deferred.sink.reserveInput?.() - expect(deferred.state().queuedOperations).toBe(0) - expect(await deferred.drained()).toEqual({ ok: true }) - unused?.cancel() - const gate = Promise.withResolvers() - releases.push(() => gate.resolve()) - const predecessor = blocked ? rig.journal.readInOrder(() => gate.promise) : Promise.resolve() - const expectedBytes = estimateStructuredAgentSessionItemBytes(id('a'), body('a')) - let innerBarrier: ReturnType> | undefined - const receive = (): void => { - deferred.sink.appendItem(id('a'), body('a'), options) - innerBarrier = deferred.sink.written?.() - expect(rig.journal.itemBody(agentJournalItemKey(id('a')))).toEqual(blocked ? null : body('a')) - deferred.sink.publish({ coalescingKey: 'repeat' }) - void rig.journal.appendItem(id('host'), body('host'), { fence: 1 }) - deferred.sink.tryAppendResolvedItemAndPublish?.(id('resolved'), body('resolved'), (journal) => { - const preceding: string[] = [] - journal.visitItems((itemId) => { preceding.push(itemId) }) - expect(preceding).toContain(agentJournalItemKey(id('host'))) - return id('resolved') - }, options) - deferred.sink.tryAppendTransition?.({ publish: true, lifecycle: false, steps: [{ kind: 'item', reservedBytes: estimateStructuredAgentSessionItemBytes(id('transition'), body('transition')), resolve: (journal) => { - expect(journal.itemBody(agentJournalItemKey(id('resolved')))).toEqual(body('resolved')) - return { identity: id('transition'), body: body('transition') } - }, options }] }) - deferred.sink.tryAppendLifecycleBatch?.('batch', [{ kind: 'item', identity: id('life'), body: body('life'), turnScope: AGENT_JOURNAL_THREAD_SCOPE }]) - deferred.sink.appendTombstone(id('a')) - deferred.sink.publish({ coalescingKey: 'repeat' }) - throw new Error('source callback failed after its received prefix') - } - expect(() => captured ? deferred.sink.reserveInput?.()?.run(receive) : receive()).toThrow('source callback failed') - expect(deferred.state().queuedBytes).toBeGreaterThanOrEqual(expectedBytes) - const registration = vi.spyOn(rig.journal, 'appendItem') - deferred.sink.sealInput?.(() => deferred.sink.appendItem(id('tail'), body('tail'), options)) - expect(registration).toHaveBeenCalledTimes(1) - gate.resolve() - await predecessor - expect(await innerBarrier).toEqual({ ok: true }) - expect(await deferred.drained()).toEqual({ ok: true }) - expect(deferred.state().queuedBytes).toBe(0) - expect(deferred.state().queuedOperations).toBe(0) - results.push({ rows: (await rig.reopenJournal()).snapshot().items.map((row) => row.body), publications }) - deferred.close() - } - expect(results[1]).toEqual(results[0]) -}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-input-capture.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-input-capture.ts deleted file mode 100644 index 16fec206ad6..00000000000 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-input-capture.ts +++ /dev/null @@ -1,128 +0,0 @@ -import type { - StructuredAgentSessionEventTarget, - StructuredAgentSessionInputPermit, - StructuredAgentSessionSinkAdmission -} from './structured-agent-session-event-sink' -import type { StructuredAgentSessionSinkOperation } from './structured-agent-session-event-sink-queue' - -type Operation = Omit -type Capture = { - operation: Operation - children: Operation[] - outcomes: Promise[] - target: StructuredAgentSessionEventTarget | null - started: boolean -} - -/** One source stack owns one queue operation; child journal calls keep their original positions. */ -export class StructuredAgentSessionInputCapture { - private permit: StructuredAgentSessionInputPermit | null = null - private capture: Capture | null = null - private sealed = false - - constructor( - private readonly deps: { - open: () => boolean - canReserve: () => boolean - submit: (operation: Operation) => StructuredAgentSessionSinkAdmission - charge: (bytes: number) => void - run: (operation: Operation, target: StructuredAgentSessionEventTarget) => Promise - pressure: () => void - } - ) {} - - get reserved(): boolean { - return this.permit !== null - } - - cancel = (): void => { - this.permit?.cancel() - } - - reserve = (): StructuredAgentSessionInputPermit | null => { - if (this.sealed || !this.deps.open() || !this.deps.canReserve()) { - return null - } - if (this.permit) { - return this.permit - } - let available = true - const permit: StructuredAgentSessionInputPermit = { - cancel: () => { - available = false - if (this.permit === permit) { - this.permit = null - } - }, - run: (receive) => { - if (!available || !this.deps.open()) { - throw new Error('structured agent-session input permit is no longer available') - } - permit.cancel() - this.receive(receive, false) - } - } - this.permit = permit - return permit - } - - seal = (receive: () => void): void => { - this.sealed = true - this.cancel() - this.receive(receive, true) - } - - append(operation: Operation): boolean { - const capture = this.capture - if (!capture) { - return false - } - if (!capture.started) { - const admission = this.deps.submit(capture.operation) - if (!admission.accepted) { - throw new Error(`structured agent-session capture refused: ${admission.reason}`) - } - capture.started = true - } - capture.operation.bytes += operation.bytes - this.deps.charge(operation.bytes) - capture.children.push(operation) - if (capture.target) { - capture.outcomes.push(this.deps.run(operation, capture.target)) - } - return true - } - - private receive(receive: () => void, finalTail: boolean): void { - if (this.capture) { - receive() - return - } - const done = Promise.withResolvers() - const capture: Capture = { - children: [], - outcomes: [], - target: null, - started: false, - operation: { - bytes: 0, - ...(finalTail ? { finalTail: true } : {}), - run: (bound) => { - capture.target = bound - for (const child of capture.children) { - capture.outcomes.push(this.deps.run(child, bound)) - } - return done.promise.then(() => Promise.all(capture.outcomes)) - } - } - } - this.capture = capture - try { - receive() - } finally { - this.capture = null - done.resolve() - this.deps.pressure() - } - } -}