Revert "Prototype source-batch journal ownership for Claude"

This reverts commit e5ed387f0c.

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<Record<string, unknown>>()
     const next = vi
       .fn<() => Promise<IteratorResult<Record<string, unknown>>>>()
@@ -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.

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<string, string>, keys: readonly string[]): Reco
 }
 
 export type ClaudeStreamJsonConnectionHandlers = {
-  reserveInput?: () => StructuredAgentSessionInputPermit | null
   onMessage?: (message: Record<string, unknown>) => 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) {

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

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 } : {}),

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) {

src/main/claude/claude-structured-session-test-support.ts

@@ -331,18 +331,7 @@ export const tick = (): Promise<void> => 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<string, unknown>): 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`. */

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<unknown> | void
 }
 
-type Runnable = Omit<StructuredAgentSessionSinkOperation, 'sequence'> & {
-  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<void> = Promise.resolve()
   private readonly buffered: Admitted[] = []
-  private readonly waitingPublications = new Map<string, Runnable>()
+  private readonly waitingPublications = new Map<string, Admitted>()
   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<StructuredAgentSessionSinkOperation, 'sequence'>,
-    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<unknown> {
     const key = operation.publicationKey
+    let outcome: Promise<unknown>
     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
 }
 

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)
       },

src/main/native-chat/agent-session-wire/structured-agent-session-input-capture.test.ts

@@ -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<void>()
-    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<void>()
-    releases.push(() => gate.resolve())
-    const predecessor = blocked ? rig.journal.readInOrder(() => gate.promise) : Promise.resolve()
-    const expectedBytes = estimateStructuredAgentSessionItemBytes(id('a'), body('a'))
-    let innerBarrier: ReturnType<NonNullable<typeof deferred.sink.written>> | 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])
-})

src/main/native-chat/agent-session-wire/structured-agent-session-input-capture.ts

@@ -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<StructuredAgentSessionSinkOperation, 'sequence'>
-type Capture = {
-  operation: Operation
-  children: Operation[]
-  outcomes: Promise<unknown>[]
-  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<unknown>
-      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<void>()
-    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()
-    }
-  }
-}