fix(session): contain checkpoint dispatch resolution

packages/core/session/src/index.ts

@@ -1071,18 +1071,22 @@ export class SessionStore extends Service {
     const durable = results.some(result => result.status === 'fulfilled' && result.value === true)
     if (durable) {
       const flushedArgs: unknown[] = [session, throughSeq]
-      const observers = collectSessionCallbacks(this.ctx, [
+      try {
-        carrier,
+        const observers = collectSessionCallbacks(this.ctx, [
-        'session/flushed',
+          carrier,
-        ...flushedArgs,
+          'session/flushed',
-      ])
+          ...flushedArgs,
-      invokeContainedSessionObservers(
+        ])
-        this.ctx,
+        invokeContainedSessionObservers(
-        'session/flushed',
+          this.ctx,
-        session.id,
+          'session/flushed',
-        flushedArgs,
+          session.id,
-        observers,
+          flushedArgs,
-      )
+          observers,
+        )
+      } catch (error: unknown) {
+        this.ctx.logger.warn(`session "${session.id}": session/flushed dispatch threw: ${String(error)}`)
+      }
     }
     return durable
   }

packages/core/session/tests/scoped.spec.ts

@@ -2,7 +2,7 @@ import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
 import { createScope, scopeOf } from '@deepseek-ai/dsh-scope'
 import type { Scope, ScopeKey } from '@deepseek-ai/dsh-scope'
-import SessionStore from '@deepseek-ai/dsh-session'
+import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import type { Session } from '@deepseek-ai/dsh-session'
 
 async function mount(): Promise<Context> {
@@ -221,6 +221,25 @@ describe('sessions.flush()', () => {
     expect(checkpoints).toEqual([0])
   })
 
+  it('contains successful-checkpoint dispatch resolution failure without reversing the barrier', async () => {
+    const ctx = await mount()
+    const warnings: string[] = []
+    ctx.logger.warn = ((message: unknown) => { warnings.push(String(message)) }) as typeof ctx.logger.warn
+    const checkpoints: number[] = []
+    ctx.on('session/flush', () => true)
+    ctx.on('internal/dispatch', (_mode, name) => {
+      if (name === 'session/flushed') throw new Error('flushed dispatch instrumentation')
+    })
+    ctx.on('session/flushed', (_session, throughSeq) => { checkpoints.push(throughSeq) })
+    const session = ctx.sessions.create(SessionId('flushed-dispatch'))
+
+    await expect(ctx.sessions.flush(session)).resolves.toBe(true)
+    expect(checkpoints).toEqual([])
+    expect(warnings).toEqual([
+      'session "flushed-dispatch": session/flushed dispatch threw: Error: flushed dispatch instrumentation',
+    ])
+  })
+
   it('may publish overlapping checkpoints out of order without widening either boundary', async () => {
     const ctx = await mount()
     const firstGate = Promise.withResolvers<undefined>()