From ab7bd24b60bec261f5e7e43df08f835095f2e1a3 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Mon, 19 Oct 2026 15:26:29 +0800 Subject: [PATCH] fix(session-title): retain fallback requests during activation --- .../session/session-title/README.i18n.yaml | 4 +- packages/session/session-title/README.md | 2 + packages/session/session-title/README.zh.md | 2 + packages/session/session-title/src/index.ts | 51 ++--- .../tests/initialization.spec.ts | 179 ++++++++++++++++++ 5 files changed, 213 insertions(+), 25 deletions(-) create mode 100644 packages/session/session-title/tests/initialization.spec.ts diff --git a/packages/session/session-title/README.i18n.yaml b/packages/session/session-title/README.i18n.yaml index 762f1ba148..1785782cb1 100644 --- a/packages/session/session-title/README.i18n.yaml +++ b/packages/session/session-title/README.i18n.yaml @@ -42,8 +42,8 @@ en: 84f4a3ff08caf847 zh: 9ed3acae2bb2a8b0 /deepseek-ai-dsh-session-title/understand-the-implementation/lifecycle-and-concurrency: - en: 0167ad2951e5ae4d - zh: 0b1df4b7f20eb2f5 + en: 7a4aeb28bde70096 + zh: 6803a7c3054e80a1 /deepseek-ai-dsh-session-title/understand-the-implementation/normalization: en: 0fa22531683cc824 zh: c6379873eb809e76 diff --git a/packages/session/session-title/README.md b/packages/session/session-title/README.md index 83b0736619..fb91354ec7 100644 --- a/packages/session/session-title/README.md +++ b/packages/session/session-title/README.md @@ -88,6 +88,8 @@ Titles are durable, log-only state: every accepted revision is a `session/title` ### Lifecycle and concurrency +Messages received during activation wait for the service to become ready and request the built-in fallback. Providers register after service readiness; earlier messages and requests do not trigger them retroactively. Activation failure, service disposal, or Session removal prevents the deferred fallback from writing. + Per-session work state tracks a revision counter, an in-flight fallback, and pending and active provider work. A newer user message, provider disposal, session disposal, or explicit refresh aborts older work through an `AbortController`; a completion whose provider, revision, session, or signal is stale cannot append. Explicit refreshes reserve their revision before provider work; overlapping automatic and explicit fallback requests share one session-local in-flight append. Service teardown cancels queued work and drains calls that ignore cancellation before unloading completes. ### Normalization diff --git a/packages/session/session-title/README.zh.md b/packages/session/session-title/README.zh.md index 5e046468f5..f42174730b 100644 --- a/packages/session/session-title/README.zh.md +++ b/packages/session/session-title/README.zh.md @@ -88,6 +88,8 @@ kind: "package-reference" ### 生命周期与并发 +激活期间收到的消息会等待服务就绪，再请求内置回退标题。提供方在服务就绪后注册；此前的消息和请求不会追溯触发提供方。激活失败、服务卸载或 Session 移除都会阻止延后的回退标题写入。 + 每个会话的工作状态维护一个修订计数器、一个进行中的回退，以及待处理与活跃的提供方工作。较新的用户消息、提供方 dispose（资源释放）、会话 dispose 或显式刷新都会通过 `AbortController` 中止旧工作；提供方、修订、会话或信号已陈旧的完成结果无法追加。显式刷新会在提供方工作之前预留修订号；重叠的自动／显式回退请求共享一个会话本地正在进行的追加操作。服务拆卸会取消排队工作，并在卸载完成前等待不响应取消的调用结算。 ### 规范化 diff --git a/packages/session/session-title/src/index.ts b/packages/session/session-title/src/index.ts index b2d43669d1..86cfce0058 100644 --- a/packages/session/session-title/src/index.ts +++ b/packages/session/session-title/src/index.ts @@ -345,29 +345,11 @@ export class SessionTitleService extends Service { this.work.clear() }, 'sessionTitle lifecycle') - await ctx.sessionProjections.register(titleProjectionDefinition) - - this.titleInputProjectionHandle = await ctx.sessionProjections.register<'titleInput', TitleInputState>({ - key: 'titleInput', - stateVersion: 3, - stateSchema: titleInputStateSchema, - init: () => EMPTY_TITLE_INPUT, - apply: (state, event) => { - const message = sessionTitleUserMessageOf(event) - if (message === undefined) return state - return { - first: state.first ?? message, - count: state.count + 1, - lastSeq: message.seq, - } - }, - }) - + // oxlint-disable-next-line typescript/no-misused-promises -- session/event reports rejected observer promises without awaiting them. ctx.on('session/event', (session, event) => { switch (event.type) { case 'user/message': - this.onUserMessage(session, event) - break + return this.onUserMessage(session, event) case 'request/header': this.onRequestHeader(session, event) break @@ -385,6 +367,24 @@ export class SessionTitleService extends Service { state.active?.controller.abort(new Error('session disposed during title generation')) this.work.delete(session) }) + + await ctx.sessionProjections.register(titleProjectionDefinition) + + this.titleInputProjectionHandle = await ctx.sessionProjections.register<'titleInput', TitleInputState>({ + key: 'titleInput', + stateVersion: 3, + stateSchema: titleInputStateSchema, + init: () => EMPTY_TITLE_INPUT, + apply: (state, event) => { + const message = sessionTitleUserMessageOf(event) + if (message === undefined) return state + return { + first: state.first ?? message, + count: state.count + 1, + lastSeq: message.seq, + } + }, + }) } /** @@ -506,10 +506,15 @@ export class SessionTitleService extends Service { return dispose } - /** Schedule fallback creation and any provider cadence for one eligible event. */ - private onUserMessage(session: Session, event: Extract): void { - if (!this.serviceActive()) return + /** Schedule current-provider work; activation-time messages only request the fallback. */ + private async onUserMessage(session: Session, event: Extract): Promise { if (event.data.source.kind !== 'user' || sessionTitleUserMessageOf(event) === undefined) return + if (this.ownerFiber.state === FiberState.LOADING) { + await this.ownerFiber.await() + if (this.serviceActive() && this.ctx.sessions.get(session.id) === session) await this.ensureFallback(session) + return + } + if (!this.serviceActive()) return // A user rename pins the title: no automatic revision may override it. if (this.get(session)?.source.kind === 'user') return const registration = this.registration diff --git a/packages/session/session-title/tests/initialization.spec.ts b/packages/session/session-title/tests/initialization.spec.ts new file mode 100644 index 0000000000..4bb8181799 --- /dev/null +++ b/packages/session/session-title/tests/initialization.spec.ts @@ -0,0 +1,179 @@ +/** Live title input remains owned while projection registration is pending. */ +import { Context, FiberState } from '@deepseek-ai/cordis' +import AgentRegistry from '@deepseek-ai/dsh-agent' +import LlmRuntime, { createUserMessage, markAgentLoopRequest } from '@deepseek-ai/dsh-llm' +import SessionStore, { type Session } from '@deepseek-ai/dsh-session' +import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection' +import SessionTitleService, { SessionTitleProviderId, type SessionTitleProviderRequest } from '@deepseek-ai/dsh-session-title' +import { deepFreeze } from '@deepseek-ai/dsh-util-values' +import { describe, expect, it, onTestFinished, vi } from 'vitest' + +const CONFIG = { fallbackMaxWords: 8, fallbackMaxBytes: 64, maxTitleBytes: 256 } + +async function setup(key: 'title' | 'titleInput') { + const ctx = new Context() + const entered = Promise.withResolvers() + const release = Promise.withResolvers() + onTestFinished(async () => { + release.resolve(undefined) + await ctx.fiber.dispose() + }) + await ctx.plugin(SessionStore) + await ctx.plugin(SessionProjectionRegistry) + await ctx.plugin(AgentRegistry) + await ctx.plugin(LlmRuntime) + const register = ctx.sessionProjections.register.bind(ctx.sessionProjections) + const spy = vi.spyOn(ctx.sessionProjections, 'register').mockImplementation(async (definition) => { + const handle = await register(definition) + if (definition.key === key) { + entered.resolve(undefined) + await release.promise + } + return handle + }) + onTestFinished(() => { spy.mockRestore() }) + return { ctx, entered: entered.promise, release } +} + +function message(session: Session, text: string) { + return session.append('user/message', createUserMessage({ + content: [{ type: 'text', text }], source: { kind: 'user' }, + }), { surfaceOp: 'append' }) +} + +function titleWritten(ctx: Context, session: Session, kind: 'fallback' | 'provider') { + const written = Promise.withResolvers() + ctx.on('session/event', (target, event) => { + if (target === session && event.type === 'session/title' && event.data.source.kind === kind) written.resolve(undefined) + }) + return written.promise +} + +describe('SessionTitleService activation', () => { + it.each(['title', 'titleInput'] as const)('retains the first message while %s registration is pending', async (key) => { + const { ctx, entered, release } = await setup(key) + const fiber = ctx.plugin(SessionTitleService, CONFIG) + await entered + expect(ctx.get('sessionTitle')).toBeUndefined() + const session = ctx.sessions.create() + const written = titleWritten(ctx, session, 'fallback') + const first = message(session, 'First message during activation') + message(session, 'Second message during activation') + + release.resolve(undefined) + await fiber + await written + + expect(ctx.sessionTitle.get(session)).toMatchObject({ + title: 'First message during activation', messageSeqs: [first.seq], source: { kind: 'fallback' }, + }) + expect(session.snapshotEvents().filter(event => event.type === 'session/title')).toHaveLength(1) + }) + + it('uses the fallback for input received before a provider registers', async () => { + const { ctx, entered, release } = await setup('titleInput') + const requests: SessionTitleProviderRequest[] = [] + const provider = ctx.inject(['sessionTitle'], (owner) => { + owner.sessionTitle.register({ + id: SessionTitleProviderId('activation'), automatic: 'all-prompts', + async generate(request) { + requests.push(request) + return { title: 'Generated after activation', messageSeqs: request.messages.map(item => item.seq) } + }, + }) + }) + const fiber = ctx.plugin(SessionTitleService, CONFIG) + await entered + const session = ctx.sessions.create() + session.append('turn/start', { turn: 1 }) + session.append('step/start', { turn: 1, step: 1 }) + const fallbackWritten = titleWritten(ctx, session, 'fallback') + const first = message(session, 'Prompt before provider registration') + const route = { provider: 'main-provider', model: 'main-model' } + session.append('request/header', { header: { config: route }, reason: 'initial' }) + void ctx.llm.stream(markAgentLoopRequest(deepFreeze({ ...route, sessionId: session.id, messages: session.deriveMessages() }))) + + release.resolve(undefined) + await fiber + await provider + await fallbackWritten + + expect(requests).toEqual([]) + expect(ctx.sessionTitle.get(session)).toMatchObject({ messageSeqs: [first.seq], source: { kind: 'fallback' } }) + + const providerWritten = titleWritten(ctx, session, 'provider') + const later = message(session, 'Prompt after provider registration') + void ctx.llm.stream(markAgentLoopRequest(deepFreeze({ ...route, sessionId: session.id, messages: session.deriveMessages() }))) + await providerWritten + + expect(requests).toHaveLength(1) + expect(requests[0]).toMatchObject({ route, messages: [ + { seq: first.seq, text: 'Prompt before provider registration' }, + { seq: later.seq, text: 'Prompt after provider registration' }, + ] }) + }) + + it('does not title a Session disposed before activation', async () => { + const { ctx, entered, release } = await setup('title') + const fiber = ctx.plugin(SessionTitleService, CONFIG) + await entered + const removed = ctx.sessions.prepare() + const detach = ctx.sessions.enter(removed) + ctx.sessions.announce(removed) + message(removed, 'Removed during activation') + detach() + const live = ctx.sessions.create() + const written = titleWritten(ctx, live, 'fallback') + message(live, 'Still live') + + release.resolve(undefined) + await fiber + await written + + expect(ctx.sessionTitle.get(removed)).toBeUndefined() + expect(ctx.sessionTitle.get(live)?.title).toBe('Still live') + }) + + it.each(['dispose', 'reject'] as const)('does not write a title after activation %s', async (ending) => { + const { ctx, entered, release } = await setup('titleInput') + const fiber = ctx.plugin(SessionTitleService, CONFIG) + await entered + const session = ctx.sessions.create() + message(session, 'Never commit this title') + if (ending === 'dispose') { + const disposal = fiber.dispose() + release.resolve(undefined) + await disposal + } else { + const failed = expect(Promise.resolve(fiber)).rejects.toThrow('registration failed') + release.reject(new Error('registration failed')) + await failed + } + + expect(session.snapshotEvents().filter(event => event.type === 'session/title')).toEqual([]) + expect(ctx.get('sessionTitle')).toBeUndefined() + }) + + it('reports one deferred fallback failure and still titles another Session', async () => { + const { ctx, entered, release } = await setup('titleInput') + const warn = vi.spyOn(ctx.logger, 'warn') + onTestFinished(() => { warn.mockRestore() }) + ctx.on('internal/status', (fiber) => { + if (fiber.runtime?.callback !== SessionTitleService || fiber.state !== FiberState.ACTIVE) return + const get = vi.spyOn(ctx.sessionTitle, 'get').mockImplementationOnce(() => { throw new Error('title read failed') }) + onTestFinished(() => { get.mockRestore() }) + }) + const fiber = ctx.plugin(SessionTitleService, CONFIG) + await entered + message(ctx.sessions.create(), 'Failed title read') + const live = ctx.sessions.create() + const written = titleWritten(ctx, live, 'fallback') + message(live, 'Unaffected title') + release.resolve(undefined) + await fiber + await written + + expect(warn).toHaveBeenCalledWith(expect.stringContaining('session/event listener rejected: Error: title read failed')) + expect(ctx.sessionTitle.get(live)?.title).toBe('Unaffected title') + }) +})