fix(session-title): honor request lifecycle

packages/session-title/session-title/README.md

@@ -7,10 +7,10 @@ Only text blocks from human `user/message` events are eligible. The first eligib
 ## Service: `SessionTitleService` (ctx key: `sessionTitle`)
 
 - `get(session)` folds the latest accepted title from a live or replayed log.
-- `refresh(session, signal?)` materializes the fallback when needed, then explicitly runs the registered provider over the current eligible messages. Provider errors and caller cancellation reject.
+- `refresh(session, signal?)` materializes the fallback when needed, then explicitly runs the registered provider over the current eligible messages. Provider errors and caller cancellation reject; cancellation does not roll back a fallback append already entering durability.
 - `register(provider)` installs the sole optional provider and returns its awaitable Cordis effect disposer. A second registration throws immediately; disposal aborts pending and active calls, waits for their settlement, and only then permits another provider to register.
 
-Automatic work never delays the main agent response. A provider starts after the matching `request/header` records the main request's exact route; its late completion joins an open turn or uses a flushed zero-step `session-title` turn through `ctx.sessions.appendOutOfBand()`. Automatic failures warn and retain the latest title. New all-message revisions, provider disposal, session disposal, and explicit refresh abort older work, and a stale completion cannot append. Concurrent explicit refreshes reserve their order before fallback durability waits, so only the newest call may reach the provider. Service teardown cancels queued work and drains calls that ignore cancellation before unloading completes.
+Automatic work never delays the main agent response. A provider starts only after a loop-built request's exact route matches the current logged `request/header`, including when the unchanged header needs no new snapshot. Its late completion joins an open turn or uses a flushed zero-step `session-title` turn through `ctx.sessions.appendOutOfBand()`. Automatic failures warn and retain the latest title. New all-message revisions, provider disposal, session disposal, and explicit refresh abort older work, and a stale completion cannot append. Concurrent explicit refreshes reserve their order before fallback durability waits, so only the newest call may reach the provider. Service teardown cancels queued work and drains calls that ignore cancellation before unloading completes.
 
 Forks inherit title events in their seed unchanged. The first-message cadence does not automatically retitle a child; the all-messages cadence may append a new revision after the child receives a later human prompt.
 

packages/session-title/session-title/src/index.ts

@@ -7,6 +7,7 @@ import { Context, FiberState, Service, type Fiber } from 'cordis'
 import z from 'schemastery'
 import type { Branded } from '@deepseek-ai/dsh-brand'
 import { deepFreeze } from '@deepseek-ai/dsh-llm'
+import type { GenerateOptions } from '@deepseek-ai/dsh-llm'
 import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
 import { fallbackSessionTitle, normalizeSessionTitle } from './normalize.ts'
 
@@ -287,6 +288,10 @@ export class SessionTitleService extends Service {
           break
       }
     })
