diff --git a/packages/core/session/src/index.ts b/packages/core/session/src/index.ts index d210351e36..e84b438b0f 100644 --- a/packages/core/session/src/index.ts +++ b/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, [ - carrier, - 'session/flushed', - ...flushedArgs, - ]) - invokeContainedSessionObservers( - this.ctx, - 'session/flushed', - session.id, - flushedArgs, - observers, - ) + try { + const observers = collectSessionCallbacks(this.ctx, [ + carrier, + 'session/flushed', + ...flushedArgs, + ]) + invokeContainedSessionObservers( + this.ctx, + 'session/flushed', + session.id, + flushedArgs, + observers, + ) + } catch (error: unknown) { + this.ctx.logger.warn(`session "${session.id}": session/flushed dispatch threw: ${String(error)}`) + } } return durable } diff --git a/packages/core/session/tests/scoped.spec.ts b/packages/core/session/tests/scoped.spec.ts index 45653da3cd..d9ea95850e 100644 --- a/packages/core/session/tests/scoped.spec.ts +++ b/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 { @@ -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()