fix(session-title): serialize provider request writes

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

@@ -8,7 +8,7 @@ This package is a library, not a Cordis plugin. The provider plugins call `regis
 
 `provider` and `model` overrides are optional but must be supplied together as non-empty strings. Without that pair, the helper uses the exact provider/model route captured from the current session's logged `request/header`; an explicit refresh before any route exists therefore needs overrides. Input exceeding `maxInputBytes` rejects instead of being truncated. Timeout, cancellation, malformed or empty output, tool calls, and non-stop finish reasons also reject; the session-title service decides whether that rejection is an automatic warning or an explicit caller failure.
 
-After route and input validation, the helper appends a log-only `session/title-llm-request` event before model dispatch. It contains the title-provider id, exact source seqs, route, system prompt, message list, and output-token cap used by the call. The dispatched envelope is deep-frozen to keep interceptors aligned with that record but deliberately lacks dsh-agent-loop's process-local request identity, so loop-only reconstruction observers do not compare it with the conversation header. A later model failure leaves that request record intact; validation failures that never become dispatchable requests do not create one. The event stays outside derived model history.
+After route and input validation, the helper appends a log-only `session/title-llm-request` event before model dispatch. It contains the title-provider id, exact source seqs, route, system prompt, message list, and output-token cap used by the call. The append shares the title capability's per-session settlement queue, so a superseding request cannot collide with an earlier fallback, request record, or accepted-title flush. The dispatched envelope is deep-frozen to keep interceptors aligned with that record but deliberately lacks dsh-agent-loop's process-local request identity, so loop-only reconstruction observers do not compare it with the conversation header. A later model failure leaves that request record intact; validation failures that never become dispatchable requests do not create one. The event stays outside derived model history.
 
 ## Configuration
 

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

@@ -9,7 +9,11 @@ import z from 'schemastery'
 import { BlockAssembler, deepFreeze } from '@deepseek-ai/dsh-llm'
 import type { FinishReason, GenerateOptions, Message } from '@deepseek-ai/dsh-llm'
 import { deadline, MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
-import { normalizeSessionTitle, SessionTitleProviderId } from '@deepseek-ai/dsh-session-title'
+import {
+  appendSessionTitleOutOfBand,
+  normalizeSessionTitle,
+  SessionTitleProviderId,
+} from '@deepseek-ai/dsh-session-title'
 import type {
   SessionTitleAutomaticMode,
   SessionTitleModelProvenance,
@@ -258,14 +262,14 @@ export async function generateSessionTitleWithLlm(
     sessionId: request.session.id,
     signal: callDeadline.signal,
   })
-  await ctx.sessions.appendOutOfBand(request.session, 'session/title-llm-request', {
+  await appendSessionTitleOutOfBand(ctx, request.session, 'session/title-llm-request', {
     titleProvider,
     messageSeqs: selectedMessages.map(message => message.seq),
     route,
     system,
     messages,
     maxTokens: config.maxOutputTokens,
-  }, { kind: 'session-title' })
+  }, callDeadline.signal)
   callDeadline.signal.throwIfAborted()
   const assembler = new BlockAssembler()
   for await (const chunk of ctx.llm.stream(options)) assembler.push(chunk)

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

@@ -10,7 +10,7 @@ Only text blocks from human `user/message` events are eligible. The first eligib
 - `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 only after a marked 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.
+Automatic work never delays the main agent response. A provider starts only after a marked 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. Service and bundled model-provider records use `appendSessionTitleOutOfBand()` to share a per-session settlement queue, so a replacement request record waits for an earlier title write without serializing the superseded model call itself. 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

@@ -8,7 +8,12 @@ import z from 'schemastery'
 import type { Branded } from '@deepseek-ai/dsh-brand'
 import { deepFreeze, isAgentLoopRequest } from '@deepseek-ai/dsh-llm'
 import type { GenerateOptions } from '@deepseek-ai/dsh-llm'
-import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
+import type {
+  OutOfBandSessionEventType,
+  Session,
+  SessionEvent,
+  SessionEventMap,
+} from '@deepseek-ai/dsh-session'
 import { fallbackSessionTitle, normalizeSessionTitle } from './normalize.ts'
 
 export { fallbackSessionTitle, normalizeSessionTitle, truncateTitleUtf8 } from './normalize.ts'
@@ -95,6 +100,46 @@ declare module '@deepseek-ai/dsh-session' {
   }
 }
 
+/** Per-session settlement tails for title-capability out-of-band writes. */
+const SESSION_TITLE_WRITE_TAILS = new WeakMap<Session, Promise<void>>()
+
+/** Convert either write outcome into a fulfilled queue tail. */
+function settleSessionTitleWrite(): void {}
+
+/**
+ * Serialize one title-capability out-of-band event with its session peers.
+ * Cancellation is checked when the write reaches the head of the queue; once
+ * the core append starts, its durability contract runs to completion.
+ * @param ctx - context exposing the live session store.
+ * @param session - exact live session that owns the title-capability event.
+ * @param type - plugin-declared log-only title event type.
+ * @param data - typed JSON payload for the event.
+ * @param signal - service or provider lifetime checked before publication starts.
+ * @returns the durably accepted event.
+ */
+export async function appendSessionTitleOutOfBand<T extends OutOfBandSessionEventType>(
+  ctx: Context,
+  session: Session,
+  type: T,
+  data: SessionEventMap[T],
+  signal: AbortSignal,
+): Promise<SessionEvent<T>> {
+  const predecessor = SESSION_TITLE_WRITE_TAILS.get(session)
+  const run = Promise.resolve(predecessor).then(() => {
+    signal.throwIfAborted()
+    return ctx.sessions.appendOutOfBand(session, type, data, { kind: 'session-title' })
+  })
+  const tail = run.then(settleSessionTitleWrite, settleSessionTitleWrite)
+  SESSION_TITLE_WRITE_TAILS.set(session, tail)
+  try {
+    return await run
+  } finally {
+    if (SESSION_TITLE_WRITE_TAILS.get(session) === tail) {
+      SESSION_TITLE_WRITE_TAILS.delete(session)
+    }
+  }
+}
+
 /** One eligible human text message exposed to title providers. */
 export interface SessionTitleUserMessage {
   /** Source `user/message` event seq. */
@@ -481,7 +526,7 @@ export class SessionTitleService extends Service {
       })
       this.assertCurrent(session, work)
       const accepted = this.validateResult(result, messages)
