fix(session-title): retain fallback requests during activation
This commit is contained in:
1 parent
feff25e8f7
commit
ab7bd24b60
5 files changed
+213
-25
No files matched your search
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -88,6 +88,8 @@ kind: "package-reference"
|
||||
|
||||
### 生命周期与并发
|
||||
|
||||
激活期间收到的消息会等待服务就绪,再请求内置回退标题。提供方在服务就绪后注册;此前的消息和请求不会追溯触发提供方。激活失败、服务卸载或 Session 移除都会阻止延后的回退标题写入。
|
||||
|
||||
每个会话的工作状态维护一个修订计数器、一个进行中的回退,以及待处理与活跃的提供方工作。较新的用户消息、提供方 dispose(资源释放)、会话 dispose 或显式刷新都会通过 `AbortController` 中止旧工作;提供方、修订、会话或信号已陈旧的完成结果无法追加。显式刷新会在提供方工作之前预留修订号;重叠的自动/显式回退请求共享一个会话本地正在进行的追加操作。服务拆卸会取消排队工作,并在卸载完成前等待不响应取消的调用结算。
|
||||
|
||||
### 规范化
|
||||
|
||||
@@ -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<SessionEvent, { type: 'user/message' }>): void {
|
||||
if (!this.serviceActive()) return
|
||||
/** Schedule current-provider work; activation-time messages only request the fallback. */
|
||||
private async onUserMessage(session: Session, event: Extract<SessionEvent, { type: 'user/message' }>): Promise<void> {
|
||||
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
|
||||
|
||||
@@ -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<undefined>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
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<undefined>()
|
||||
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')
|
||||
})
|
||||
})
|
||||
Reference in new issue
Block a user