fix(session): contain post-commit observers

packages/ui/acp/tests/dispose.spec.ts

@@ -160,7 +160,7 @@ describe('acp bridge — disposal & HMR safety', () => {
     // The teardown-ORDER guarantee: a per-agent dispose must stop the loop,
     // AWAIT its exit (so the loop's final `turn/end` + `session/flush` fire
     // through the still-attached store observer → `session/event`), and only
-    // THEN detach that observer + remove the session. If the order were inverted
+    // THEN remove its publication hooks and session entry. If the order were inverted
     // (detach first), the closing events would never reach persistence. Drive a
     // CLEAN turn to completion, dispose JUST the bridge, then re-load the
     // persisted log from disk and assert the closing turn/end is on disk — the
@@ -190,7 +190,7 @@ describe('acp bridge — disposal & HMR safety', () => {
     // produced BY the dispose itself. Here the model stream HANGS, so the turn is
     // still open when teardown runs: the composite agent effect stops the loop,
     // the loop unwinds and appends `turn/end {disposed}` + runs its final
-    // `session/flush` — all while the store-owned append observer is still attached (the session
+    // `session/flush` — all while the store-owned publication hooks are still attached (the session
     // detach is the LAST disposer in the same effect's LIFO chain) — and only
     // THEN is the session detached. If the order were inverted (or the session
     // were a racing SIBLING effect), the abort-produced `turn/end` would never
@@ -255,7 +255,7 @@ describe('acp bridge — disposal & HMR safety', () => {
     // into ONE composite effect whose disposers run as a `.then()` chain. The
     // register disposer emits `agent/disposed`; if a listener throws and the
     // emit is UNCONTAINED, the rejected chain skips the LATER session-detach
-    // disposer — stranding the session in the store with its append observer attached (a
+    // disposer — stranding the session in the store with its publication hooks attached (a
     // leak AND a durability hole, since the new design relies on detach
     // running). The emit must be contained. Register a throwing listener, drive
     // a clean turn, dispose, and assert the session was STILL removed.

packages/ui/acp/tests/turns.spec.ts

@@ -3,7 +3,7 @@ import { mkdtemp, rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { defineTool } from '@deepseek-ai/dsh-tools'
-import { AgentId } from '@deepseek-ai/dsh-agent'
+import { AgentId, agentEvents } from '@deepseek-ai/dsh-agent'
 import { PROTOCOL_VERSION } from '@agentclientprotocol/sdk'
 import {
   errorResponse,
@@ -235,12 +235,9 @@ describe('acp bridge — turn outcomes', () => {
     expect(failed).toHaveLength(1)
   })
 
-  it('settles via the log fallback when a prior session/event listener throws (starvation)', async () => {
-    // A peer session/event listener that runs BEFORE the bridge's listener
-    // throws on turn/end (prepend: true puts it first). cordis emit stops at the
-    // throw, so the bridge's session/event listener never sees turn/end and
-    // cannot settle there. The agent/status idle-fallback must reconcile the
-    // prompt from the log so the RPC settles instead of hanging.
+  it('settles successfully when an earlier turn/end observer throws', async () => {
+    // Session contains each post-commit observer failure, so a prepended peer
+    // cannot starve the bridge's live turn/end delivery.
     harness = await makeBridgeHarness({ storageDir, script: [textResponse('answer')] })
     harness.ctx.on('session/event', (_s, event) => {
       if (event.type === 'turn/end') throw new Error('peer listener boom')
@@ -250,9 +247,7 @@ describe('acp bridge — turn outcomes', () => {
     expect(res.stopReason).toBe('end_turn')
   })
 
-  it('log fallback REJECTS when the starved turn ended in error', async () => {
-    // Same starvation as above, but the turn fails: the idle-fallback must
-    // reject the RPC from the logged turn/end{error}, not resolve.
+  it('still rejects a failed turn when an earlier turn/end observer throws', async () => {
     harness = await makeBridgeHarness({ storageDir, script: [errorResponse('starved boom')] })
     harness.ctx.on('session/event', (_s, event) => {
       if (event.type === 'turn/end') throw new Error('peer listener boom')
@@ -262,21 +257,49 @@ describe('acp bridge — turn outcomes', () => {
       .rejects.toThrow(/turn failed: starved boom/)
   })
 
-  it('log fallback infers the owning turn when turn/START capture is starved', async () => {
-    // A peer listener throws on turn/START (not turn/end): the bridge never
+  it('captures and settles the owning turn when an earlier turn-start observer throws', async () => {
+    // Turn correlation still reaches the bridge after the throwing peer and
     // captures inflight.turn via the live stream. A throwing turn/start listener
-    // also FAILS the turn (the throw is recorded as the turn's error). Without
-    // the watermark inference the fallback would resolve `cancelled` (the bug);
-    // with it, it infers the owning turn from the log and REJECTS from that
-    // turn's error turn/end. (The model's own error is never reached — the turn
-    // failed at start — so the rejection carries the listener's failure.)
-    harness = await makeBridgeHarness({ storageDir, script: [textResponse('never runs')] })
+    // Session contains post-commit callbacks independently.
+    // The model request and normal turn outcome therefore still occur.
+    harness = await makeBridgeHarness({ storageDir, script: [textResponse('answer')] })
     harness.ctx.on('session/event', (_s, event) => {
       if (event.type === 'turn/start') throw new Error('peer listener boom on start')
     }, { prepend: true })
     const sessionId = await newSession(harness)
-    await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }))
-      .rejects.toThrow(/turn failed:/)
+    const result = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
+    expect(result.stopReason).toBe('end_turn')
+  })
+
+  it('status reconciliation infers the owning message turn when teardown wins after turn/start', async () => {
+    harness = await makeBridgeHarness({ storageDir, script: [textResponse('background completion')] })
+    const sessionId = await newSession(harness)
+    const agent = harness.ctx.agents.get(AgentId(sessionId))!
+    harness.ctx.on('session/event', (session, event) => {
+      if (session !== agent.session || event.type !== 'turn/start') return
+      // Inject the signal ordering the defensive fallback handles: disposal
+      // status after turn/start commits but before ACP's later live observer.
+      // This is event-level simulation; it does not mutate the test agent.
+      agentEvents(harness!.ctx, agent).emit('agent/status', 'disposed')
+    }, { prepend: true })
+
+    const result = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
+    expect(result.stopReason).toBe('cancelled')
+  })
+
+  it('status reconciliation can settle from a committed turn/end before live delivery', async () => {
+    harness = await makeBridgeHarness({ storageDir, script: [textResponse('answer')] })
+    const sessionId = await newSession(harness)
+    const agent = harness.ctx.agents.get(AgentId(sessionId))!
+    harness.ctx.on('session/event', (session, event) => {
+      if (session !== agent.session || event.type !== 'turn/end') return
+      // Inject a reentrant status signal after the boundary commits to exercise
+      // the defensive log path before ACP's captured callback runs.
+      agentEvents(harness!.ctx, agent).emit('agent/status', 'idle')
+    }, { prepend: true })
+
+    const result = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
+    expect(result.stopReason).toBe('end_turn')
   })
 
   it('a between-turn injection does not settle the prompt early (message-trigger correlation)', async () => {