-      await this.ctx.sessions.appendOutOfBand(session, 'session/title', {
+      await appendSessionTitleOutOfBand(this.ctx, session, 'session/title', {
         title: accepted.title,
         messageSeqs: [...accepted.messageSeqs],
         source: {
@@ -489,7 +534,7 @@ export class SessionTitleService extends Service {
           provider: work.registration.provider.id,
           ...accepted.model === undefined ? {} : { model: accepted.model },
         },
-      }, { kind: 'session-title' })
+      }, work.signal)
       return this.get(session)
     } finally {
       const state = this.work.get(session)
@@ -662,11 +707,11 @@ export class SessionTitleService extends Service {
       this.config.fallbackMaxBytes,
     )
     if (title.length === 0) return undefined
-    await this.ctx.sessions.appendOutOfBand(session, 'session/title', {
+    await appendSessionTitleOutOfBand(this.ctx, session, 'session/title', {
       title,
       messageSeqs: [first.seq],
       source: { kind: 'fallback' },
-    }, { kind: 'session-title' })
+    }, this.lifetime.signal)
     return this.get(session)
   }
 }

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

@@ -2,6 +2,7 @@ import { Context, type Fiber } from 'cordis'
 import { describe, expect, it, vi } from 'vitest'
 import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session'
 import SessionTitleService, {
+  appendSessionTitleOutOfBand,
   SessionTitleProviderId,
   type Config,
   type SessionTitleProvider,
@@ -9,6 +10,16 @@ import SessionTitleService, {
   type SessionTitleProviderResult,
 } from '@deepseek-ai/dsh-session-title'
 
+declare module '@deepseek-ai/dsh-session' {
+  interface SessionEventMap {
+    'test/title-provider-request': { revision: number }
+  }
+
+  interface OutOfBandSessionEventMap {
+    'test/title-provider-request': true
+  }
+}
+
 const CONFIG = {
   fallbackMaxWords: 5,
   fallbackMaxBytes: 40,
@@ -247,6 +258,78 @@ describe('SessionTitleService configuration and refresh boundaries', () => {
     expect(olderError.message).toMatch(/superseded/)
   })
 
+  it('serializes a newer provider write after the superseded write', async () => {
+    const ctx = await setup()
+    const session = startSession(ctx, 'refresh-provider-write-order')
+    const source = appendPrompt(session, 'Serialize explicit provider writes')
+    await settle()
+    session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+    const flushStarted = deferred<undefined>()
+    const releaseFlush = deferred<undefined>()
+    let flushCount = 0
+    ctx.on('session/flush', async (subject) => {
+      if (subject !== session || ++flushCount !== 1) return
+      flushStarted.resolve(undefined)
+      await releaseFlush.promise
+    })
+    let generation = 0
+    ctx.sessionTitle.register({
+      id: SessionTitleProviderId('refresh-provider-write-order'),
+      automatic: 'first-message',
+      async generate(request) {
+        generation += 1
+        const revision = generation
+        await appendSessionTitleOutOfBand(ctx, request.session, 'test/title-provider-request', {
+          revision,
+        }, request.signal)
+        return {
+          title: `Generated title ${revision}`,
+          messageSeqs: [source.seq],
+        }
+      },
+    })
+
+    const older = ctx.sessionTitle.refresh(session)
+    const olderOutcome = older.then(
+      () => undefined,
+      (error: unknown) => error,
+    )
+    await flushStarted.promise
+    const middle = ctx.sessionTitle.refresh(session)
+    const middleOutcome = middle.then(
+      value => value,
+      (error: unknown) => error,
+    )
+    await settle()
+
+    expect(generation).toBe(2)
+    expect(session.events.filter(event => event.type === 'test/title-provider-request'))
+      .toHaveLength(1)
+    const newer = ctx.sessionTitle.refresh(session)
+    const newerOutcome = newer.then(
+      value => value,
+      (error: unknown) => error,
+    )
+    await settle()
+    expect(generation).toBe(3)
+    expect(session.events.filter(event => event.type === 'test/title-provider-request'))
+      .toHaveLength(1)
+
+    releaseFlush.resolve(undefined)
+    const newerResult = await newerOutcome
+    expect(newerResult).toMatchObject({ title: 'Generated title 3' })
+    const olderError = await olderOutcome
+    expect(olderError).toBeInstanceOf(Error)
+    if (!(olderError instanceof Error)) throw new Error('expected older refresh to reject')
+    expect(olderError.message).toMatch(/superseded/)
+    const middleError = await middleOutcome
+    expect(middleError).toBeInstanceOf(Error)
+    if (!(middleError instanceof Error)) throw new Error('expected middle refresh to reject')
+    expect(middleError.message).toMatch(/superseded/)
+    expect(session.events.filter(event => event.type === 'test/title-provider-request').map(event => event.data.revision))
+      .toEqual([1, 3])
+  })
+
   it('cancels a queued fallback when the session-title service unloads', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)