Join structured attach teardown through journal bind

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

@@ -56,6 +56,7 @@ import {
   AgentSessionSubscribers,
   type AgentSessionSubscriberEmit
 } from './structured-agent-session-subscribers'
+import { StructuredAgentSessionTaskQueue } from './structured-agent-session-task-queue'
 import { restoreStructuredAgentSessionRead } from './structured-agent-session-read-restore'
 
 export type StructuredAgentSessionCaller = {
@@ -88,7 +89,7 @@ type SessionState = {
 export class StructuredAgentSessionHost {
   private readonly sessions = new Map<string, SessionState>()
   private readonly subscribers = new AgentSessionSubscribers()
-  private readonly chains = new Map<string, Promise<unknown>>()
+  private readonly tasks = new StructuredAgentSessionTaskQueue()
   private readonly eventSinks = new Map<string, DeferredStructuredAgentSessionEventSink>()
   private readonly reconcileLeases: (sessionId: string) => Promise<AgentSessionWireRefusal | null>
 
@@ -147,13 +148,7 @@ export class StructuredAgentSessionHost {
 
   /** Per-session single file. A rejected task must not poison the chain. */
   private serialize<T>(sessionId: string, task: () => Promise<T>): Promise<T> {
-    const prior = this.chains.get(sessionId) ?? Promise.resolve()
-    const next = prior.then(task, task)
-    this.chains.set(
-      sessionId,
-      next.catch(() => undefined)
-    )
-    return next
+    return this.tasks.serialize(sessionId, task)
   }
 
   /**
@@ -164,7 +159,7 @@ export class StructuredAgentSessionHost {
     caller: StructuredAgentSessionCaller,
     params: AgentSessionAttachParams
   ): Promise<AgentSessionMutationResult<AgentSessionAttachResult>> {
-    return this.serialize(params.envelope.sessionId, async () => {
+    const attaching = this.serialize(params.envelope.sessionId, async () => {
       const sessionId = params.envelope.sessionId
       const unreconciled = await this.reconcileLeases(sessionId)
       if (unreconciled) {
@@ -219,6 +214,7 @@ export class StructuredAgentSessionHost {
       }
       return attached
     })
+    return this.tasks.trackAttach(attaching)
   }
 
   /** Settles every row the provider streamed. Callers that must read the
@@ -229,6 +225,8 @@ export class StructuredAgentSessionHost {
 
   /** Settles final provider rows after every child has stopped producing them. */
   async flushAllStreamedEvents(): Promise<void> {
+    // An acquired provider can still be opening its journal; bind before draining.
+    await this.tasks.drainAttaches()
     await Promise.all([...this.eventSinks.values()].map((sink) => sink.drained()))
   }
 

src/main/native-chat/agent-session-wire/structured-agent-session-task-queue.ts

@@ -0,0 +1,29 @@
+export class StructuredAgentSessionTaskQueue {
+  private readonly chains = new Map<string, Promise<unknown>>()
+  private readonly attaching = new Set<Promise<unknown>>()
+
+  serialize<T>(sessionId: string, task: () => Promise<T>): Promise<T> {
+    const prior = this.chains.get(sessionId) ?? Promise.resolve()
+    const next = prior.then(task, task)
+    this.chains.set(
+      sessionId,
+      next.catch(() => undefined)
+    )
+    return next
+  }
+
+  trackAttach<T>(operation: Promise<T>): Promise<T> {
+    this.attaching.add(operation)
+    void operation.then(
+      () => this.attaching.delete(operation),
+      () => this.attaching.delete(operation)
+    )
+    return operation
+  }
+
+  async drainAttaches(): Promise<void> {
+    while (this.attaching.size > 0) {
+      await Promise.allSettled(this.attaching)
+    }
+  }
+}

src/main/runtime/structured-agent-session-integration.test.ts

@@ -15,6 +15,7 @@ import type {
   CodexAppServerConnectionHandlers,
   openCodexAppServerConnection
 } from '../codex/codex-app-server-connection'
+import type { CodexStructuredSessionAdapter } from '../codex/codex-structured-session-adapter'
 import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope'
 import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../shared/protocol-version'
 import type { AgentJournalRenderItem } from '../../shared/agent-session-journal-types'
@@ -565,6 +566,73 @@ describe('a structured codex session over agentSession.*', () => {
         )
     ).toBe(false)
   })
+
+  it('joins an acquired attach through journal bind before draining final rows', async () => {
+    const host = await ensureStructuredAgentSessionHost({
+      stateDirectory: root,
+      hostId: 'local',
+      claimKeyId: 'key-1',
+      resolveWorkspacePath: async (workspaceId) => `/repos/${workspaceId}`,
+      resolveCodexCommand: () => '/usr/local/bin/codex',
+      openCodexConnection: codex.openConnection
+    })
+    const adapter = (host as unknown as { deps: { adapter: CodexStructuredSessionAdapter } }).deps
+      .adapter
+    const historyEntered = Promise.withResolvers<void>()
+    const historyGate = Promise.withResolvers<void>()
+    const originalHistoryFilePath = adapter.historyFilePath.bind(adapter)
+    vi.spyOn(adapter, 'historyFilePath').mockImplementation(async (input) => {
+      historyEntered.resolve()
+      await historyGate.promise
+      return originalHistoryFilePath(input)
+    })
+
+    const creating = ok<{ fence: number }>('agentSession.create', createIntentParams())
+    await historyEntered.promise
+    codex.notify('turn/started', { threadId: THREAD, turn: { id: TURN } })
+    codex.notify('item/started', {
+      threadId: THREAD,
+      turnId: TURN,
+      item: { type: 'agentMessage', id: 'item-bind-window', text: '' }
+    })
+    codex.notify('item/agentMessage/delta', {
+      threadId: THREAD,
+      turnId: TURN,
+      itemId: 'item-bind-window',
+      delta: 'Buffered while the journal opens.'
+    })
+
+    let stopped = false
+    const stopping = stopStructuredAgentSessionRuntime().then(() => {
+      stopped = true
+    })
+    await new Promise<void>((resolve) => setImmediate(resolve))
+    const waitedForJournalBind = !stopped
+    historyGate.resolve()
+    await creating
+    await stopping
+    expect(waitedForJournalBind).toBe(true)
+
+    const identity = {
+      sessionId: SESSION,
+      workspaceId: WORKSPACE,
+      hostId: 'local',
+      agent: 'codex' as const,
+      providerHandle: { kind: 'codex' as const, threadId: THREAD }
+    }
+    const reopened = await openAgentSessionJournal({
+      identity,
+      journalDir: journalDirectoryFor(root, identity)
+    })
+    expect(reopened.snapshot().items.map(textOf)).toContain('Buffered while the journal opens.')
+    expect(
+      reopened
+        .snapshot()
+        .items.some(
+          (item) => item.body?.kind === 'status' && item.body.turnLifecycle?.state === 'running'
+        )
+    ).toBe(false)
+  })
 })
 
 /** Every item the subscription has published, latest revision per id. */