index.ts 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835
  1. /**
  2. * Log-backed session title service, deterministic fallback, and provider contract.
  3. * @module @deepseek-ai/dsh-session-title
  4. */
  5. import { Context, FiberState, Service, type Fiber } from '@deepseek-ai/cordis'
  6. import z from '@deepseek-ai/schemastery'
  7. import { z as zod } from 'zod'
  8. import type { ZodType } from 'zod'
  9. import type { Branded } from '@deepseek-ai/dsh-brand'
  10. import { isAgentLoopRequest } from '@deepseek-ai/dsh-llm'
  11. import type { GenerateOptions } from '@deepseek-ai/dsh-llm'
  12. import { assertNever, deepFreeze } from '@deepseek-ai/dsh-util-values'
  13. import type {
  14. Session,
  15. SessionEvent,
  16. } from '@deepseek-ai/dsh-session'
  17. import { SessionSeq } from '@deepseek-ai/dsh-session'
  18. import type {} from '@deepseek-ai/dsh-session-projection'
  19. import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
  20. import type {} from '@deepseek-ai/dsh-agent'
  21. export type {
  22. SessionTitleEventData,
  23. SessionTitleModelIdentity,
  24. SessionTitleSnapshot,
  25. SessionTitleSource,
  26. SessionTitleUserMessage,
  27. TitleProjection,
  28. } from './types.ts'
  29. import { fallbackSessionTitle, normalizeSessionTitle } from './normalize.ts'
  30. import type {
  31. SessionTitleEventData,
  32. SessionTitleModelIdentity,
  33. SessionTitleSnapshot,
  34. SessionTitleSource,
  35. SessionTitleUserMessage,
  36. TitleInputState,
  37. TitleProjection,
  38. } from './types.ts'
  39. /** Identifies one session-title provider registration. */
  40. export type SessionTitleProviderId = Branded<'SessionTitleProviderId'>
  41. /**
  42. * Brand a raw provider id.
  43. * @param id - stable non-empty provider identifier supplied by a plugin.
  44. * @returns the same string with the session-title provider brand.
  45. */
  46. export function SessionTitleProviderId(id: string): SessionTitleProviderId {
  47. return id as SessionTitleProviderId
  48. }
  49. export { fallbackSessionTitle, normalizeSessionTitle, truncateTitleUtf8 } from './normalize.ts'
  50. /** Required deterministic fallback and accepted-title limits. */
  51. export interface Config {
  52. /** Maximum whitespace-delimited words in the built-in fallback. */
  53. readonly fallbackMaxWords: number
  54. /** Maximum UTF-8 bytes in the built-in fallback. */
  55. readonly fallbackMaxBytes: number
  56. /** Maximum UTF-8 bytes in any accepted title. */
  57. readonly maxTitleBytes: number
  58. }
  59. declare module '@deepseek-ai/cordis' {
  60. interface Context {
  61. sessionTitle: SessionTitleService
  62. }
  63. }
  64. declare module '@deepseek-ai/dsh-session/types' {
  65. interface SessionEventMap {
  66. /**
  67. * Latest-wins session title snapshot. Log-only: it never enters the model
  68. * surface or derived history.
  69. */
  70. 'session/title': SessionTitleEventData
  71. }
  72. }
  73. /**
  74. * Rejection of an explicit user title whose text normalizes to empty — the
  75. * one {@link SessionTitleService.rename} failure that blames the input.
  76. * Callers translating rename failures onto a wire (`title-invalid`) narrow on
  77. * this class; liveness and disposal failures stay plain `Error`s.
  78. */
  79. export class SessionTitleInvalidError extends Error {
  80. override readonly name = 'SessionTitleInvalidError'
  81. }
  82. /** Automatic generation cadence owned by a registered provider. */
  83. export type SessionTitleAutomaticMode = 'first-prompt' | 'all-prompts'
  84. /** Immutable input supplied to one title-provider call. */
  85. export interface SessionTitleProviderRequest {
  86. /** Live session being titled. */
  87. readonly session: Session
  88. /** All eligible human messages through this generation revision. */
  89. readonly messages: readonly SessionTitleUserMessage[]
  90. /** Exact current logged main-request route, when one has been recorded. */
  91. readonly route?: SessionTitleModelIdentity
  92. /** Cancellation for supersession, disposal, timeout composition, or the explicit caller. */
  93. readonly signal: AbortSignal
  94. }
  95. /** Provider output before service-owned normalization and log acceptance. */
  96. export interface SessionTitleProviderResult {
  97. /** Proposed title text. */
  98. readonly title: string
  99. /** Exact seqs from `request.messages` used by this result. */
  100. readonly messageSeqs: readonly SessionSeq[]
  101. /** Auxiliary LLM route, when generation used a model. */
  102. readonly model?: SessionTitleModelIdentity
  103. }
  104. /** One optional asynchronous title implementation registered with the service. */
  105. export interface SessionTitleProvider {
  106. /** Stable id of the provider recorded with the title. */
  107. readonly id: SessionTitleProviderId
  108. /** When new human prompts start automatic generation. */
  109. readonly automatic: SessionTitleAutomaticMode
  110. /**
  111. * Produce one title revision.
  112. * @param request - message snapshot, current route, session, and cancellation.
  113. * @returns proposed title plus exact input seqs and the optional provider/model route used to generate it.
  114. */
  115. generate(request: SessionTitleProviderRequest): Promise<SessionTitleProviderResult>
  116. }
  117. /** Extract one eligible human text message from a session event. */
  118. function sessionTitleUserMessageOf(event: SessionEvent): SessionTitleUserMessage | undefined {
  119. if (event.type !== 'user/message' || event.data.source.kind !== 'user') return undefined
  120. const content = event.data.content
  121. const text = content
  122. .filter((block): block is Extract<(typeof content)[number], { type: 'text' }> => block.type === 'text')
  123. .map(block => block.text)
  124. .join('\n')
  125. if (normalizeSessionTitle(text, Number.MAX_SAFE_INTEGER).length === 0) return undefined
  126. return { seq: event.seq, text }
  127. }
  128. /** Defensive copy of a logged title source (the snapshot must not alias log-owned objects). */
  129. function copySessionTitleSource(source: SessionTitleSource): SessionTitleSource {
  130. switch (source.kind) {
  131. case 'fallback': return { kind: 'fallback' }
  132. case 'provider': return {
  133. kind: 'provider',
  134. provider: source.provider,
  135. ...(source.model === undefined ? {} : { model: { ...source.model } }),
  136. }
  137. case 'user': return { kind: 'user' }
  138. /* v8 ignore next -- closed-union exhaustiveness guard */
  139. default: return assertNever(source, 'SessionTitleSource')
  140. }
  141. }
  142. /** Service-owned resolved limits. */
  143. interface ResolvedConfig {
  144. readonly fallbackMaxWords: number
  145. readonly fallbackMaxBytes: number
  146. readonly maxTitleBytes: number
  147. }
  148. /** One exact provider registration generation. */
  149. interface ProviderRegistration {
  150. readonly provider: SessionTitleProvider
  151. readonly active: Set<Promise<unknown>>
  152. closing: boolean
  153. }
  154. /** Automatic work waiting for the matching main-request header. */
  155. interface PendingAutomaticWork {
  156. readonly registration: ProviderRegistration
  157. readonly revision: number
  158. readonly throughSeq: SessionSeq
  159. }
  160. /** Provider call currently allowed to commit for one session. */
  161. interface ActiveProviderWork extends PendingAutomaticWork {
  162. readonly controller: AbortController
  163. readonly signal: AbortSignal
  164. }
  165. /** Mutable concurrency state scoped to one live session. */
  166. interface SessionTitleWorkState {
  167. revision: number
  168. fallback?: Promise<SessionTitleSnapshot | undefined>
  169. pending?: PendingAutomaticWork
  170. active?: ActiveProviderWork
  171. }
  172. /** Validate one positive integer configuration field. */
  173. function assertPositiveInteger(name: keyof Config, value: number): void {
  174. if (!Number.isInteger(value) || value <= 0) {
  175. throw new Error(`session-title: ${name} must be a positive integer`)
  176. }
  177. }
  178. /**
  179. * Convert title projection state into an immutable snapshot.
  180. * @param state - the title unit's folded state.
  181. * @returns the immutable snapshot.
  182. */
  183. function titleSnapshotFromState(state: TitleProjection): SessionTitleSnapshot {
  184. return deepFreeze({
  185. title: state.title,
  186. messageSeqs: [...state.messageSeqs],
  187. source: copySessionTitleSource(state.source),
  188. eventSeq: state.eventSeq,
  189. updatedAt: state.updatedAt,
  190. })
  191. }
  192. const EMPTY_TITLE_INPUT: TitleInputState = { first: null, count: 0, lastSeq: null }
  193. const sessionTitleUserMessageSchema: ZodType<SessionTitleUserMessage> = zod.object({
  194. seq: zod.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER).transform(SessionSeq),
  195. text: zod.string(),
  196. }).strict()
  197. const titleInputStateSchema: ZodType<TitleInputState> = zod.object({
  198. first: sessionTitleUserMessageSchema.nullable(),
  199. count: zod.number().int().nonnegative(),
  200. lastSeq: zod.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER).transform(SessionSeq).nullable(),
  201. }).strict().superRefine((state, context) => {
  202. const empty = state.first === null && state.lastSeq === null && state.count === 0
  203. const populated = state.first !== null
  204. && state.lastSeq !== null
  205. && state.count > 0
  206. && state.first.seq <= state.lastSeq
  207. if (!empty && !populated) {
  208. context.addIssue({
  209. code: 'custom',
  210. message: 'title input state must pair its count with first and last message seqs',
  211. })
  212. }
  213. })
  214. /**
  215. * Collect eligible human text messages from a session log, in seq order.
  216. * The full eligible prefix is only materialized for one provider generation,
  217. * so it is scanned from the log at execution time rather than retained by
  218. * the O(1) `titleInput` projection.
  219. * @param events - the session event log.
  220. * @param throughSeq - optional inclusive upper seq bound.
  221. * @returns eligible messages with exact source seqs.
  222. */
  223. function collectSessionTitleMessages(
  224. events: readonly SessionEvent[],
  225. throughSeq?: SessionSeq,
  226. ): SessionTitleUserMessage[] {
  227. const messages: SessionTitleUserMessage[] = []
  228. for (const event of events) {
  229. if (throughSeq !== undefined && event.seq > throughSeq) break
  230. const message = sessionTitleUserMessageOf(event)
  231. if (message !== undefined) messages.push(message)
  232. }
  233. return messages
  234. }
  235. const titleViewSchema: ZodType<string | null> = zod.string().min(1).nullable()
  236. /** Latest logged title text and its client view. */
  237. export const titleProjectionDefinition = {
  238. key: 'title',
  239. stateVersion: 1,
  240. stateSchema: titleViewSchema,
  241. init: () => null,
  242. apply: (state, event) => (event.type === 'session/title'
  243. ? event.data.title
  244. : state),
  245. wire: {
  246. viewSchema: titleViewSchema,
  247. view: state => state,
  248. },
  249. } satisfies ProjectionDefinition<'title', string | null>
  250. /**
  251. * Fold the latest logged title without consulting mutable metadata.
  252. * @param events - live or persisted session log.
  253. * @returns the latest immutable title snapshot, or `undefined`.
  254. */
  255. export function foldSessionTitle(events: readonly SessionEvent[]): SessionTitleSnapshot | undefined {
  256. const event = events.findLast(item => item.type === 'session/title')
  257. if (event === undefined) return undefined
  258. return titleSnapshotFromState({
  259. title: event.data.title,
  260. messageSeqs: event.data.messageSeqs,
  261. source: event.data.source,
  262. eventSeq: event.seq,
  263. updatedAt: event.time,
  264. })
  265. }
  266. /** Log-backed title fold plus asynchronous fallback generation. */
  267. export class SessionTitleService extends Service {
  268. static inject = ['sessions', 'sessionProjections']
  269. static Config: z<Config> = z.object({
  270. fallbackMaxWords: z.number().step(1).min(1).required(),
  271. fallbackMaxBytes: z.number().step(1).min(1).required(),
  272. maxTitleBytes: z.number().step(1).min(1).required(),
  273. })
  274. private readonly config: ResolvedConfig
  275. private readonly ownerFiber: Fiber
  276. private registration: ProviderRegistration | undefined
  277. private readonly work = new Map<Session, SessionTitleWorkState>()
  278. private readonly lifetime = new AbortController()
  279. private readonly inFlight = new Set<Promise<unknown>>()
  280. constructor(ctx: Context, config: Config) {
  281. super(ctx, 'sessionTitle')
  282. this.ownerFiber = ctx.fiber
  283. const candidate: unknown = config
  284. if (candidate === null || typeof candidate !== 'object') {
  285. throw new Error('session-title: configuration is required')
  286. }
  287. const value = candidate as Config
  288. assertPositiveInteger('fallbackMaxWords', value.fallbackMaxWords)
  289. assertPositiveInteger('fallbackMaxBytes', value.fallbackMaxBytes)
  290. assertPositiveInteger('maxTitleBytes', value.maxTitleBytes)
  291. if (value.fallbackMaxBytes > value.maxTitleBytes) {
  292. throw new Error('session-title: fallbackMaxBytes must not exceed maxTitleBytes')
  293. }
  294. this.config = deepFreeze({ ...value })
  295. ctx.effect(() => async () => {
  296. this.lifetime.abort(new Error('session-title service disposed'))
  297. if (this.registration !== undefined) this.registration.closing = true
  298. this.registration = undefined
  299. for (const state of this.work.values()) {
  300. delete state.pending
  301. state.active?.controller.abort(new Error('session-title service disposed'))
  302. }
  303. await this.drain(this.inFlight)
  304. this.work.clear()
  305. }, 'sessionTitle lifecycle')
  306. ctx.sessionProjections.register(titleProjectionDefinition)
  307. ctx.sessionProjections.register<'titleInput', TitleInputState>({
  308. key: 'titleInput',
  309. stateVersion: 3,
  310. stateSchema: titleInputStateSchema,
  311. init: () => EMPTY_TITLE_INPUT,
  312. apply: (state, event) => {
  313. const message = sessionTitleUserMessageOf(event)
  314. if (message === undefined) return state
  315. return {
  316. first: state.first ?? message,
  317. count: state.count + 1,
  318. lastSeq: message.seq,
  319. }
  320. },
  321. })
  322. ctx.on('session/event', (session, event) => {
  323. switch (event.type) {
  324. case 'user/message':
  325. this.onUserMessage(session, event)
  326. break
  327. case 'request/header':
  328. this.onRequestHeader(session, event)
  329. break
  330. default:
  331. break
  332. }
  333. })
  334. ctx.on('llm/stream', (options, next) => {
  335. this.onMainRequest(options)
  336. return next()
  337. }, { global: true, prepend: true })
  338. ctx.on('session/disposed', (session) => {
  339. const state = this.work.get(session)
  340. if (state === undefined) return
  341. state.active?.controller.abort(new Error('session disposed during title generation'))
  342. this.work.delete(session)
  343. })
  344. }
  345. /**
  346. * Read the latest folded title from one live or replayed session.
  347. * @param session - session whose log is the title source of truth.
  348. * @returns latest title snapshot, or `undefined` before eligible input.
  349. */
  350. get(session: Session): SessionTitleSnapshot | undefined {
  351. // oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred.
  352. return foldSessionTitle(session.snapshotEvents())
  353. }
  354. /**
  355. * Accept an explicit user title. Appends a `session/title` event with the
  356. * `user` source, which pins the title: in-flight automatic generation is
  357. * superseded and later user messages schedule none (an explicit
  358. * {@link SessionTitleService.refresh} remains the deliberate unpin).
  359. * @param session - exact live session to rename.
  360. * @param title - raw user input; normalized before acceptance.
  361. * @returns the accepted title snapshot.
  362. * @throws {SessionTitleInvalidError} when the title normalizes to empty.
  363. * @throws {Error} when the session is not live or the service is disposed.
  364. */
  365. rename(session: Session, title: string): SessionTitleSnapshot {
  366. this.assertServiceActive()
  367. if (this.ctx.sessions.get(session.id) !== session) {
  368. throw new Error(`session "${session.id}" is not live in this store`)
  369. }
  370. const normalized = normalizeSessionTitle(title, this.config.maxTitleBytes)
  371. if (normalized.length === 0) {
  372. throw new SessionTitleInvalidError('session title must contain visible characters')
  373. }
  374. const state = this.stateFor(session)
  375. this.supersede(state, 'user rename superseded automatic title generation')
  376. session.append('session/title', {
  377. title: normalized,
  378. messageSeqs: [],
  379. source: { kind: 'user' },
  380. })
  381. const snapshot = this.get(session)
  382. /* v8 ignore next -- unreachable: the append above just committed a session/title event. */
  383. if (snapshot === undefined) throw new Error('renamed title failed to fold')
  384. return snapshot
  385. }
  386. /**
  387. * Explicitly retry the registered provider, or materialize the built-in
  388. * fallback when no provider is registered.
  389. * @param session - exact live session to refresh.
  390. * @param signal - optional caller cancellation.
  391. * @returns latest accepted title, or `undefined` when no eligible text exists.
  392. */
  393. async refresh(session: Session, signal?: AbortSignal): Promise<SessionTitleSnapshot | undefined> {
  394. signal?.throwIfAborted()
  395. this.assertServiceActive()
  396. if (this.ctx.sessions.get(session.id) !== session) {
  397. throw new Error(`session "${session.id}" is not live in this store`)
  398. }
  399. const registration = this.registration
  400. const input = this.titleInputOf(session)
  401. if (registration === undefined || registration.closing || input.lastSeq === null) {
  402. // Explicit refresh is the unpin even without a provider: a standing
  403. // user title must not short-circuit ensureFallback into a no-op, so
  404. // re-derive and append the fallback over it when one is derivable.
  405. const current = this.get(session)
  406. const first = input.first
  407. if (current?.source.kind === 'user' && first !== null) {
  408. this.appendFallback(session, first)
  409. signal?.throwIfAborted()
  410. return this.get(session)
  411. }
  412. const fallback = await this.ensureFallback(session)
  413. signal?.throwIfAborted()
  414. return fallback
  415. }
  416. const state = this.stateFor(session)
  417. const revision = this.supersede(state, 'explicit title refresh superseded older generation')
  418. const work = this.activate({
  419. registration,
  420. revision,
  421. throughSeq: input.lastSeq,
  422. }, state, signal)
  423. const config = session.requestHeader()?.config
  424. const route = config === undefined ? undefined : { provider: config.provider, model: config.model }
  425. return this.startProvider(session, work, route)
  426. }
  427. /**
  428. * Register the sole optional title provider. Disposal aborts its pending and
  429. * active work before another provider may register.
  430. * @param provider - provider identity, cadence, and generation function.
  431. * @returns exact Cordis effect disposer, which settles after active calls quiesce.
  432. */
  433. register(provider: SessionTitleProvider): () => Promise<void> {
  434. this.validateProvider(provider)
  435. if (this.registration !== undefined) {
  436. throw new Error(`session-title provider "${this.registration.provider.id}" is already registered`)
  437. }
  438. const registration: ProviderRegistration = {
  439. provider,
  440. active: new Set(),
  441. closing: false,
  442. }
  443. const dispose = this.ctx.effect(function* (this: SessionTitleService) {
  444. this.registration = registration
  445. yield async () => {
  446. registration.closing = true
  447. for (const state of this.work.values()) {
  448. if (state.pending?.registration === registration) delete state.pending
  449. if (state.active?.registration === registration) {
  450. state.active.controller.abort(new Error(`session-title provider "${provider.id}" was disposed`))
  451. }
  452. }
  453. await this.drain(registration.active)
  454. if (this.registration === registration) this.registration = undefined
  455. }
  456. }.bind(this), 'sessionTitle.register()')
  457. return dispose
  458. }
  459. /** Schedule fallback creation and any provider cadence for one eligible event. */
  460. private onUserMessage(session: Session, event: Extract<SessionEvent, { type: 'user/message' }>): void {
  461. if (!this.serviceActive()) return
  462. if (event.data.source.kind !== 'user' || sessionTitleUserMessageOf(event) === undefined) return
  463. // A user rename pins the title: no automatic revision may override it.
  464. if (this.get(session)?.source.kind === 'user') return
  465. const registration = this.registration
  466. if (registration !== undefined && !registration.closing) {
  467. const count = this.titleInputOf(session).count
  468. const shouldSchedule = registration.provider.automatic === 'all-prompts'
  469. || (session.header.parentSession === undefined && count === 1 && this.get(session) === undefined)
  470. if (shouldSchedule) {
  471. const state = this.stateFor(session)
  472. const revision = this.supersede(state, 'newer user message superseded title generation')
  473. state.pending = { registration, revision, throughSeq: event.seq }
  474. }
  475. }
  476. this.defer(async () => {
  477. try {
  478. await this.ensureFallback(session)
  479. } catch (error: unknown) {
  480. if (!this.serviceActive()) return
  481. this.ctx.logger.warn(`session "${session.id}": fallback title update failed: ${String(error)}`)
  482. }
  483. })
  484. }
  485. /** Start pending automatic work only after its exact main-request route is logged. */
  486. private onRequestHeader(session: Session, event: Extract<SessionEvent, { type: 'request/header' }>): void {
  487. if (!this.serviceActive()) return
  488. const state = this.work.get(session)
  489. const pending = state?.pending
  490. if (state === undefined || pending === undefined || pending.throughSeq >= event.seq) return
  491. const route = {
  492. provider: event.data.header.config.provider,
  493. model: event.data.header.config.model,
  494. }
  495. this.startPending(session, state, pending, route)
  496. }
  497. /** Start unchanged-route work from the marked loop request after its header fold is current. */
  498. private onMainRequest(options: GenerateOptions): void {
  499. if (!this.serviceActive() || options.sessionId === undefined || !isAgentLoopRequest(options)) return
  500. const session = this.ctx.sessions.get(options.sessionId)
  501. const state = session === undefined ? undefined : this.work.get(session)
  502. const pending = state?.pending
  503. if (session === undefined || state === undefined || pending === undefined) return
  504. const boundary = this.ctx.sessionProjections.stateOf(session, 'turnBoundary')?.lastStepBoundary
  505. const route = session.requestHeader()?.config
  506. if (boundary?.kind !== 'start'
  507. || boundary.seq <= pending.throughSeq
  508. || route?.provider !== options.provider
  509. || route.model !== options.model) return
  510. this.startPending(session, state, pending, { provider: options.provider, model: options.model })
  511. }
  512. /** Consume one pending revision and schedule its non-blocking provider call. */
  513. private startPending(
  514. session: Session,
  515. state: SessionTitleWorkState,
  516. pending: PendingAutomaticWork,
  517. route: SessionTitleModelIdentity,
  518. ): void {
  519. delete state.pending
  520. this.defer(async () => {
  521. if (this.registration !== pending.registration
  522. || pending.registration.closing
  523. || this.work.get(session) !== state
  524. || state.revision !== pending.revision) return
  525. const work = this.activate(pending, state)
  526. try {
  527. await this.startProvider(session, work, route)
  528. } catch (error: unknown) {
  529. if (work.signal.aborted || !this.serviceActive()) return
  530. this.ctx.logger.warn(`session "${session.id}": automatic title generation failed: ${String(error)}`)
  531. }
  532. })
  533. }
  534. /** Start one tracked provider call after publishing its active revision. */
  535. private startProvider(
  536. session: Session,
  537. work: ActiveProviderWork,
  538. route?: SessionTitleModelIdentity,
  539. ): Promise<SessionTitleSnapshot | undefined> {
  540. const run = Promise.resolve().then(() => this.runProvider(session, work, route))
  541. return this.track(run, work.registration)
  542. }
  543. /** Execute and accept one current provider revision. */
  544. private async runProvider(
  545. session: Session,
  546. work: ActiveProviderWork,
  547. route?: SessionTitleModelIdentity,
  548. ): Promise<SessionTitleSnapshot | undefined> {
  549. try {
  550. this.assertCurrent(session, work)
  551. await this.ensureFallback(session)
  552. this.assertCurrent(session, work)
  553. // oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred.
  554. const messages = collectSessionTitleMessages(session.snapshotEvents(), work.throughSeq)
  555. const result = await work.registration.provider.generate({
  556. session,
  557. messages,
  558. ...route === undefined ? {} : { route },
  559. signal: work.signal,
  560. })
  561. this.assertCurrent(session, work)
  562. const accepted = this.validateResult(result, messages)
  563. session.append('session/title', {
  564. title: accepted.title,
  565. messageSeqs: [...accepted.messageSeqs],
  566. source: {
  567. kind: 'provider',
  568. provider: work.registration.provider.id,
  569. ...accepted.model === undefined ? {} : { model: accepted.model },
  570. },
  571. })
  572. return this.get(session)
  573. } finally {
  574. const state = this.work.get(session)
  575. if (state?.active === work) delete state.active
  576. }
  577. }
  578. /** Validate and normalize provider output against the supplied message snapshot. */
  579. private validateResult(
  580. result: unknown,
  581. messages: readonly SessionTitleUserMessage[],
  582. ): SessionTitleProviderResult {
  583. if (result === null || typeof result !== 'object') {
  584. throw new Error('session-title provider returned an invalid result')
  585. }
  586. const candidate = result as Record<string, unknown>
  587. if (typeof candidate.title !== 'string') throw new Error('session-title provider title must be a string')
  588. const title = normalizeSessionTitle(candidate.title, this.config.maxTitleBytes)
  589. if (title.length === 0) throw new Error('session-title provider returned an empty title')
  590. if (!Array.isArray(candidate.messageSeqs) || candidate.messageSeqs.length === 0) {
  591. throw new Error('session-title provider must identify at least one source message seq')
  592. }
  593. const messageSeqs: SessionSeq[] = []
  594. const order = new Map(messages.map((message, index) => [message.seq, index]))
  595. let previous = -1
  596. for (const seq of candidate.messageSeqs as unknown[]) {
  597. if (typeof seq !== 'number' || !Number.isSafeInteger(seq) || seq < 0) {
  598. throw new Error('session-title provider messageSeqs must be unique, ordered seqs from the request')
  599. }
  600. const sessionSeq = SessionSeq(seq)
  601. const index = order.get(sessionSeq)
  602. if (index === undefined || index <= previous) {
  603. throw new Error('session-title provider messageSeqs must be unique, ordered seqs from the request')
  604. }
  605. messageSeqs.push(sessionSeq)
  606. previous = index
  607. }
  608. const modelCandidate = candidate.model
  609. let model: SessionTitleModelIdentity | undefined
  610. if (modelCandidate !== undefined) {
  611. if (modelCandidate === null || typeof modelCandidate !== 'object') {
  612. throw new Error('session-title provider result model must contain non-empty provider and model strings')
  613. }
  614. const record = modelCandidate as Record<string, unknown>
  615. if (typeof record.provider !== 'string' || record.provider.length === 0
  616. || typeof record.model !== 'string' || record.model.length === 0) {
  617. throw new Error('session-title provider result model must contain non-empty provider and model strings')
  618. }
  619. model = { provider: record.provider, model: record.model }
  620. }
  621. return {
  622. title,
  623. messageSeqs,
  624. ...(model === undefined ? {} : { model }),
  625. }
  626. }
  627. /** Fail a completion whose provider, revision, session, or signal is stale. */
  628. private assertCurrent(session: Session, work: ActiveProviderWork): void {
  629. this.assertServiceActive()
  630. work.signal.throwIfAborted()
  631. const state = this.work.get(session)
  632. /* v8 ignore next -- every supported supersession, provider disposal, and session disposal aborts
  633. * the work signal before changing this state. */
  634. if (this.registration !== work.registration
  635. || state?.active !== work
  636. || state.revision !== work.revision
  637. || this.ctx.sessions.get(session.id) !== session) {
  638. throw new Error('session title generation state changed without cancellation')
  639. }
  640. }
  641. /** Create and publish an active provider call from one fixed revision. */
  642. private activate(
  643. pending: PendingAutomaticWork,
  644. state: SessionTitleWorkState,
  645. upstream?: AbortSignal,
  646. ): ActiveProviderWork {
  647. const controller = new AbortController()
  648. const signal = upstream === undefined
  649. ? AbortSignal.any([controller.signal, this.lifetime.signal])
  650. : AbortSignal.any([controller.signal, this.lifetime.signal, upstream])
  651. const work: ActiveProviderWork = { ...pending, controller, signal }
  652. state.active = work
  653. return work
  654. }
  655. /** Abort older active work and reserve the next session-local revision. */
  656. private supersede(state: SessionTitleWorkState, reason: string): number {
  657. state.active?.controller.abort(new Error(reason))
  658. delete state.pending
  659. state.revision += 1
  660. return state.revision
  661. }
  662. /** Return mutable work state for one session. */
  663. private stateFor(session: Session): SessionTitleWorkState {
  664. let state = this.work.get(session)
  665. if (state === undefined) {
  666. state = { revision: 0 }
  667. this.work.set(session, state)
  668. }
  669. return state
  670. }
  671. private titleInputOf(session: Session): TitleInputState {
  672. return this.ctx.sessionProjections.stateOf(session, 'titleInput') as TitleInputState
  673. }
  674. /** Queue detached service work and retain it through service disposal. */
  675. private defer(task: () => Promise<void>): void {
  676. const run = Promise.resolve().then(async () => {
  677. if (!this.serviceActive()) return
  678. await task()
  679. })
  680. void this.track(run)
  681. }
  682. /** Retain one promise until settlement for service and optional provider teardown. */
  683. private track<T>(run: Promise<T>, registration?: ProviderRegistration): Promise<T> {
  684. this.inFlight.add(run)
  685. registration?.active.add(run)
  686. const settled = (): void => {
  687. this.inFlight.delete(run)
  688. registration?.active.delete(run)
  689. }
  690. void run.then(settled, settled)
  691. return run
  692. }
  693. /** Await every current and settling promise in one lifecycle registry. */
  694. private async drain(active: Set<Promise<unknown>>): Promise<void> {
  695. while (active.size > 0) await Promise.allSettled([...active])
  696. }
  697. /** Whether the owning plugin fiber can still start or commit title work. */
  698. private serviceActive(): boolean {
  699. return !this.lifetime.signal.aborted
  700. && this.ownerFiber.uid !== null
  701. && this.ownerFiber.state === FiberState.ACTIVE
  702. }
  703. /** Reject work once the owning plugin fiber has begun unloading. */
  704. private assertServiceActive(): void {
  705. if (!this.serviceActive()) throw new Error('session-title service disposed')
  706. }
  707. /** Reject malformed provider registrations before publishing an effect. */
  708. private validateProvider(provider: unknown): asserts provider is SessionTitleProvider {
  709. if (provider === null || typeof provider !== 'object') {
  710. throw new Error('session-title provider must be an object')
  711. }
  712. const candidate = provider as Record<string, unknown>
  713. if (typeof candidate.id !== 'string' || candidate.id.length === 0) {
  714. throw new Error('session-title provider id must be a non-empty string')
  715. }
  716. if (candidate.automatic !== 'first-prompt' && candidate.automatic !== 'all-prompts') {
  717. throw new Error('session-title provider automatic mode is invalid')
  718. }
  719. if (typeof candidate.generate !== 'function') {
  720. throw new Error(`session-title provider "${candidate.id}" requires generate()`)
  721. }
  722. }
  723. /**
  724. * Derive and append the deterministic fallback title over whatever stands
  725. * (the refresh unpin path: overwriting a pinned user title is the point).
  726. * Synchronous on purpose — no await may separate derivation from append, so
  727. * it needs neither ensureFallback's in-flight dedup nor its liveness
  728. * re-check. An underivable fallback (empty after the caps) appends nothing.
  729. */
  730. private appendFallback(session: Session, first: SessionTitleUserMessage): void {
  731. const title = fallbackSessionTitle(first.text, this.config.fallbackMaxWords, this.config.fallbackMaxBytes)
  732. if (title.length === 0) return
  733. session.append('session/title', {
  734. title,
  735. messageSeqs: [first.seq],
  736. source: { kind: 'fallback' },
  737. })
  738. }
  739. /** Create the first deterministic fallback if the session still lacks a title. */
  740. private async ensureFallback(session: Session): Promise<SessionTitleSnapshot | undefined> {
  741. this.assertServiceActive()
  742. const current = this.get(session)
  743. if (current !== undefined) return current
  744. const first = this.titleInputOf(session).first
  745. if (first === null) return undefined
  746. const title = fallbackSessionTitle(
  747. first.text,
  748. this.config.fallbackMaxWords,
  749. this.config.fallbackMaxBytes,
  750. )
  751. if (title.length === 0) return undefined
  752. const state = this.stateFor(session)
  753. if (state.fallback !== undefined) return state.fallback
  754. const fallback = Promise.resolve().then(() => {
  755. this.assertServiceActive()
  756. if (this.ctx.sessions.get(session.id) !== session) {
  757. throw new Error(`session "${session.id}" is not live in this store`)
  758. }
  759. const accepted = this.get(session)
  760. if (accepted !== undefined) return accepted
  761. session.append('session/title', {
  762. title,
  763. messageSeqs: [first.seq],
  764. source: { kind: 'fallback' },
  765. })
  766. return this.get(session)
  767. })
  768. state.fallback = fallback
  769. try {
  770. return await fallback
  771. } finally {
  772. delete state.fallback
  773. }
  774. }
  775. }
  776. export default SessionTitleService