+    ctx.on('llm/stream', (options, next) => {
+      this.onMainRequest(options)
+      return next()
+    }, { global: true, prepend: true })
     ctx.on('session/disposed', (session) => {
       const state = this.work.get(session)
       if (state === undefined) return
@@ -308,7 +313,7 @@ export class SessionTitleService extends Service {
    * Explicitly retry the registered provider, or materialize the built-in
    * fallback when no provider is registered.
    * @param session - exact live session to refresh.
-   * @param signal - optional caller cancellation.
+   * @param signal - optional caller cancellation; an in-progress fallback append may finish durably before rejection.
    * @returns latest accepted title, or `undefined` when no eligible text exists.
    */
   async refresh(session: Session, signal?: AbortSignal): Promise<SessionTitleSnapshot | undefined> {
@@ -321,7 +326,9 @@ export class SessionTitleService extends Service {
     const messages = collectSessionTitleMessages(session.events)
     const latest = messages.at(-1)
     if (registration === undefined || registration.closing || latest === undefined) {
-      return this.ensureFallback(session)
+      const fallback = await this.ensureFallback(session)
+      signal?.throwIfAborted()
+      return fallback
     }
     const state = this.stateFor(session)
     const revision = this.supersede(state, 'explicit title refresh superseded older generation')
@@ -399,11 +406,37 @@ export class SessionTitleService extends Service {
     const state = this.work.get(session)
     const pending = state?.pending
     if (state === undefined || pending === undefined || pending.throughSeq >= event.seq) return
-    delete state.pending
     const route = {
       provider: event.data.header.config.provider,
       model: event.data.header.config.model,
     }
+    this.startPending(session, state, pending, route)
+  }
+
+  /** Start unchanged-route work from the frozen loop request after its header fold is current. */
+  private onMainRequest(options: GenerateOptions): void {
+    if (!this.serviceActive() || options.sessionId === undefined || !Object.isFrozen(options)) return
+    const session = this.ctx.sessions.get(options.sessionId)
+    const state = session === undefined ? undefined : this.work.get(session)
+    const pending = state?.pending
+    if (session === undefined || state === undefined || pending === undefined) return
+    const boundary = session.events.findLast(event => event.type === 'step/start' || event.type === 'step/end')
+    const route = session.requestHeader()?.config
+    if (boundary?.type !== 'step/start'
+      || boundary.seq <= pending.throughSeq
+      || route?.provider !== options.provider
+      || route.model !== options.model) return
+    this.startPending(session, state, pending, { provider: options.provider, model: options.model })
+  }
+
+  /** Consume one pending revision and schedule its non-blocking provider call. */
+  private startPending(
+    session: Session,
+    state: SessionTitleWorkState,
+    pending: PendingAutomaticWork,
+    route: SessionTitleModelProvenance,
+  ): void {
+    delete state.pending
     this.defer(async () => {
       if (this.registration !== pending.registration
         || pending.registration.closing

packages/session-title/session-title/tests/provider.spec.ts

@@ -1,5 +1,6 @@
 import { Context } from 'cordis'
 import { describe, expect, it, vi } from 'vitest'
+import LlmService, { deepFreeze } from '@deepseek-ai/dsh-llm'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import SessionTitleService, {
   SessionTitleProviderId,
@@ -265,6 +266,95 @@ describe('SessionTitleService provider lifecycle', () => {
     expect(ctx.sessionTitle.get(session)?.title).toBe('Newest complete title')
   })
 
+  it('runs an all-messages revision when the next main request reuses its logged header', async () => {
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(SessionTitleService, CONFIG)
+    const requests: SessionTitleProviderRequest[] = []
+    ctx.sessionTitle.register({
+      id: SessionTitleProviderId('unchanged-route'),
+      automatic: 'all-user-messages',
+      async generate(request) {
+        requests.push(request)
+        return {
+          title: `Revision ${requests.length}`,
+          messageSeqs: request.messages.map(message => message.seq),
+        }
+      },
+    })
+    const session = ctx.sessions.create(SessionId('unchanged-route'))
+    session.append('turn/start', {
+      turn: 1,
+      trigger: { kind: 'message', source: { kind: 'user' } },
+    })
+    const first = appendHumanPrompt(session, 'First routed prompt')
+    await settle()
+    session.append('step/start', { turn: 1, step: 1 })
+    appendRoute(session)
+    await settle()
+    session.append('step/end', { turn: 1, step: 1 })
+    session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+
+    session.append('turn/start', {
+      turn: 2,
+      trigger: { kind: 'message', source: { kind: 'user' } },
+    })
+    const second = appendHumanPrompt(session, 'Second prompt on the same route')
+    await settle()
+    session.append('step/start', { turn: 2, step: 1 })
+    void ctx.llm.stream(deepFreeze({
+      provider: 'main-route',
+      model: 'chat-model',
+      messages: session.deriveMessages(),
+      sessionId: session.id,
+    }))
+    await settle()
+
+    expect(session.events.filter(event => event.type === 'request/header')).toHaveLength(1)
+    expect(requests).toHaveLength(2)
+    expect(requests[1]).toMatchObject({
+      messages: [
+        { seq: first.seq, text: 'First routed prompt' },
+        { seq: second.seq, text: 'Second prompt on the same route' },
+      ],
+      route: { provider: 'main-route', model: 'chat-model' },
+    })
+  })
+
+  it('ignores model streams that are not a matching loop request', async () => {
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(SessionTitleService, CONFIG)
+    const generate = vi.fn(async (request: SessionTitleProviderRequest): Promise<SessionTitleProviderResult> => ({
+      title: 'Unexpected title',
+      messageSeqs: request.messages.map(message => message.seq),
+    }))
+    ctx.sessionTitle.register({
+      id: SessionTitleProviderId('request-filter'),
+      automatic: 'all-user-messages',
+      generate,
+    })
+    const options = { provider: 'main-route', model: 'chat-model', messages: [] }
+
+    void ctx.llm.stream(deepFreeze(options))
+    void ctx.llm.stream(deepFreeze({ ...options, sessionId: SessionId('missing') }))
+    const quiet = ctx.sessions.create(SessionId('quiet'))
+    void ctx.llm.stream(deepFreeze({ ...options, sessionId: quiet.id }))
+    const pending = ctx.sessions.create(SessionId('unmatched-boundary'))
+    pending.append('turn/start', {
+      turn: 1,
+      trigger: { kind: 'message', source: { kind: 'user' } },
+    })
+    appendHumanPrompt(pending, 'Wait for a matching request boundary')
+    await settle()
+    void ctx.llm.stream(deepFreeze({ ...options, sessionId: pending.id }))
+    await settle()
+
+    expect(generate).not.toHaveBeenCalled()
+  })
+
   it('contains automatic failures but lets explicit refresh reject', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)

packages/session-title/session-title/tests/service-contracts.spec.ts

@@ -162,6 +162,37 @@ describe('SessionTitleService configuration and refresh boundaries', () => {
     expect(disposeSignal?.aborted).toBe(true)
   })
 
+  it('rejects fallback refresh cancellation that arrives during durability flush', async () => {
+    const ctx = await setup()
+    const seed = new Session(SessionId('fallback-cancel-seed'))
+    seed.append('turn/start', {
+      turn: 1,
+      trigger: { kind: 'message', source: { kind: 'user' } },
+    })
+    const source = appendPrompt(seed, 'Persist this fallback despite caller cancellation')
+    seed.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+    const session = ctx.sessions.create(SessionId('fallback-cancel'), { seed: seed.events })
+    const flushStarted = deferred<undefined>()
+    const releaseFlush = deferred<undefined>()
+    ctx.on('session/flush', async (subject) => {
+      if (subject !== session) return
+      flushStarted.resolve(undefined)
+      await releaseFlush.promise
+    })
+    const controller = new AbortController()
+
+    const refresh = ctx.sessionTitle.refresh(session, controller.signal)
+    await flushStarted.promise
+    controller.abort(new Error('cancelled while fallback flushed'))
+    releaseFlush.resolve(undefined)
+
+    await expect(refresh).rejects.toThrow('cancelled while fallback flushed')
+    expect(ctx.sessionTitle.get(session)).toMatchObject({
+      messageSeqs: [source.seq],
+      source: { kind: 'fallback' },
+    })
+  })
+
   it('reserves overlapping refresh order before fallback durability settles', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)