Merge remote-tracking branch 'origin/master' into xtr/identified-immutable-messages
# Conflicts: # .agents/notes/implemented/architecture/2026-07-24-separate-context-injection-from-turn-execution.i18n.yaml # docs/cordis-catalog/services.md # docs/core-data-structures/core.i18n.yaml # docs/core-data-structures/session.i18n.yaml # docs/event-producer-consumer.md # docs/persistence-catalog.md # packages/core/session/README.i18n.yaml # packages/session-title/session-title/tests/persistence.spec.ts
This commit is contained in:
_Kerman
@@ -1,6 +1,6 @@
|
||||
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write
|
||||
README.md: 5d004b634e6242a8a558182e6c953e4c43b25e1f
|
||||
README.zh.md: 36567d950e612561d8af804eca64ac6f9f3cd788
|
||||
# pnpm run verify-translation-pairing --write packages/session-title/session-title-llm/README.md
|
||||
README.md: 342687b5aa0cd35a70abf8cc66b3fe80342bce0b
|
||||
README.zh.md: 2bb48b52ae6298c5c895ef99ff54cf5f1d7a6d8c
|
||||
|
||||
@@ -10,7 +10,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. The helper measures the final JSON-framed user prompt, including seq fields, wrappers, and JSON escaping, against `maxInputBytes` before logging or dispatch instead of truncating it. Timeout and caller cancellation are rechecked while consuming the stream and after it completes, so a late successful result cannot be accepted even if an interceptor or adapter ignores abort. 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 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, carries `purpose: 'session-title'`, and deliberately lacks dsh-agent-loop's process-local request identity. Interceptors stay aligned with the record while loop-only reconstruction observers do not compare it with the conversation header. The DeepSeek adapter maps that purpose to thinking-disabled so the small output budget is reserved for visible title text; other adapters own their purpose-specific behavior. A later model failure leaves the 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 directly through `Session` 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. Persistence observes the record eagerly; the append does not need a title-specific marker, cast, settlement queue, or flush. The dispatched envelope is deep-frozen, carries `purpose: 'session-title'`, and deliberately lacks dsh-agent-loop's process-local request identity. Interceptors stay aligned with the record while loop-only reconstruction observers do not compare it with the conversation header. The DeepSeek adapter maps that purpose to thinking-disabled so the small output budget is reserved for visible title text; other adapters own their purpose-specific behavior. A later model failure leaves the request record intact; validation failures that never become dispatchable requests do not create one. The event stays outside derived model history.
|
||||
|
||||
## Configuration
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
|
||||
`provider` 和 `model` 覆盖项都是可选的，但必须同时作为非空字符串提供。如果没有这一对取值，辅助模块会使用当前会话已记录 `request/header` 中捕获的确切提供方／模型路由；因此，在任何路由出现前显式刷新时必须提供覆盖项。辅助模块在记录或分发前，以 `maxInputBytes` 测量最终 JSON 封装的用户提示词，包括 seq 字段、包装层与 JSON 转义，而不是将其截断。消费流期间和流完成后都会重新检查超时与调用方取消，因此即使 interceptor 或适配器忽略 abort，也不能接受迟到的成功结果。格式错误或空输出、工具调用和非 stop 结束原因同样会 reject；会话标题服务决定该 reject 属于自动警告还是显式调用方失败。
|
||||
|
||||
路由与输入验证完成后，辅助模块会在模型分发前追加仅写入日志的 `session/title-llm-request` 事件。它包含标题提供方 id、确切来源 seq、路由、系统提示词、消息列表，以及该调用使用的输出 token 上限。追加操作共享标题能力的逐会话结算队列，因此取代当前请求的新请求不会与更早回退、请求记录或已接受标题的 flush 冲突。分发的 envelope 会深度冻结，携带 `purpose: 'session-title'`，且有意不包含 dsh-agent-loop 的进程本地请求身份。Interceptor 会与记录保持一致，而循环专用重建观察者不会把它与对话 header 比较。DeepSeek 适配器会将该 purpose 映射为关闭 thinking，使少量输出预算全部用于可见标题文本；其他适配器负责自身 purpose 专用行为。后续模型失败会保留请求记录；从未成为可分发请求的验证失败不会创建记录。该事件始终位于派生模型历史之外。
|
||||
路由与输入验证完成后，辅助模块会在模型分发前直接通过 `Session` 追加仅写入日志的 `session/title-llm-request` 事件。它包含标题提供方 id、确切来源 seq、路由、系统提示词、消息列表，以及该调用使用的输出 token 上限。持久化会尽快观察该记录；追加不需要标题专属标记、类型断言、结算队列或 flush。分发的 envelope 会深度冻结，携带 `purpose: 'session-title'`，且有意不包含 dsh-agent-loop 的进程本地请求身份。Interceptor 会与记录保持一致，而循环专用重建观察者不会把它与对话 header 比较。DeepSeek 适配器会将该 purpose 映射为关闭 thinking，使少量输出预算全部用于可见标题文本；其他适配器负责自身 purpose 专用行为。后续模型失败会保留请求记录；从未成为可分发请求的验证失败不会创建记录。该事件始终位于派生模型历史之外。
|
||||
|
||||
## 配置
|
||||
|
||||
|
||||
@@ -10,7 +10,6 @@ import { createUserMessage, BlockAssembler, deepFreeze } from '@deepseek-ai/dsh-
|
||||
import type { FinishReason, GenerateOptions, Message } from '@deepseek-ai/dsh-llm'
|
||||
import { deadline, MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
|
||||
import {
|
||||
appendSessionTitleOutOfBand,
|
||||
normalizeSessionTitle,
|
||||
SessionTitleProviderId,
|
||||
} from '@deepseek-ai/dsh-session-title'
|
||||
@@ -43,10 +42,6 @@ declare module '@deepseek-ai/dsh-session' {
|
||||
/** Log-only pre-dispatch record of one session-title model request. */
|
||||
'session/title-llm-request': SessionTitleLlmRequestEventData
|
||||
}
|
||||
|
||||
interface OutOfBandSessionEventMap {
|
||||
'session/title-llm-request': true
|
||||
}
|
||||
}
|
||||
|
||||
/** Capability-owned timeout reason code for auxiliary title requests. */
|
||||
@@ -264,14 +259,14 @@ export async function generateSessionTitleWithLlm(
|
||||
purpose: 'session-title',
|
||||
signal: callDeadline.signal,
|
||||
})
|
||||
await appendSessionTitleOutOfBand(ctx, request.session, 'session/title-llm-request', {
|
||||
request.session.append('session/title-llm-request', {
|
||||
titleProvider,
|
||||
messageSeqs: selectedMessages.map(message => message.seq),
|
||||
route,
|
||||
system,
|
||||
messages,
|
||||
maxTokens: config.maxOutputTokens,
|
||||
}, callDeadline.signal)
|
||||
})
|
||||
callDeadline.signal.throwIfAborted()
|
||||
const assembler = new BlockAssembler()
|
||||
for await (const chunk of ctx.llm.stream(options)) {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write
|
||||
README.md: 2be6f37ce37f525dd567772144e925886ad51b42
|
||||
README.zh.md: 7f55b7bba912bf4611f5a058a0390d67b8860b24
|
||||
# pnpm run verify-translation-pairing --write packages/session-title/session-title/README.md
|
||||
README.md: 1939d00f7e78834ec19e2d6b4590cf12af297a30
|
||||
README.zh.md: d373c212a193a82567686a634bad79726185e832
|
||||
|
||||
@@ -9,10 +9,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; cancellation does not roll back a fallback append already entering durability.
|
||||
- `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 an already accepted fallback event.
|
||||
- `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, while overlapping automatic and explicit fallback requests share one session-local in-flight append. 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.
|
||||
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 appends a standalone log-only event directly through `Session` without opening a turn. Persistence observes that event eagerly and drains on ordinary lifecycle checkpoints; title publication itself does not force a flush. 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 revision before provider work, while overlapping automatic and explicit fallback requests share one session-local in-flight append. The service and bundled model provider each append their own literal event type, so no generic title-write marker, cast, or settlement queue is needed. 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.
|
||||
|
||||
|
||||
@@ -9,10 +9,10 @@
|
||||
## 服务：`SessionTitleService`（ctx 键：`sessionTitle`）
|
||||
|
||||
- `get(session)` 从活跃或回放日志折叠最新已接受标题。
|
||||
- `refresh(session, signal?)` 在需要时物化回退，然后显式运行已注册提供方，处理当前符合条件的消息。提供方错误与调用方取消会 reject；取消不会回滚已经进入持久化流程的回退追加。
|
||||
- `refresh(session, signal?)` 在需要时物化回退，然后显式运行已注册提供方，处理当前符合条件的消息。提供方错误与调用方取消会 reject；取消不会回滚已接受的回退事件。
|
||||
- `register(provider)` 安装唯一可选提供方，并返回可等待的 Cordis effect disposer。第二次注册会立即抛出；资源释放会中止待处理和活跃调用，等待其结算，之后才允许注册另一个提供方。
|
||||
|
||||
自动工作绝不会延迟主 agent 响应。只有当带标记、由循环构建的请求，其确切路由与当前已记录的 `request/header` 匹配时，提供方才会启动；即使 header 未变而无需新快照，也适用此规则。延迟完成会加入开放轮次，或使用已经 flush 的零步骤 `session-title` 轮次，并通过 `ctx.sessions.appendOutOfBand()` 追加。自动失败会发出警告并保留最新标题。新的全消息 revision、提供方资源释放、会话资源释放和显式刷新都会中止旧工作，陈旧完成值无法追加。并发显式刷新会在等待回退持久化前预留顺序；重叠的自动／显式回退请求共享一个会话本地进行中追加。服务与随附模型提供方记录使用 `appendSessionTitleOutOfBand()`，共享逐会话结算队列，因此替换请求记录会等待更早标题写入，但无需串行等待被取代的模型调用本身。服务 teardown 会取消排队工作，并在卸载完成前排空忽略取消的调用。
|
||||
自动工作绝不会延迟主 agent（智能体）响应。只有当带标记、由循环构建的请求，其确切路由与当前已记录的 `request/header` 匹配时，提供方才会启动；即使 header 未变而无需新快照，也适用此规则。延迟完成会直接通过 `Session` 追加一个独立的纯日志事件，而不打开轮次。持久化会尽快观察该事件，并在常规生命周期检查点排空；标题发布本身不会强制 flush。自动失败会发出警告并保留最新标题。新的全消息 revision、提供方资源释放、会话资源释放和显式刷新都会中止旧工作，陈旧完成值无法追加。并发显式刷新会在提供方工作之前预留修订号；重叠的自动／显式回退请求共享一个会话本地进行中追加。服务与随附模型提供方各自追加自己的字面量事件类型，因此不需要通用标题写入标记、类型断言或结算队列。服务 teardown 会取消排队工作，并在卸载完成前排空忽略取消的调用。
|
||||
|
||||
Fork 会原样继承 seed 中的标题事件。首消息节奏不会自动为子会话重新生成标题；全消息节奏可以在子会话收到后续用户提示词后追加新 revision。
|
||||
|
||||
|
||||
@@ -9,10 +9,8 @@ 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 {
|
||||
OutOfBandSessionEventType,
|
||||
Session,
|
||||
SessionEvent,
|
||||
SessionEventMap,
|
||||
} from '@deepseek-ai/dsh-session'
|
||||
import { fallbackSessionTitle, normalizeSessionTitle } from './normalize.ts'
|
||||
|
||||
@@ -82,11 +80,6 @@ declare module 'cordis' {
|
||||
}
|
||||
|
||||
declare module '@deepseek-ai/dsh-session' {
|
||||
interface TurnTriggerMap {
|
||||
/** Zero-step turn opened only to durably append a late title update. */
|
||||
'session-title': { kind: 'session-title' }
|
||||
}
|
||||
|
||||
interface SessionEventMap {
|
||||
/**
|
||||
* Latest-wins session title snapshot. Log-only: it never enters the model
|
||||
@@ -94,50 +87,6 @@ declare module '@deepseek-ai/dsh-session' {
|
||||
*/
|
||||
'session/title': SessionTitleEventData
|
||||
}
|
||||
|
||||
interface OutOfBandSessionEventMap {
|
||||
'session/title': true
|
||||
}
|
||||
}
|
||||
|
||||
/** 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. */
|
||||
@@ -163,7 +112,7 @@ export interface SessionTitleProviderRequest {
|
||||
readonly signal: AbortSignal
|
||||
}
|
||||
|
||||
/** Provider output before service-owned normalization and durable acceptance. */
|
||||
/** Provider output before service-owned normalization and log acceptance. */
|
||||
export interface SessionTitleProviderResult {
|
||||
/** Proposed title text. */
|
||||
readonly title: string
|
||||
@@ -360,7 +309,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; an in-progress fallback append may finish durably before rejection.
|
||||
* @param signal - optional caller cancellation.
|
||||
* @returns latest accepted title, or `undefined` when no eligible text exists.
|
||||
*/
|
||||
async refresh(session: Session, signal?: AbortSignal): Promise<SessionTitleSnapshot | undefined> {
|
||||
@@ -509,7 +458,7 @@ export class SessionTitleService extends Service {
|
||||
return this.track(run, work.registration)
|
||||
}
|
||||
|
||||
/** Execute and durably accept one current provider revision. */
|
||||
/** Execute and accept one current provider revision. */
|
||||
private async runProvider(
|
||||
session: Session,
|
||||
work: ActiveProviderWork,
|
||||
@@ -528,7 +477,7 @@ export class SessionTitleService extends Service {
|
||||
})
|
||||
this.assertCurrent(session, work)
|
||||
const accepted = this.validateResult(result, messages)
|
||||
await appendSessionTitleOutOfBand(this.ctx, session, 'session/title', {
|
||||
session.append('session/title', {
|
||||
title: accepted.title,
|
||||
messageSeqs: [...accepted.messageSeqs],
|
||||
source: {
|
||||
@@ -536,7 +485,7 @@ export class SessionTitleService extends Service {
|
||||
provider: work.registration.provider.id,
|
||||
...accepted.model === undefined ? {} : { model: accepted.model },
|
||||
},
|
||||
}, work.signal)
|
||||
})
|
||||
return this.get(session)
|
||||
} finally {
|
||||
const state = this.work.get(session)
|
||||
@@ -711,11 +660,20 @@ export class SessionTitleService extends Service {
|
||||
if (title.length === 0) return undefined
|
||||
const state = this.stateFor(session)
|
||||
if (state.fallback !== undefined) return state.fallback
|
||||
const fallback = appendSessionTitleOutOfBand(this.ctx, session, 'session/title', {
|
||||
title,
|
||||
messageSeqs: [first.seq],
|
||||
source: { kind: 'fallback' },
|
||||
}, this.lifetime.signal).then(() => this.get(session))
|
||||
const fallback = Promise.resolve().then(() => {
|
||||
this.assertServiceActive()
|
||||
if (this.ctx.sessions.get(session.id) !== session) {
|
||||
throw new Error(`session "${session.id}" is not live in this store`)
|
||||
}
|
||||
const accepted = this.get(session)
|
||||
if (accepted !== undefined) return accepted
|
||||
session.append('session/title', {
|
||||
title,
|
||||
messageSeqs: [first.seq],
|
||||
source: { kind: 'fallback' },
|
||||
})
|
||||
return this.get(session)
|
||||
})
|
||||
state.fallback = fallback
|
||||
try {
|
||||
return await fallback
|
||||
|
||||
@@ -15,8 +15,9 @@ export const name = 'session-title-invariant'
|
||||
export const inject = ['invariants']
|
||||
|
||||
/**
|
||||
* No runtime invariant: the service validates provider revisions before their single durable
|
||||
* append, and its remaining provider lifecycle state is process-local and covered by package tests.
|
||||
* No runtime invariant: the service validates provider revisions before their
|
||||
* title append, and its remaining lifecycle state is process-local and covered
|
||||
* by package tests.
|
||||
*/
|
||||
const install: InvariantInstaller = () => {}
|
||||
|
||||
|
||||
@@ -31,9 +31,8 @@ async function appendPersistedTitle(ctx: Context, id: ReturnType<typeof SessionI
|
||||
content: [{ type: 'text', text: 'Persist this session title' }],
|
||||
source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
await new Promise(resolve => setTimeout(resolve, 0))
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
await ctx.parallel('session/flush', session)
|
||||
await ctx.sessionTitle.refresh(session)
|
||||
}
|
||||
|
||||
async function expectPersistedTitle(ctx: Context, id: ReturnType<typeof SessionId>): Promise<void> {
|
||||
@@ -42,13 +41,13 @@ async function expectPersistedTitle(ctx: Context, id: ReturnType<typeof SessionI
|
||||
title: 'Persist this session title',
|
||||
messageSeqs: [1],
|
||||
source: { kind: 'fallback' },
|
||||
eventSeq: 2,
|
||||
eventSeq: 3,
|
||||
})
|
||||
expect(loaded.events.map(event => event.type)).toEqual([
|
||||
'turn/start',
|
||||
'user/message',
|
||||
'session/title',
|
||||
'turn/end',
|
||||
'session/title',
|
||||
])
|
||||
}
|
||||
|
||||
|
||||
@@ -3,7 +3,6 @@ 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,
|
||||
@@ -11,16 +10,6 @@ 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,
|
||||
@@ -174,38 +163,7 @@ 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('shares one durable fallback across concurrent refreshes', async () => {
|
||||
it('shares one fallback across concurrent refreshes', async () => {
|
||||
const ctx = await setup()
|
||||
const seed = new Session(SessionId('fallback-concurrency-seed'))
|
||||
seed.append('turn/start', {
|
||||
@@ -215,10 +173,6 @@ describe('SessionTitleService configuration and refresh boundaries', () => {
|
||||
const source = appendPrompt(seed, 'Create exactly one fallback title')
|
||||
seed.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
const session = ctx.sessions.create(SessionId('fallback-concurrency'), { seed: seed.events })
|
||||
let flushes = 0
|
||||
ctx.on('session/flush', (subject) => {
|
||||
if (subject === session) flushes += 1
|
||||
})
|
||||
|
||||
const results = await Promise.all([
|
||||
ctx.sessionTitle.refresh(session),
|
||||
@@ -227,136 +181,62 @@ describe('SessionTitleService configuration and refresh boundaries', () => {
|
||||
|
||||
expect(results[0]).toEqual(results[1])
|
||||
expect(session.events.filter(event => event.type === 'session/title')).toHaveLength(1)
|
||||
expect(session.events.filter(event => event.type === 'turn/start'
|
||||
&& event.data.trigger.kind === 'session-title')).toHaveLength(1)
|
||||
expect(session.events.map(event => event.type)).toEqual([
|
||||
'turn/start',
|
||||
'user/message',
|
||||
'turn/end',
|
||||
'session/title',
|
||||
])
|
||||
expect(ctx.sessionTitle.get(session)?.messageSeqs).toEqual([source.seq])
|
||||
expect(flushes).toBe(1)
|
||||
})
|
||||
|
||||
it('reserves overlapping refresh order before fallback durability settles', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionTitleService, CONFIG)
|
||||
const seed = new Session(SessionId('refresh-order-seed'))
|
||||
seed.append('turn/start', {
|
||||
turn: 1,
|
||||
trigger: { kind: 'message', source: { kind: 'user' } },
|
||||
it('reuses a title accepted before the queued fallback commits', async () => {
|
||||
const ctx = await setup()
|
||||
const session = startSession(ctx, 'fallback-already-accepted')
|
||||
const source = appendPrompt(session, 'Reuse the title that wins the fallback race')
|
||||
|
||||
const refresh = ctx.sessionTitle.refresh(session)
|
||||
session.append('session/title', {
|
||||
title: 'Already accepted',
|
||||
messageSeqs: [source.seq],
|
||||
source: { kind: 'fallback' },
|
||||
})
|
||||
const source = appendPrompt(seed, 'Keep the newest explicit refresh')
|
||||
seed.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
const session = ctx.sessions.create(SessionId('refresh-order'), { seed: seed.events })
|
||||
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
|
||||
})
|
||||
const result = deferred<SessionTitleProviderResult>()
|
||||
|
||||
await expect(refresh).resolves.toMatchObject({ title: 'Already accepted' })
|
||||
expect(session.events.filter(event => event.type === 'session/title')).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('lets the newest overlapping explicit refresh win', async () => {
|
||||
const ctx = await setup()
|
||||
const session = startSession(ctx, 'refresh-order')
|
||||
const source = appendPrompt(session, 'Keep the newest explicit refresh')
|
||||
await settle()
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
const requests: SessionTitleProviderRequest[] = []
|
||||
const results: Array<ReturnType<typeof deferred<SessionTitleProviderResult>>> = []
|
||||
ctx.sessionTitle.register({
|
||||
id: SessionTitleProviderId('refresh-order'),
|
||||
automatic: 'first-message',
|
||||
generate(request) {
|
||||
requests.push(request)
|
||||
const result = deferred<SessionTitleProviderResult>()
|
||||
results.push(result)
|
||||
return result.promise
|
||||
},
|
||||
})
|
||||
|
||||
const older = ctx.sessionTitle.refresh(session)
|
||||
const olderOutcome = older.then(
|
||||
() => undefined,
|
||||
(error: unknown) => error,
|
||||
)
|
||||
await flushStarted.promise
|
||||
await settle()
|
||||
const newer = ctx.sessionTitle.refresh(session)
|
||||
await settle()
|
||||
expect(requests).toHaveLength(1)
|
||||
expect(requests[0]?.signal.aborted).toBe(false)
|
||||
|
||||
releaseFlush.resolve(undefined)
|
||||
await settle()
|
||||
expect(requests).toHaveLength(1)
|
||||
expect(requests[0]?.signal.aborted).toBe(false)
|
||||
result.resolve({ title: 'Newest explicit title', messageSeqs: [source.seq] })
|
||||
expect(requests).toHaveLength(2)
|
||||
expect(requests[0]?.signal.aborted).toBe(true)
|
||||
expect(requests[1]?.signal.aborted).toBe(false)
|
||||
results[0]?.resolve({ title: 'Obsolete title', messageSeqs: [source.seq] })
|
||||
await expect(older).rejects.toThrow(/superseded/)
|
||||
results[1]?.resolve({ title: 'Newest explicit title', messageSeqs: [source.seq] })
|
||||
await expect(newer).resolves.toMatchObject({ title: 'Newest explicit title' })
|
||||
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/)
|
||||
})
|
||||
|
||||
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 () => {
|
||||
@@ -391,6 +271,21 @@ describe('SessionTitleService configuration and refresh boundaries', () => {
|
||||
expect(inactiveError.message).toBe('session-title service disposed')
|
||||
})
|
||||
|
||||
it('suppresses a queued fallback failure after service unload begins', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionTitleService, CONFIG)
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
|
||||
const session = startSession(ctx, 'service-unload-started-fallback')
|
||||
appendPrompt(session, 'Start fallback before unloading the service')
|
||||
|
||||
await Promise.resolve()
|
||||
await fiber.dispose()
|
||||
|
||||
expect(session.events.some(event => event.type === 'session/title')).toBe(false)
|
||||
expect(warn).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('aborts pending and active provider work and drains ignored cancellation during service unload', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
@@ -431,31 +326,6 @@ describe('SessionTitleService configuration and refresh boundaries', () => {
|
||||
await expect(refreshOutcome).resolves.toEqual(expect.objectContaining({ message: 'session-title service disposed' }))
|
||||
})
|
||||
|
||||
it('suppresses a queued fallback failure after service unload begins', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionTitleService, CONFIG)
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
|
||||
const session = startSession(ctx, 'service-unload-flush')
|
||||
appendPrompt(session, 'Fallback whose flush outlives the service')
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
const flushStarted = deferred<undefined>()
|
||||
const releaseFlush = deferred<undefined>()
|
||||
ctx.on('session/flush', async (subject) => {
|
||||
if (subject !== session) return
|
||||
flushStarted.resolve(undefined)
|
||||
await releaseFlush.promise
|
||||
throw new Error('flush failed during service unload')
|
||||
})
|
||||
|
||||
await flushStarted.promise
|
||||
const disposal = fiber.dispose()
|
||||
releaseFlush.resolve(undefined)
|
||||
await disposal
|
||||
|
||||
expect(warn).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('warns when a detached session prevents queued fallback publication', async () => {
|
||||
const ctx = await setup()
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
|
||||
|
||||
Reference in New Issue
Block a user