diff --git a/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.i18n.yaml b/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.i18n.yaml index 361145cb02..7425619403 100644 --- a/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.i18n.yaml +++ b/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.i18n.yaml @@ -2,5 +2,5 @@ # 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 -2026-07-24-model-facing-session-query-tools.md: 68169d8af7176ee1725a3c58bf97530f56a0765b -2026-07-24-model-facing-session-query-tools.zh.md: dfeae26a548b5e498e76fce259f7610106de7ec2 +2026-07-24-model-facing-session-query-tools.md: 0551adc431388d6cdd94b8e03c2976020ec90de4 +2026-07-24-model-facing-session-query-tools.zh.md: f82c0fac52d63ac3c11f48ee2769cb9e9590317c diff --git a/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.md b/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.md index 68169d8af7..0551adc431 100644 --- a/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.md +++ b/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.md @@ -28,7 +28,7 @@ Neither search tool exposes a cursor, offset, page size, or model-controlled res Trace and read tools likewise expose no lineage or character pagination. Canonical results are plain text and remain complete within the service's existing event-window and search-count resource bounds. The generic `tools/post-execute` spill policy owns inline byte retention: when a configured deployment receives oversized text, it replaces that text with a bounded preview plus an opaque locator and retrieval hint while preserving the complete result in its spill store. The session-query consumer neither imports `ctx.spillStore` nor implements a second truncation format. -Session-level results include the latest folded title when available. Absence is rendered as untitled; an operational title read failure preserves the base result, renders an unavailable marker, and logs the underlying error, while an authorization mismatch fails closed. Search results include the strongest matching event and provider excerpt, traces include complete authorized relationships, and event reads keep neighbor presentation readable while reserving exact JSON for the requested target. +Session-level results include the latest folded title when available. Each tool execution batches its unique title ids through one live-preferred corpus observation with at most four persisted-inspection workers and passes the exact tool-execution signal through persisted listing and inspection. Live sources fold directly; each persisted worker folds its completed source to a detached header/title observation and releases the full log before dequeuing another id, so the batch retains only small projected values. For the search tools, the execution signal carries the configured search deadline. Cancellation starts no queued title inspections and rejects the complete tool execution after already-started inspections settle; a missing, malformed, or operationally failed title remains isolated to that id, preserves the base result, renders an unavailable marker, and logs the underlying error, while an authorization mismatch fails closed. Search results include the strongest matching event and provider excerpt, traces include complete authorized relationships, and event reads keep neighbor presentation readable while reserving exact JSON for the requested target. ## Host composition @@ -44,7 +44,7 @@ The shipped ACP, TUI, and Web compositions all mount the consumer beside `ctx.se ## Verification -Package tests pin argument validation, filter translation, timestamp normalization, exact-workspace authorization, changed-observation rejection, missing-identity behavior, hidden-boundary pruning, current-step exclusion, internal provider paging, count caps, cancellation, title fallbacks, representative search/trace/read rendering, generic presentation, and disposable registration. Integration coverage uses the real SQLite FTS provider over live and persisted sessions. Loader and assembled-host coverage proves that ACP, TUI, and Web register the tools with timeout and spill support, while keyless assembled ACP snapshots pin the prompt guidance and schemas plus path-independent exact event-read spill and retention behavior. +Package tests pin argument validation, filter translation, timestamp normalization, exact-workspace authorization, changed-observation rejection, missing-identity behavior, hidden-boundary pruning, current-step exclusion, internal provider paging, count caps, cancellation, one-scan bounded batch title enrichment, projection-before-dequeue ordering, queued-work suppression, started-worker quiescence, per-header validation, title fallbacks, representative search/trace/read rendering, generic presentation, and disposable registration. Integration coverage uses the real SQLite FTS provider over live and persisted sessions. Loader and assembled-host coverage proves that ACP, TUI, and Web register the tools with timeout and spill support, while keyless assembled ACP snapshots pin the prompt guidance and schemas plus path-independent exact event-read spill and retention behavior. ## Consequences diff --git a/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.zh.md b/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.zh.md index dfeae26a54..f82c0fac52 100644 --- a/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.zh.md +++ b/.agents/notes/implemented/feature/2026-07-24-model-facing-session-query-tools.zh.md @@ -28,7 +28,7 @@ Status: implemented 追踪与读取工具同样不公开谱系分页或字符分页。规范结果采用纯文本，并在服务已有的事件窗口与搜索数量资源边界内保持完整。通用的 `tools/post-execute` spill 策略负责行内字节保留：当已配置的部署收到过大的文本时，该策略会用有界预览、不可透明推导的定位符与读取提示替换文本，同时在 spill 存储中保留完整结果。会话查询消费者既不导入 `ctx.spillStore`，也不实现第二套截断格式。 -会话级结果在可用时包含最新折叠标题。没有标题时渲染为未命名；标题读取发生操作性失败时保留基础结果、渲染不可用标记并记录底层错误，而授权不匹配则按失败关闭处理。搜索结果包含最强匹配事件与提供方摘录，追踪包含完整的已授权关系，事件读取保持邻近事件表现易读，同时只为被请求的目标保留精确 JSON。 +会话级结果在可用时包含最新折叠标题。每次工具执行都会通过一次优先使用实时数据的语料观测批量读取唯一标题 id，最多使用 4 个持久化检查 worker，并将准确的工具执行信号传递给持久化列表与检查操作。实时来源会直接折叠；每个持久化 worker 都会把已完成的来源折叠为分离的会话头／标题观测，并在取出下一个 id 前释放完整日志，因此批次只保留小型投影值。对于搜索工具，该执行信号携带已配置的搜索截止时间。取消不会启动排队中的标题检查，并会在已经启动的检查全部完成后拒绝完整的工具执行；标题缺失、格式错误或发生操作性失败时，错误只影响对应 id，同时保留基础结果、渲染不可用标记并记录底层错误，而授权不匹配则按失败关闭处理。搜索结果包含最强匹配事件与提供方摘录，追踪包含完整的已授权关系，事件读取保持邻近事件表现易读，同时只为被请求的目标保留精确 JSON。 ## 宿主组合 @@ -44,7 +44,7 @@ Status: implemented ## 验证 -包级测试固定参数校验、过滤条件转换、时间戳规范化、精确工作区授权、变更观测拒绝、身份缺失行为、隐藏边界裁剪、当前步骤排除、内部提供方翻页、数量上限、取消、标题回退、代表性搜索／追踪／读取渲染、通用表现与可释放注册。集成覆盖使用真实 SQLite FTS 提供方查询实时与持久化会话。Loader 与组装宿主覆盖证明 ACP、TUI 和 Web 会注册带超时及 spill 支持的工具；无密钥组装 ACP 快照则固定提示词指导与 schema，以及与路径无关的精确事件读取 spill 与保留行为。 +包级测试固定参数校验、过滤条件转换、时间戳规范化、精确工作区授权、变更观测拒绝、身份缺失行为、隐藏边界裁剪、当前步骤排除、内部提供方翻页、数量上限、取消、单次扫描且并发有界的批量标题扩充、先投影再取出下一个任务的顺序、抑制排队工作、等待已启动 worker 静止、逐会话头校验、标题回退、代表性搜索／追踪／读取渲染、通用表现与可释放注册。集成覆盖使用真实 SQLite FTS 提供方查询实时与持久化会话。Loader 与组装宿主覆盖证明 ACP、TUI 和 Web 会注册带超时及 spill 支持的工具；无密钥组装 ACP 快照则固定提示词指导与 schema，以及与路径无关的精确事件读取 spill 与保留行为。 ## 后果 diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index de4296b85e..c2669885a4 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -941,15 +941,17 @@ abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEven * This read is serialized with writes for the same id and returns detached * values, so observers cannot mutate backend-owned state. * @param id - the persisted session to inspect. + * @param signal - optional cancellation for queued and backend read work. * @returns the header and valid stored event prefix exactly as observed. */ -abstract inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> +abstract inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> /** * Lightweight listing from metadata, without a full-log parse. + * @param signal - optional cancellation for backend listing work. * @returns one header per materialized session. */ -abstract list(): Promise +abstract list(signal?: AbortSignal): Promise /** * List materialized sessions with cheap per-log change tokens. @@ -1014,16 +1016,29 @@ async filterSessions(filters: readonly SessionResultFilter[]): Promise +async readTitle( sessionId: SessionId, signal?: AbortSignal, ): Promise /** * Fold the latest title and return its source header from one corpus observation. * @param sessionId - live or persisted session id to read. + * @param signal - optional cancellation for source resolution and title folding. * @returns cloned source header and optional latest title snapshot. */ -async readTitleSnapshot(sessionId: SessionId): Promise +async readTitleSnapshot( sessionId: SessionId, signal?: AbortSignal, ): Promise + +/** + * Fold titles for unique sessions from one cancellable corpus observation. + * + * Results preserve first-occurrence input order. Operational failures stay + * isolated per session, while cancellation rejects the complete operation. + * @param sessionIds - live or persisted session ids to observe. + * @param signal - optional cancellation shared by all source reads. + * @returns one fulfilled or rejected result per unique requested id. + */ +async readTitleSnapshots( sessionIds: readonly SessionId[], signal?: AbortSignal, ): Promise /** * List lightweight raw-log event records for one logical session. @@ -1072,9 +1087,9 @@ async traceEvent(request: SessionEventTraceRequest): Promise ``` -Types: [SessionEventReadRequest](../core-data-structures/session-query.md) · [SessionEventRecord](../core-data-structures/session-query.md) · [SessionEventResultFilter](../core-data-structures/session-query.md) · [SessionEventSearchDocument](../core-data-structures/session-query.md) · [SessionEventSearchPage](../core-data-structures/session-query.md) · [SessionEventSearchRequest](../core-data-structures/session-query.md) · [SessionEventTraceObservation](../core-data-structures/session-query.md) · [SessionEventTraceRequest](../core-data-structures/session-query.md) · [SessionEventWindow](../core-data-structures/session-query.md) · [SessionId](../core-data-structures/core.md) · [SessionLineageTrace](../core-data-structures/session-query.md) · [SessionLogSnapshot](../core-data-structures/session-query.md) · [SessionRecord](../core-data-structures/session-query.md) · [SessionResultFilter](../core-data-structures/session-query.md) · [SessionSearchExecContext](../core-data-structures/session-query.md) · [SessionSearchHit](../core-data-structures/session-query.md) · [SessionSearchPage](../core-data-structures/session-query.md) · [SessionSearchRequest](../core-data-structures/session-query.md) · [SessionSurfaceSnapshot](../core-data-structures/session-query.md) · [SessionTitleObservation](../core-data-structures/session-query.md) · [SessionTitleSnapshot](../core-data-structures/session-title.md) +Types: [SessionEventReadRequest](../core-data-structures/session-query.md) · [SessionEventRecord](../core-data-structures/session-query.md) · [SessionEventResultFilter](../core-data-structures/session-query.md) · [SessionEventSearchDocument](../core-data-structures/session-query.md) · [SessionEventSearchPage](../core-data-structures/session-query.md) · [SessionEventSearchRequest](../core-data-structures/session-query.md) · [SessionEventTraceObservation](../core-data-structures/session-query.md) · [SessionEventTraceRequest](../core-data-structures/session-query.md) · [SessionEventWindow](../core-data-structures/session-query.md) · [SessionId](../core-data-structures/core.md) · [SessionLineageTrace](../core-data-structures/session-query.md) · [SessionLogSnapshot](../core-data-structures/session-query.md) · [SessionRecord](../core-data-structures/session-query.md) · [SessionResultFilter](../core-data-structures/session-query.md) · [SessionSearchExecContext](../core-data-structures/session-query.md) · [SessionSearchHit](../core-data-structures/session-query.md) · [SessionSearchPage](../core-data-structures/session-query.md) · [SessionSearchRequest](../core-data-structures/session-query.md) · [SessionSurfaceSnapshot](../core-data-structures/session-query.md) · [SessionTitleObservation](../core-data-structures/session-query.md) · [SessionTitleObservationResult](../core-data-structures/session-query.md) · [SessionTitleSnapshot](../core-data-structures/session-title.md) -Source: [`packages/session-query/session-query/src/index.ts:75`](../../packages/session-query/session-query/src/index.ts) +Source: [`packages/session-query/session-query/src/index.ts:76`](../../packages/session-query/session-query/src/index.ts) ## `ctx.sessionReferences` — `SessionReferenceService` diff --git a/docs/core-data-structures/session-query.md b/docs/core-data-structures/session-query.md index 265b49d3a2..4886eb3087 100644 --- a/docs/core-data-structures/session-query.md +++ b/docs/core-data-structures/session-query.md @@ -49,7 +49,7 @@ interface SessionSurfaceSnapshot { } ``` -`SessionTitleObservation` applies the same atomic-observation rule to title folding, so an authorization consumer can validate the source header that supplied the title. +`SessionTitleObservation` applies the same atomic-observation rule to title folding, so an authorization consumer can validate the source header that supplied the title. Batch reads return one ordered `SessionTitleObservationResult` per unique requested id: operational failures remain local to that id, while cancellation rejects the complete operation. ```ts type-equiv /** Latest folded title bound to the same session-header observation. */ @@ -61,6 +61,27 @@ interface SessionTitleObservation { } ``` +```ts type-equiv +/** One ordered result from a batch title observation. */ +type SessionTitleObservationResult = + | { + /** Requested session id. */ + sessionId: SessionId + /** Successful atomic header/title observation. */ + status: 'fulfilled' + /** Header and optional latest title from one logical source. */ + value: SessionTitleObservation + } + | { + /** Requested session id. */ + sessionId: SessionId + /** Operational failure isolated to this session. */ + status: 'rejected' + /** Original failure from logical-source resolution or title folding. */ + reason: unknown + } +``` + ```ts type-equiv /** Lightweight metadata for one event within a logical session. */ interface SessionEventRecord { diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index 43723acfa8..72785f4e69 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -469,12 +469,12 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ jsDoc: '/**\n * Load a header and balanced contiguous log. A complete interrupted final\n * turn is preserved and durably closed with missing tool errors plus any open\n * step and turn boundaries; only a torn final record is discarded. Unknown\n * versions and corruption in the committed prefix reject. Implementations\n * MUST NOT crash-repair an identity still bound to a live Session: a balanced\n * live log may return with its stored header as a durable snapshot, while an\n * open live turn rejects.\n * A coordinator-backed cold load reserves the identity across storage awaits,\n * so concurrent publication of a same-id live Session rejects.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */', }, { - signature: 'abstract inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', - jsDoc: '/**\n * Inspect a header and its valid contiguous stored prefix without repairing\n * a torn tail, closing an interrupted turn, or publishing coordinator state.\n * This read is serialized with writes for the same id and returns detached\n * values, so observers cannot mutate backend-owned state.\n * @param id - the persisted session to inspect.\n * @returns the header and valid stored event prefix exactly as observed.\n */', + signature: 'abstract inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', + jsDoc: '/**\n * Inspect a header and its valid contiguous stored prefix without repairing\n * a torn tail, closing an interrupted turn, or publishing coordinator state.\n * This read is serialized with writes for the same id and returns detached\n * values, so observers cannot mutate backend-owned state.\n * @param id - the persisted session to inspect.\n * @param signal - optional cancellation for queued and backend read work.\n * @returns the header and valid stored event prefix exactly as observed.\n */', }, { - signature: 'abstract list(): Promise', - jsDoc: '/**\n * Lightweight listing from metadata, without a full-log parse.\n * @returns one header per materialized session.\n */', + signature: 'abstract list(signal?: AbortSignal): Promise', + jsDoc: '/**\n * Lightweight listing from metadata, without a full-log parse.\n * @param signal - optional cancellation for backend listing work.\n * @returns one header per materialized session.\n */', }, { signature: 'abstract listSnapshots(): Promise', @@ -507,12 +507,16 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ jsDoc: '/**\n * Filter the complete logical corpus with provider-independent predicates.\n * @param filters - ANDed session metadata and availability clauses.\n * @returns matching cloned records in deterministic newest-first order.\n */', }, { - signature: 'async readTitle(sessionId: SessionId): Promise', - jsDoc: '/**\n * Fold the latest log-backed title from one live-preferred logical session.\n * @param sessionId - live or persisted session id to read.\n * @returns latest title snapshot, or `undefined` when the log has no title event.\n */', + signature: 'async readTitle( sessionId: SessionId, signal?: AbortSignal, ): Promise', + jsDoc: '/**\n * Fold the latest log-backed title from one live-preferred logical session.\n * @param sessionId - live or persisted session id to read.\n * @param signal - optional cancellation for source resolution and title folding.\n * @returns latest title snapshot, or `undefined` when the log has no title event.\n */', }, { - signature: 'async readTitleSnapshot(sessionId: SessionId): Promise', - jsDoc: '/**\n * Fold the latest title and return its source header from one corpus observation.\n * @param sessionId - live or persisted session id to read.\n * @returns cloned source header and optional latest title snapshot.\n */', + signature: 'async readTitleSnapshot( sessionId: SessionId, signal?: AbortSignal, ): Promise', + jsDoc: '/**\n * Fold the latest title and return its source header from one corpus observation.\n * @param sessionId - live or persisted session id to read.\n * @param signal - optional cancellation for source resolution and title folding.\n * @returns cloned source header and optional latest title snapshot.\n */', + }, + { + signature: 'async readTitleSnapshots( sessionIds: readonly SessionId[], signal?: AbortSignal, ): Promise', + jsDoc: '/**\n * Fold titles for unique sessions from one cancellable corpus observation.\n *\n * Results preserve first-occurrence input order. Operational failures stay\n * isolated per session, while cancellation rejects the complete operation.\n * @param sessionIds - live or persisted session ids to observe.\n * @param signal - optional cancellation shared by all source reads.\n * @returns one fulfilled or rejected result per unique requested id.\n */', }, { signature: 'async listEvents(sessionId: SessionId): Promise', @@ -1897,6 +1901,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [ name: 'SessionTitleObservation', declaration: 'export interface SessionTitleObservation {\n session: SessionHeader;\n title?: SessionTitleSnapshot;\n}', }, + { + name: 'SessionTitleObservationResult', + declaration: 'export type SessionTitleObservationResult = {\n sessionId: SessionId;\n status: \'fulfilled\';\n value: SessionTitleObservation;\n} | {\n sessionId: SessionId;\n status: \'rejected\';\n reason: unknown;\n};', + }, { name: 'SessionTitleProvider', declaration: 'export interface SessionTitleProvider {\n readonly id: SessionTitleProviderId;\n readonly automatic: SessionTitleAutomaticMode;\n generate(request: SessionTitleProviderRequest): Promise;\n}', diff --git a/packages/session-persistence/session-persistence-jsonl/src/index.ts b/packages/session-persistence/session-persistence-jsonl/src/index.ts index 629c0e3ff1..6b2fe3d0cf 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/index.ts @@ -131,8 +131,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi return this.coordinator.load(id) } - inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - return this.coordinator.inspect(id) + inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + return this.coordinator.inspect(id, signal) } // One method serves both public `list` and the backend hook; delegating it to @@ -142,24 +142,33 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi // --- PersistenceBackend hooks (the file-bytes storage primitives) --- /** Read a stored prefix by id across all cwd buckets when cwd is unknown. */ - async loadStored(id: SessionId): Promise | undefined> { + async loadStored(id: SessionId, signal?: AbortSignal): Promise | undefined> { + signal?.throwIfAborted() await this.ensureRootEncoding() - const path = await this.findLog(id) + signal?.throwIfAborted() + const path = await this.findLog(id, signal) if (path === undefined) return undefined - return this.readPrefix(path, id) + return this.readPrefix(path, id, signal) } /** * Read a stored prefix and convert torn-tail state to the opaque marker the * coordinator can round-trip without knowing the physical encoding. */ - private async readPrefix(path: string, expectedId?: SessionId): Promise> { - const buffer = await readFile(path) + private async readPrefix( + path: string, + expectedId?: SessionId, + signal?: AbortSignal, + ): Promise> { + const buffer = await readFile(path, { signal }) + signal?.throwIfAborted() let prefix: StoredPrefix if (this.compression === 'zstd') { - prefix = await this.readZstdPrefix(buffer) + prefix = await this.readZstdPrefix(buffer, signal) } else { + signal?.throwIfAborted() const { meta, events, committedBytes } = scanLog(buffer) + signal?.throwIfAborted() prefix = { meta, events, @@ -168,30 +177,45 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi : {}, } } + signal?.throwIfAborted() this.assertStoredIdentity(path, prefix.meta, expectedId) return prefix } /** Decode complete frames and retain complete JSONL records from a torn final frame. */ - private async readZstdPrefix(buffer: Buffer): Promise> { + private async readZstdPrefix( + buffer: Buffer, + signal?: AbortSignal, + ): Promise> { + signal?.throwIfAborted() const { frames, tornStart } = scanZstdFrames(buffer) + signal?.throwIfAborted() if (frames.length === 0) throw new Error('empty or header-less Zstandard session log') const plaintextFrames: Buffer[] = [] for (const frame of frames) { + let plaintext: Buffer try { - plaintextFrames.push(await decompressZstdFrame(buffer.subarray(frame.start, frame.end))) + signal?.throwIfAborted() + plaintext = await decompressZstdFrame(buffer.subarray(frame.start, frame.end)) } catch (error) { + /* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */ + if (signal?.aborted) signal.throwIfAborted() throw new Error(`corrupt Zstandard session log: frame at byte ${frame.start} failed validation`, { cause: error }) } + signal?.throwIfAborted() + plaintextFrames.push(plaintext) } const headerFrame = plaintextFrames[0] if (headerFrame === undefined || headerFrame.length === 0 || headerFrame.indexOf(0x0A) !== headerFrame.length - 1) { throw new Error('corrupt Zstandard session log: first frame is not exactly one header line') } + signal?.throwIfAborted() const completePlaintext = Buffer.concat(plaintextFrames) + signal?.throwIfAborted() const completePrefix = scanLog(completePlaintext) + signal?.throwIfAborted() if (completePrefix.committedBytes !== completePlaintext.length) { throw new Error('corrupt Zstandard session log: complete frame contains a torn JSONL record') } @@ -201,12 +225,17 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi let recoveredPlaintext: Buffer = Buffer.alloc(0) try { + signal?.throwIfAborted() recoveredPlaintext = await decompressZstdFrame(buffer.subarray(tornStart)) } catch { + /* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */ + if (signal?.aborted) signal.throwIfAborted() // A structurally incomplete final frame may end before Node's decoder can // emit any plaintext; the complete prior frames remain recoverable. } + signal?.throwIfAborted() const recoveredPrefix = scanLog(Buffer.concat([completePlaintext, recoveredPlaintext])) + signal?.throwIfAborted() /* v8 ignore next 3 -- appending plaintext cannot shorten the already-scanned complete prefix */ if (recoveredPrefix.events.length < completePrefix.events.length) { throw new Error('corrupt Zstandard session log: recovered prefix does not extend complete frames') @@ -247,8 +276,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi } /** List valid unique stored sessions' metadata (header line only — no full-log parse). */ - async list(): Promise { - return (await this.listArtifacts()).map(artifact => artifact.header) + async list(signal?: AbortSignal): Promise { + return (await this.listArtifacts(signal)).map(artifact => artifact.header) } /** List metadata plus a stat-derived identity for each append-only log. */ @@ -274,17 +303,21 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi return snapshots } - private async listArtifacts(): Promise> { + private async listArtifacts(signal?: AbortSignal): Promise> { + signal?.throwIfAborted() await this.ensureRootEncoding() + signal?.throwIfAborted() const artifacts: Array<{ header: SessionHeader; path: string }> = [] const ids = new Set() - for (const dir of await this.listCwdDirs()) { - for (const name of await this.listArtifactNames(dir)) { + for (const dir of await this.listCwdDirs(signal)) { + for (const name of await this.listArtifactNames(dir, signal)) { + signal?.throwIfAborted() const path = join(dir, name) // Read only headers so listing scales with session count, not log size. const first = this.compression === 'zstd' - ? await this.readFirstZstdLine(path) - : await this.readFirstLine(path) + ? await this.readFirstZstdLine(path, signal) + : await this.readFirstLine(path, signal) + signal?.throwIfAborted() if (first === undefined) continue // empty/half-written file const meta = parseHeaderMeta(first) if (meta === undefined) continue // not a session header @@ -492,18 +525,23 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi * file. Returns undefined if the file is empty or has no complete first line. * Reads in bounded chunks so a huge log costs only the header read. */ - private async readFirstLine(path: string): Promise { + private async readFirstLine(path: string, signal?: AbortSignal): Promise { + signal?.throwIfAborted() const handle = await open(path, 'r') try { + signal?.throwIfAborted() const chunks: Buffer[] = [] const buf = Buffer.alloc(8192) for (;;) { + signal?.throwIfAborted() const { bytesRead } = await handle.read(buf, 0, buf.length, null) + signal?.throwIfAborted() if (bytesRead === 0) return undefined // EOF with no newline → no complete line const slice = buf.subarray(0, bytesRead) const nl = slice.indexOf(0x0a) if (nl !== -1) { chunks.push(slice.subarray(0, nl)) + signal?.throwIfAborted() return Buffer.concat(chunks).toString('utf8') } chunks.push(Buffer.from(slice)) @@ -514,23 +552,34 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi } /** Read and validate only the independently compressed header frame. */ - private async readFirstZstdLine(path: string): Promise { + private async readFirstZstdLine(path: string, signal?: AbortSignal): Promise { + signal?.throwIfAborted() const handle = await open(path, 'r') try { + signal?.throwIfAborted() let content = Buffer.alloc(0) const chunk = Buffer.alloc(8192) for (;;) { + signal?.throwIfAborted() const { bytesRead } = await handle.read(chunk, 0, chunk.length, null) + signal?.throwIfAborted() if (bytesRead === 0) return undefined + signal?.throwIfAborted() content = Buffer.concat([content, chunk.subarray(0, bytesRead)]) + signal?.throwIfAborted() const first = scanZstdFrames(content, 1).frames[0] + signal?.throwIfAborted() if (first === undefined) continue let plaintext: Buffer try { + signal?.throwIfAborted() plaintext = await decompressZstdFrame(content.subarray(first.start, first.end)) } catch (error) { + /* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */ + if (signal?.aborted) signal.throwIfAborted() throw new Error('corrupt Zstandard session log: header frame failed validation', { cause: error }) } + signal?.throwIfAborted() if (plaintext.length === 0 || plaintext.indexOf(0x0A) !== plaintext.length - 1) { throw new Error('corrupt Zstandard session log: first frame is not exactly one header line') } @@ -542,11 +591,12 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi } /** Find the unique physical log for an id across every cwd bucket. */ - private async findLog(id: SessionId): Promise { + private async findLog(id: SessionId, signal?: AbortSignal): Promise { const target = encodeSegment(id) + logSuffix(this.compression) const oppositeTarget = encodeSegment(id) + logSuffix(this.oppositeCompression()) const matches: string[] = [] - for (const dir of await this.listCwdDirs()) { + for (const dir of await this.listCwdDirs(signal)) { + signal?.throwIfAborted() const path = join(dir, target) const opposite = join(dir, oppositeTarget) if (await this.exists(opposite)) throw this.encodingMismatch(opposite) @@ -585,9 +635,11 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi } /** The cwd-bucket directories under the root (absolute paths). */ - private async listCwdDirs(): Promise { + private async listCwdDirs(signal?: AbortSignal): Promise { try { + signal?.throwIfAborted() const entries = await readdir(this.root, { withFileTypes: true }) + signal?.throwIfAborted() return entries.filter(e => e.isDirectory()).map(e => join(this.root, e.name)) } catch (error) { // Only an absent root means no sessions; rethrow every other I/O failure. @@ -596,8 +648,10 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi } } - private async listArtifactNames(dir: string): Promise { + private async listArtifactNames(dir: string, signal?: AbortSignal): Promise { + signal?.throwIfAborted() const entries = await readdir(dir) + signal?.throwIfAborted() const oppositeSuffix = logSuffix(this.oppositeCompression()) const incompatible = entries.find(name => name.endsWith(oppositeSuffix)) if (incompatible !== undefined) throw this.encodingMismatch(`${dir}/${incompatible}`) diff --git a/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts index fcadac1f04..2777281481 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts @@ -16,6 +16,18 @@ const MAGIC = Buffer.from([0x28, 0xB5, 0x2F, 0xFD]) const roots: string[] = [] const contexts: Context[] = [] +interface ZstdReaderInternals { + readZstdPrefix(buffer: Buffer, signal?: AbortSignal): Promise +} + +type HeaderRead = ( + this: FileHandle, + buffer: Buffer, + offset: number, + length: number, + position: number | null, +) => Promise<{ bytesRead: number; buffer: Buffer }> + async function freshRoot(prefix = 'dsh-jsonl-zstd-'): Promise { const root = await mkdtemp(join(tmpdir(), prefix)) roots.push(root) @@ -275,6 +287,65 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => { await expect(ctx.sessionPersistence.load(header.id)).rejects.toThrow(/frame at byte .* failed validation/) }) + it('stops multi-frame inspection after cancellation interrupts the active decode', async () => { + const root = await freshRoot() + const ctx = await mount(root) + const header = meta('cancel-zstd-frames') + const headerFrame = await compressZstdFrame(`${JSON.stringify(toHeaderLine(header))}\n`) + const eventFrame = await compressZstdFrame(`${JSON.stringify(oneTurnLog()[0])}\n`) + const laterFrame = await compressZstdFrame(`${JSON.stringify(oneTurnLog()[1])}\n`) + const stream = Buffer.concat([headerFrame, eventFrame, laterFrame]) + expect(scanZstdFrames(stream).frames).toHaveLength(3) + const controller = new AbortController() + const reason = new Error('cancel after Zstandard decode starts') + const reader = ctx.sessionPersistence as unknown as ZstdReaderInternals + const zstdModule = await import('../src/zstd.ts') + const decode = vi.spyOn(zstdModule, 'decompressZstdFrame') + + // readZstdPrefix reaches its first asynchronous decompression before it + // returns this promise. The microtask abort therefore occurs after decode + // starts and must prevent every later frame from reaching the decoder. + const pending = reader.readZstdPrefix(stream, controller.signal) + queueMicrotask(() => { controller.abort(reason) }) + + await expect(pending).rejects.toBe(reason) + expect(decode).toHaveBeenCalledTimes(1) + expect(decode).toHaveBeenCalledWith(headerFrame) + }) + + it.each(['none', 'zstd'] as const)( + 'observes cancellation after each async %s header read during listing', + async (compression) => { + const root = await freshRoot() + const ctx = await mount(root, compression) + const header = meta(`cancel-${compression}-header-read`, '/work') + await ctx.sessionPersistence.create(header) + await ctx.sessionPersistence.append(header.id, oneTurnLog()) + await ctx.sessionPersistence.list() + const path = logPath(root, header.cwd, header.id, compression) + const probe = await open(path, 'r') + const prototype = Object.getPrototypeOf(probe) as { read: HeaderRead } + const originalRead = prototype.read + await probe.close() + const controller = new AbortController() + const reason = new Error(`cancel ${compression} header read`) + const read = vi.spyOn(prototype, 'read').mockImplementation(async function ( + this: FileHandle, + buffer: Buffer, + offset: number, + length: number, + position: number | null, + ) { + const result = await originalRead.call(this, buffer, offset, length, position) + controller.abort(reason) + return result + }) + + await expect(ctx.sessionPersistence.list(controller.signal)).rejects.toBe(reason) + expect(read).toHaveBeenCalledTimes(1) + }, + ) + it('preserves complete records from a torn frame and re-encodes them with crash closers', async () => { const root = await freshRoot() const ctx = await mount(root) diff --git a/packages/session-persistence/session-persistence-sqlite/src/index.ts b/packages/session-persistence/session-persistence-sqlite/src/index.ts index 5804c18282..0c1159f139 100644 --- a/packages/session-persistence/session-persistence-sqlite/src/index.ts +++ b/packages/session-persistence/session-persistence-sqlite/src/index.ts @@ -157,8 +157,8 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers return this.coordinator.load(id) } - inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - return this.coordinator.inspect(id) + inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + return this.coordinator.inspect(id, signal) } // One method serves both public `list` and the backend hook; delegating it to @@ -167,8 +167,8 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers // --- PersistenceBackend hooks (the SQLite storage primitives) --- /** Read a stored prefix by id (ids are globally unique — no scope to scan). */ - loadStored(id: SessionId): Promise | undefined> { - return this.readPrefix(id) + loadStored(id: SessionId, signal?: AbortSignal): Promise | undefined> { + return this.readPrefix(id, signal) } /** @@ -176,14 +176,17 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers * torn-tail marker is the seq from which a never-committed tail must be deleted * (`scanRows` already returns it as `number | undefined`). */ - private async readPrefix(id: SessionId): Promise | undefined> { + private async readPrefix(id: SessionId, signal?: AbortSignal): Promise | undefined> { + signal?.throwIfAborted() await this.ready + signal?.throwIfAborted() const row = this.rowFor(id) if (row === undefined) return undefined const meta = rowToMeta(row) const eventRows = this.db .prepare('SELECT seq, type, time, data, source_event_seqs, surface_op FROM events WHERE session_id = ? ORDER BY seq') .all(id) as unknown as EventRow[] + signal?.throwIfAborted() const { preserved, tornFrom } = scanRows(eventRows) return { meta, events: preserved, ...tornFrom !== undefined ? { tornMarker: tornFrom } : {} } } @@ -251,11 +254,14 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers } /** List all materialized sessions' metadata (every row is a materialized session). */ - async list(): Promise { + async list(signal?: AbortSignal): Promise { + signal?.throwIfAborted() await this.ready + signal?.throwIfAborted() const rows = this.db .prepare('SELECT * FROM sessions') .all() as unknown as SessionRow[] + signal?.throwIfAborted() return rows.map(rowToMeta) } diff --git a/packages/session-persistence/session-persistence/README.md b/packages/session-persistence/session-persistence/README.md index 25429bd720..fa734fb736 100644 --- a/packages/session-persistence/session-persistence/README.md +++ b/packages/session-persistence/session-persistence/README.md @@ -12,8 +12,8 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l | `create(meta): Promise` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). | | `append(id, events): Promise` | Durably persist a batch. Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. | | `load(id): Promise<{ meta; events }>` | Return a stored header plus a balanced contiguous log. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption and unknown `version` reject. | -| `inspect(id): Promise<{ meta; events }>` | Return a detached valid stored prefix without truncating a torn tail, synthesizing recovery closers, or publishing coordinator state. Serialized with same-id writes; intended for read models and other observers that must never recover a log. | -| `list(): Promise` | Lightweight listing from metadata, no full-log parse. A zero-event lazily-materialized session is absent from `list`. | +| `inspect(id, signal?): Promise<{ meta; events }>` | Return a detached valid stored prefix without truncating a torn tail, synthesizing recovery closers, or publishing coordinator state. Serialized with same-id writes; the optional signal promptly rejects a queued caller, prevents that queued backend read from starting, and cancels active backend read work. Intended for read models and other observers that must never recover a log. | +| `list(signal?): Promise` | Lightweight listing from metadata, no full-log parse. The optional signal cancels backend listing work. A zero-event lazily-materialized session is absent from `list`. | | `listSnapshots(): Promise` | Lightweight metadata plus an opaque branded per-log revision, without loading event logs. A revision stays equal while that log and its backing store are unchanged, changes after append or mutating load repair, and cannot collide solely because two stores use the same local counter. | ## Invariants every backend must honor @@ -40,10 +40,10 @@ The `PersistenceBackend` hooks (the only seam between the coordinato | Hook | Role | |---|---| | `name` | Backend label for the dispose-failure `AggregateError`. | -| `loadStored(id)` | Read a stored prefix by id across every storage scope. Used by resume/load, non-mutating inspect, live adoption, and the create-collision probe. Returned metadata identifies `id`; an opaque `tornMarker` is present iff a torn tail must be truncated. | +| `loadStored(id, signal?)` | Read a stored prefix by id across every storage scope. Used by resume/load, non-mutating inspect, live adoption, and the create-collision probe. The optional signal belongs to observation-only reads. Returned metadata identifies `id`; an opaque `tornMarker` is present iff a torn tail must be truncated. | | `appendBatch(meta, events, isMaterialized)` | Durably append a contiguous batch, lazily materializing ATOMICALLY when not yet materialized. | | `commitRepair(meta, tornMarker, closers)` | Make a crash repair durable: truncate the torn tail (iff `tornMarker !== undefined` — a marker may be falsy, e.g. seq/offset `0`) and append `closers`. NOT required to be atomic. Used by load (truncate + closers) and live-adoption (truncate only). | -| `list()` | List all stored metadata. | +| `list(signal?)` | List all stored metadata, observing optional cancellation. | | `close?()` | Optional lifecycle teardown (e.g. close a db handle), awaited after the dispose drain. | The coordinator asserts the stored id and compares stored/live cwd before repair or live adoption. Its `inspect()` path validates and clones the prefix without calling `commitRepair` or publishing write state. The `tornMarker` is fully OPAQUE: the coordinator only tests `!== undefined` and round-trips it to `commitRepair`, never inspecting its value (the JSONL backend uses the byte offset to truncate to, the SQLite backend the seq to delete from). A third-party backend MAY implement the abstract service directly without the coordinator, but it must provide the same non-mutating inspection and trustworthy lightweight snapshot revisions. See [the write-coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md). diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index fb46aa4877..442011bd2c 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -40,8 +40,10 @@ export interface PersistenceBackend { * `id` before repair or state publication. Used by resume/load, live adoption, * and — via `!== undefined` — the create-collision probe. The returned * `tornMarker` is present iff there is a torn tail to truncate. + * @param id - persisted session id to resolve. + * @param signal - optional cancellation for backend read work. */ - loadStored(id: SessionId): Promise | undefined> + loadStored(id: SessionId, signal?: AbortSignal): Promise | undefined> /** * Durably append a CONTIGUOUS batch, lazily materializing the session first @@ -60,8 +62,11 @@ export interface PersistenceBackend { */ commitRepair(meta: SessionHeader, tornMarker: TornMarker | undefined, closers: readonly SessionEvent[]): Promise - /** List all stored (materialized) sessions' metadata. */ - list(): Promise + /** + * List all stored (materialized) sessions' metadata. + * @param signal - optional cancellation for backend listing work. + */ + list(signal?: AbortSignal): Promise /** * Optional lifecycle teardown (e.g. close a database handle). Awaited by the @@ -267,14 +272,26 @@ export class PersistenceCoordinator { * Read a detached valid stored prefix without recovery mutations or * coordinator-state publication. * @param id - persisted session to inspect. + * @param signal - optional cancellation for queued and backend read work. * @returns stored header and events before any synthetic recovery closers. */ - inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - return this.serialize(id, () => this.inspectCore(id)) + inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + return this.serialize(id, () => this.inspectCore(id, signal), signal) } - private async inspectCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - const stored = await this.backend.loadStored(id) + private async inspectCore( + id: SessionId, + signal?: AbortSignal, + ): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + signal?.throwIfAborted() + let stored: StoredPrefix | undefined + try { + stored = await this.backend.loadStored(id, signal) + } catch (error: unknown) { + if (signal?.aborted) signal.throwIfAborted() + throw error + } + signal?.throwIfAborted() if (stored === undefined) throw new Error(`session "${id}" not found`) this.assertStoredId(id, stored.meta) this.assertVersion(stored.meta) @@ -331,9 +348,19 @@ export class PersistenceCoordinator { * public methods must NOT call each other (deadlock); they call the unserialized * `*Core` helpers instead. */ - private serialize(id: SessionId, op: () => Promise | T): Promise { + private serialize( + id: SessionId, + op: () => Promise | T, + signal?: AbortSignal, + ): Promise { const prior = this.chains.get(id) ?? Promise.resolve() - const next = prior.then(op, op) + let started = false + const run = (): Promise | T => { + signal?.throwIfAborted() + started = true + return op() + } + const next = prior.then(run, run) // Keep the chain alive but swallow this op's rejection for the NEXT waiter // (the caller still sees the real rejection via `next`). const tail = next.then(() => undefined, () => undefined) @@ -343,7 +370,7 @@ export class PersistenceCoordinator { void tail.then(() => { if (this.chains.get(id) === tail) this.chains.delete(id) }) - return next + return signal === undefined ? next : observeQueuedAbort(next, signal, () => started) } /** Build a state for a session discovered in storage but not yet in memory. */ @@ -615,3 +642,50 @@ export class PersistenceCoordinator { live.pending.splice(0, batch.length) } } + +/** + * Give an observation caller a prompt cancellation view of queued work. + * + * The serialized `operation` remains in the same-id chain and checks the signal + * before invoking backend work. Observing its settlement here therefore cannot + * detach a storage read or let a later operation overtake its predecessor. + */ +function observeQueuedAbort( + operation: Promise, + signal: AbortSignal, + started: () => boolean, +): Promise { + return new Promise((resolve, reject) => { + let settled = false + const finish = (callback: () => void): void => { + if (settled) return + settled = true + signal.removeEventListener('abort', onAbort) + callback() + } + const onAbort = (): void => { + if (started()) return + finish(() => { + try { + signal.throwIfAborted() + } catch (reason: unknown) { + rejectObservation(reject, reason) + return + } + /* v8 ignore next -- a native AbortSignal emits abort only after becoming aborted */ + reject(new Error('persistence observation abort event lacked an aborted signal')) + }) + } + signal.addEventListener('abort', onAbort, { once: true }) + operation.then( + (value) => { finish(() => { resolve(value) }) }, + (reason: unknown) => { finish(() => { rejectObservation(reject, reason) }) }, + ) + if (signal.aborted) onAbort() + }) +} + +/** Preserve an exact provider or AbortSignal reason, including legacy non-Error values. */ +function rejectObservation(reject: (reason?: unknown) => void, reason: unknown): void { + reject(reason) +} diff --git a/packages/session-persistence/session-persistence/src/index.ts b/packages/session-persistence/session-persistence/src/index.ts index c785c9354c..9eee07a323 100644 --- a/packages/session-persistence/session-persistence/src/index.ts +++ b/packages/session-persistence/session-persistence/src/index.ts @@ -103,15 +103,17 @@ export abstract class SessionPersistence extends Service { * This read is serialized with writes for the same id and returns detached * values, so observers cannot mutate backend-owned state. * @param id - the persisted session to inspect. + * @param signal - optional cancellation for queued and backend read work. * @returns the header and valid stored event prefix exactly as observed. */ - abstract inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> + abstract inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> /** * Lightweight listing from metadata, without a full-log parse. + * @param signal - optional cancellation for backend listing work. * @returns one header per materialized session. */ - abstract list(): Promise + abstract list(signal?: AbortSignal): Promise /** * List materialized sessions with cheap per-log change tokens. diff --git a/packages/session-persistence/session-persistence/tests/contract.ts b/packages/session-persistence/session-persistence/tests/contract.ts index ae07bf77aa..eb77235057 100644 --- a/packages/session-persistence/session-persistence/tests/contract.ts +++ b/packages/session-persistence/session-persistence/tests/contract.ts @@ -222,6 +222,21 @@ export function runPersistenceContract(name: string, make: () => Promise { + const { persistence, dispose } = await make() + try { + const reason = new Error('persistence observation cancelled') + const controller = new AbortController() + controller.abort(reason) + + await expect(persistence.list(controller.signal)).rejects.toBe(reason) + await expect(persistence.inspect(SessionId('cancelled-inspect'), controller.signal)) + .rejects.toBe(reason) + } finally { + await dispose() + } + }) + it('lists stable lightweight revisions that change after an append', async () => { const { persistence, dispose } = await make() try { diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 6b31d0843b..523a88089e 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -94,8 +94,8 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend return this.coordinator.load(id) } - inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - return this.coordinator.inspect(id) + inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + return this.coordinator.inspect(id, signal) } // --- PersistenceBackend hooks (the Map storage primitives) --- @@ -132,7 +132,8 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend if (closers.length > 0) entry.events.push(...structuredClone(closers) as SessionEvent[]) } - async list(): Promise { + async list(signal?: AbortSignal): Promise { + signal?.throwIfAborted() return [...this.store.values()].map(e => structuredClone(e.meta)) } @@ -153,10 +154,10 @@ class ControlledBackend implements PersistenceBackend { loadAttempts = 0 repairAttempts = 0 beforeAppend?: (attempt: number) => Promise - beforeLoadStored?: (attempt: number) => Promise + beforeLoadStored?: (attempt: number, signal?: AbortSignal) => Promise - async loadStored(id: SessionId): Promise | undefined> { - await this.beforeLoadStored?.(++this.loadAttempts) + async loadStored(id: SessionId, signal?: AbortSignal): Promise | undefined> { + await this.beforeLoadStored?.(++this.loadAttempts, signal) const entry = this.store.get(id) if (entry === undefined) return undefined return { meta: structuredClone(entry.meta), events: structuredClone(entry.events) } @@ -348,6 +349,109 @@ describe('PersistenceCoordinator stored identity', () => { }) }) +describe('PersistenceCoordinator observation cancellation', () => { + it('promptly rejects a queued inspect without invoking it and keeps the same-id chain healthy', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('queued-inspect-cancellation') + backend.store.set(id, { meta: meta(id), events: oneTurnLog() }) + const loadGate = Promise.withResolvers() + backend.beforeLoadStored = async (attempt) => { + if (attempt === 1) await loadGate.promise + } + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + + try { + const prior = coordinator.inspect(id) + await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) }) + const controller = new AbortController() + const reason = new Error('queued inspect cancelled') + const queued = coordinator.inspect(id, controller.signal) + let observedReason: unknown + const observedAbort = queued.catch((error: unknown) => { + observedReason = error + }) + + controller.abort(reason) + + await vi.waitFor(() => { expect(observedReason).toBe(reason) }) + expect(backend.loadAttempts).toBe(1) + const subsequent = coordinator.inspect(id) + expect(backend.loadAttempts).toBe(1) + + loadGate.resolve(true) + await expect(prior).resolves.toMatchObject({ meta: { id } }) + await observedAbort + await expect(subsequent).resolves.toMatchObject({ meta: { id } }) + expect(backend.loadAttempts).toBe(2) + await vi.waitFor(() => { + expect((coordinator as unknown as CoordinatorInternals).chains.size).toBe(0) + }) + } finally { + loadGate.resolve(true) + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('waits for active cooperative inspection cleanup before rejecting cancellation', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('active-inspect-cancellation') + backend.store.set(id, { meta: meta(id), events: oneTurnLog() }) + const cleanupGate = Promise.withResolvers() + let cleanupComplete = false + backend.beforeLoadStored = async (_attempt, signal) => { + await new Promise((resolve) => { + signal?.addEventListener('abort', () => { + void cleanupGate.promise.then(() => { + cleanupComplete = true + resolve() + }) + }, { once: true }) + }) + throw new Error('backend cancellation after cleanup') + } + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + + try { + const controller = new AbortController() + const reason = new Error('active inspect cancelled') + const pending = coordinator.inspect(id, controller.signal) + let observedReason: unknown + const observed = pending.catch((error: unknown) => { + observedReason = error + }) + await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) }) + + controller.abort(reason) + await Promise.resolve() + + expect(observedReason).toBeUndefined() + expect(cleanupComplete).toBe(false) + cleanupGate.resolve(true) + await observed + expect(cleanupComplete).toBe(true) + expect(observedReason).toBe(reason) + const backendFailure = new Error('later inspection failure') + backend.beforeLoadStored = () => Promise.reject(backendFailure) + await expect(coordinator.inspect(id)).rejects.toBe(backendFailure) + } finally { + cleanupGate.resolve(true) + await fiber.dispose() + await ctx.fiber.dispose() + } + }) +}) + describe('PersistenceCoordinator retirement', () => { it('a retiring unmaterialized owner without buffered events releases its id', async () => { const ctx = new Context() diff --git a/packages/session-query/session-query/README.md b/packages/session-query/session-query/README.md index 9ee495f48a..8c74f96e3c 100644 --- a/packages/session-query/session-query/README.md +++ b/packages/session-query/session-query/README.md @@ -8,14 +8,14 @@ - `readSession(sessionId)` returns one complete detached raw log after the same core replay validation used by resume; it never enters the session into the live store. - `filterSessions(filters)` applies provider-independent session metadata and availability predicates to that same cloned logical corpus. - `filterEvents(sessionId, filters)` extracts first-party semantic documents and applies provider-independent metadata and literal-text predicates in ascending seq order. -- `readTitleSnapshot(sessionId)` loads one live-preferred or persisted log and returns the cloned source header with its latest folded `session/title` event. `readTitle(sessionId)` is the title-only convenience view; it returns `undefined` when the known session has no title. +- `readTitleSnapshots(sessionIds, signal?)` resolves unique ids from one live-preferred corpus observation, passes cancellation through persisted listing and inspection, and returns ordered per-session settlements so one missing or malformed title source does not discard its peers. Each live source is folded directly, and each persisted worker folds to a detached header/title result and releases the full log before dequeuing another id. Cancellation rejects the whole batch. `readTitleSnapshot(sessionId, signal?)` is the one-observation view; `readTitle(sessionId, signal?)` returns only its optional folded `session/title`. - `listEvents(sessionId)` loads the live-preferred raw log and classifies each event as `current`, `shadowed`, or `log-only` with the shared `dsh-session` surface fold. - `readSurface(sessionId)` returns one cloned header, raw-log capture boundary, and the complete folded current surface in model-history order. A live session wins over persistence; compaction is observed before or after its replacement append, never as a synthetic mixture. - `readEvent(request)` returns a cloned header, the full target event, and a bounded raw-seq window. `before` and `after` default to zero and may not exceed `readWindowMax`. - `traceSession(sessionId)` reads the corpus once and returns immediate-to-outward ancestors plus deterministic recursive descendant trees. `complete: false` identifies the first missing parent; a target-connected cycle fails with `SESSION_QUERY_INVALID_LINEAGE`. - `traceEvent(request)` loads the logical log once and returns its cloned source header with direct positional replacements and direct logged provenance. `replacementChain` follows positional replacers to the final replacement; provenance links remain non-transitive. -Persistence is optional and may mount or unmount dynamically. Cross-corpus listing and lineage tracing fail with `SESSION_QUERY_PERSISTENCE_FAILED` while mounted persistence is unreadable. A title, event read, or trace targeting a known live session does not consult persistence, so durable backend health cannot make current in-memory state unreadable. Persisted title and event operations list before loading and reject a metadata mismatch rather than combining inconsistent observations. `listSessions()` remains lightweight and does not load logs or index titles. +Persistence is optional and may mount or unmount dynamically. Cross-corpus listing and lineage tracing fail with `SESSION_QUERY_PERSISTENCE_FAILED` while mounted persistence is unreadable. A title, event read, or trace targeting a known live session does not consult persistence, so durable backend health cannot make current in-memory state unreadable. Persisted title and event operations list before loading and reject a metadata mismatch rather than combining inconsistent observations. A batch title observation performs one metadata listing, inspects its unique persisted ids with at most four workers, and preserves each title's own observed header for downstream authorization. Cancellation starts no queued inspections and rejects only after already-started workers settle. `listSessions()` remains lightweight and does not load logs or index titles. ## Filtering and extraction diff --git a/packages/session-query/session-query/src/corpus.ts b/packages/session-query/session-query/src/corpus.ts index 0e1753d5ce..ebf0f40bbc 100644 --- a/packages/session-query/session-query/src/corpus.ts +++ b/packages/session-query/session-query/src/corpus.ts @@ -15,6 +15,22 @@ export interface LogicalSession { events: SessionEvent[] } +/** Borrowed source visible only during one synchronous batch projection. */ +export interface LogicalSessionSource { + /** Header selected with `events`; callers must clone retained output. */ + readonly header: SessionHeader + /** Raw events selected with `header`; valid only for the projection call. */ + readonly events: readonly SessionEvent[] +} + +/** One source-projection result in a batch logical-corpus observation. */ +export type LogicalProjectionResult = + | { sessionId: SessionId; status: 'fulfilled'; value: Value } + | { sessionId: SessionId; status: 'rejected'; reason: unknown } + +/** Bound persisted observation fan-out for public batch title reads. */ +const PERSISTED_INSPECT_CONCURRENCY = 4 + /** Resolves a live-preferred corpus against the persistence service mounted now. */ export class SessionCorpus { private _persistence: SessionPersistence | undefined @@ -72,16 +88,7 @@ export class SessionCorpus { if (persistence === undefined) throw notFound(sessionId) const listed = (await listPersisted(persistence)).find(header => header.id === sessionId) if (listed === undefined) throw notFound(sessionId) - let loaded: Awaited> - try { - loaded = await persistence.inspect(sessionId) - } catch (error: unknown) { - throw new SessionQueryError( - `failed to inspect session "${sessionId}": ${errorMessage(error)}`, - 'SESSION_QUERY_PERSISTENCE_FAILED', - { cause: error }, - ) - } + const loaded = await inspectPersisted(persistence, sessionId) const attached = this._ctx.sessions.get(sessionId) if (attached !== undefined) return snapshotLive(attached) assertSessionHeadersCompatible(loaded.meta, listed) @@ -90,11 +97,147 @@ export class SessionCorpus { events: loaded.events.map(event => structuredClone(event)), } } + + /** + * Project unique logical sources immediately from one persistence listing. + * + * The synchronous projector runs before a persisted worker claims its next id. + * Full logs are borrowed only for that call and never retained by the batch. + * @param sessionIds - sessions to resolve in first-occurrence order. + * @param project - synchronous fold that owns/clones every retained value. + * @param signal - cancellation shared by listing and every persisted inspection. + * @returns one fulfilled or rejected projected result per unique requested id. + */ + async projectMany( + sessionIds: readonly SessionId[], + project: (source: LogicalSessionSource) => Value, + signal?: AbortSignal, + ): Promise[]> { + const ids = [...new Set(sessionIds)] + signal?.throwIfAborted() + const resolved = new Map>() + const unresolved: SessionId[] = [] + for (const id of ids) { + const session = this._ctx.sessions.get(id) + if (session === undefined) { + unresolved.push(id) + } else { + resolved.set(id, projectSource(id, sourceLive(session), project, signal)) + } + } + if (unresolved.length === 0) return orderedResults(ids, resolved) + + const persistence = this._persistence + if (persistence === undefined) { + for (const sessionId of unresolved) { + resolved.set(sessionId, { sessionId, status: 'rejected', reason: notFound(sessionId) }) + } + return orderedResults(ids, resolved) + } + + let persisted: SessionHeader[] + try { + persisted = await listPersisted(persistence, signal) + signal?.throwIfAborted() + } catch (error: unknown) { + if (signal?.aborted) signal.throwIfAborted() + for (const sessionId of unresolved) { + resolved.set(sessionId, { sessionId, status: 'rejected', reason: error }) + } + return orderedResults(ids, resolved) + } + const persistedById = new Map(persisted.map(header => [header.id, header])) + const resolvePersisted = async (sessionId: SessionId): Promise => { + const listed = persistedById.get(sessionId) + if (listed === undefined) { + const attached = this._ctx.sessions.get(sessionId) + resolved.set(sessionId, attached === undefined + ? { sessionId, status: 'rejected', reason: notFound(sessionId) } + : projectSource(sessionId, sourceLive(attached), project, signal)) + return + } + try { + signal?.throwIfAborted() + const loaded = await inspectPersisted(persistence, sessionId, signal) + signal?.throwIfAborted() + const attached = this._ctx.sessions.get(sessionId) + if (attached !== undefined) { + resolved.set(sessionId, projectSource(sessionId, sourceLive(attached), project, signal)) + return + } + assertSessionHeadersCompatible(loaded.meta, listed) + resolved.set(sessionId, projectSource(sessionId, { + header: loaded.meta, + events: loaded.events, + }, project, signal)) + } catch (error: unknown) { + if (signal?.aborted) signal.throwIfAborted() + resolved.set(sessionId, { sessionId, status: 'rejected', reason: error }) + } + } + let cursor = 0 + const worker = async (): Promise => { + for (;;) { + signal?.throwIfAborted() + const index = cursor + if (index >= unresolved.length) return + cursor += 1 + await resolvePersisted(unresolved[index] as SessionId) + } + } + const workerCount = Math.min(PERSISTED_INSPECT_CONCURRENCY, unresolved.length) + const settlements = await Promise.allSettled( + Array.from({ length: workerCount }, () => worker()), + ) + if (signal?.aborted) signal.throwIfAborted() + /* v8 ignore start -- per-id failures settle inside resolvePersisted; workers reject only on abort above */ + for (const settlement of settlements) { + if (settlement.status === 'rejected') { + const reason: unknown = settlement.reason + throw reason + } + } + /* v8 ignore stop */ + signal?.throwIfAborted() + return orderedResults(ids, resolved) + } } -async function listPersisted(persistence: SessionPersistence): Promise { +function projectSource( + sessionId: SessionId, + source: LogicalSessionSource, + project: (source: LogicalSessionSource) => Value, + signal?: AbortSignal, +): LogicalProjectionResult { try { - return await persistence.list() + signal?.throwIfAborted() + const value = project(source) + signal?.throwIfAborted() + return { sessionId, status: 'fulfilled', value } + } catch (reason: unknown) { + /* v8 ignore next -- the synchronous projector has no external cancellation yield */ + if (signal?.aborted) signal.throwIfAborted() + return { sessionId, status: 'rejected', reason } + } +} + +function sourceLive(session: Session): LogicalSessionSource { + return { header: session.header, events: session.events } +} + +function orderedResults( + ids: readonly SessionId[], + resolved: ReadonlyMap>, +): LogicalProjectionResult[] { + return ids.map(sessionId => resolved.get(sessionId) as LogicalProjectionResult) +} + +async function listPersisted( + persistence: SessionPersistence, + signal?: AbortSignal, +): Promise { + try { + return await persistence.list(signal) } catch (error: unknown) { throw new SessionQueryError( `session persistence listing failed: ${errorMessage(error)}`, @@ -104,6 +247,23 @@ async function listPersisted(persistence: SessionPersistence): Promise>> { + try { + return await persistence.inspect(sessionId, signal) + } catch (error: unknown) { + if (signal?.aborted) signal.throwIfAborted() + throw new SessionQueryError( + `failed to inspect session "${sessionId}": ${errorMessage(error)}`, + 'SESSION_QUERY_PERSISTENCE_FAILED', + { cause: error }, + ) + } +} + function snapshotLive(session: Session): LogicalSession { return { header: structuredClone(session.header), diff --git a/packages/session-query/session-query/src/index.ts b/packages/session-query/session-query/src/index.ts index b38914b3b6..000eb83425 100644 --- a/packages/session-query/session-query/src/index.ts +++ b/packages/session-query/session-query/src/index.ts @@ -28,6 +28,7 @@ import type { SessionSearchRequest, SessionSurfaceSnapshot, SessionTitleObservation, + SessionTitleObservationResult, } from './types.ts' import { SESSION_QUERY_READ_WINDOW_MAX, @@ -148,24 +149,51 @@ export abstract class SessionQueryService extends Service { /** * Fold the latest log-backed title from one live-preferred logical session. * @param sessionId - live or persisted session id to read. + * @param signal - optional cancellation for source resolution and title folding. * @returns latest title snapshot, or `undefined` when the log has no title event. */ - async readTitle(sessionId: SessionId): Promise { - return (await this.readTitleSnapshot(sessionId)).title + async readTitle( + sessionId: SessionId, + signal?: AbortSignal, + ): Promise { + return (await this.readTitleSnapshot(sessionId, signal)).title } /** * Fold the latest title and return its source header from one corpus observation. * @param sessionId - live or persisted session id to read. + * @param signal - optional cancellation for source resolution and title folding. * @returns cloned source header and optional latest title snapshot. */ - async readTitleSnapshot(sessionId: SessionId): Promise { - const loaded = await this._corpus.load(sessionId) - const title = foldSessionTitle(loaded.events) - return { - session: loaded.header, - ...title === undefined ? {} : { title }, - } + async readTitleSnapshot( + sessionId: SessionId, + signal?: AbortSignal, + ): Promise { + const result = (await this.readTitleSnapshots([sessionId], signal))[0] as SessionTitleObservationResult + if (result.status === 'rejected') throw result.reason + return result.value + } + + /** + * Fold titles for unique sessions from one cancellable corpus observation. + * + * Results preserve first-occurrence input order. Operational failures stay + * isolated per session, while cancellation rejects the complete operation. + * @param sessionIds - live or persisted session ids to observe. + * @param signal - optional cancellation shared by all source reads. + * @returns one fulfilled or rejected result per unique requested id. + */ + async readTitleSnapshots( + sessionIds: readonly SessionId[], + signal?: AbortSignal, + ): Promise { + return this._corpus.projectMany(sessionIds, (source): SessionTitleObservation => { + const title = foldSessionTitle(source.events) + return { + session: structuredClone(source.header), + ...title === undefined ? {} : { title }, + } + }, signal) } /** diff --git a/packages/session-query/session-query/src/types.ts b/packages/session-query/session-query/src/types.ts index d3e7196152..b01d80dade 100644 --- a/packages/session-query/session-query/src/types.ts +++ b/packages/session-query/session-query/src/types.ts @@ -157,6 +157,25 @@ export interface SessionTitleObservation { title?: SessionTitleSnapshot } +/** One ordered result from a batch title observation. */ +export type SessionTitleObservationResult = + | { + /** Requested session id. */ + sessionId: SessionId + /** Successful atomic header/title observation. */ + status: 'fulfilled' + /** Header and optional latest title from one logical source. */ + value: SessionTitleObservation + } + | { + /** Requested session id. */ + sessionId: SessionId + /** Operational failure isolated to this session. */ + status: 'rejected' + /** Original failure from logical-source resolution or title folding. */ + reason: unknown + } + /** Inclusive numeric interval used by time and sequence filters. */ export interface SessionResultRange { /** Inclusive lower bound. */ diff --git a/packages/session-query/session-query/tests/session-query.spec.ts b/packages/session-query/session-query/tests/session-query.spec.ts index a69c3bcb93..f88d8be0f6 100644 --- a/packages/session-query/session-query/tests/session-query.spec.ts +++ b/packages/session-query/session-query/tests/session-query.spec.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { Context, type Fiber } from 'cordis' import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionHeader, SessionId as SessionIdType } from '@deepseek-ai/dsh-session' @@ -27,16 +27,31 @@ function eventLog(text = 'hello'): SessionEvent[] { class TestPersistence extends SessionPersistence { static entries = new Map() static listFailure: unknown + static listOverride: ((signal?: AbortSignal) => Promise) | undefined static inspectFailure: unknown static inspectEffect: (() => void) | undefined + static inspectOverride: (( + id: SessionIdType, + signal?: AbortSignal, + ) => Promise<{ meta: SessionHeader; events: SessionEvent[] }>) | undefined static afterList: (() => void) | undefined + static listCalls = 0 + static inspectCalls: SessionIdType[] = [] + static listSignals: Array = [] + static inspectSignals: Array = [] static reset(entries: readonly { meta: SessionHeader; events: SessionEvent[] }[] = []): void { this.entries = new Map(entries.map(entry => [entry.meta.id, structuredClone(entry)])) this.listFailure = undefined + this.listOverride = undefined this.inspectFailure = undefined this.inspectEffect = undefined + this.inspectOverride = undefined this.afterList = undefined + this.listCalls = 0 + this.inspectCalls = [] + this.listSignals = [] + this.inspectSignals = [] } locate(_meta: SessionHeader): undefined { @@ -59,7 +74,15 @@ class TestPersistence extends SessionPersistence { return this.inspect(id) } - inspect(id: SessionIdType): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + inspect( + id: SessionIdType, + signal?: AbortSignal, + ): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { + TestPersistence.inspectCalls.push(id) + TestPersistence.inspectSignals.push(signal) + if (TestPersistence.inspectOverride !== undefined) { + return TestPersistence.inspectOverride(id, signal) + } if (TestPersistence.inspectFailure !== undefined) return rejectUnknown(TestPersistence.inspectFailure) const entry = TestPersistence.entries.get(id) if (entry === undefined) return Promise.reject(new Error('missing test session')) @@ -69,7 +92,10 @@ class TestPersistence extends SessionPersistence { return Promise.resolve(result) } - list(): Promise { + list(signal?: AbortSignal): Promise { + TestPersistence.listCalls += 1 + TestPersistence.listSignals.push(signal) + if (TestPersistence.listOverride !== undefined) return TestPersistence.listOverride(signal) if (TestPersistence.listFailure !== undefined) return rejectUnknown(TestPersistence.listFailure) const headers = [...TestPersistence.entries.values()].map(entry => structuredClone(entry.meta)) TestPersistence.afterList?.() @@ -192,6 +218,336 @@ describe('session-query exact reads', () => { expect(Object.keys((await ctx.sessionQuery.listSessions())[0]!)).toEqual(['header', 'live', 'persisted']) }) + it('batches unique persisted title observations through one cancellable corpus scan', async () => { + const first = header('batch-title-first', 1) + const second = header('batch-title-second', 2) + const titleEvent = (title: string, time: number): SessionEvent => ({ + type: 'session/title', + seq: 0, + time, + data: { + title, + messageSeqs: [], + source: { kind: 'fallback' }, + }, + }) + TestPersistence.reset([ + { meta: first, events: [titleEvent('First title', 10)] }, + { meta: second, events: [titleEvent('Second title', 20)] }, + ]) + const ctx = await liveContext() + await ctx.plugin(TestPersistence) + const signal = new AbortController().signal + const missing = SessionId('batch-title-missing') + + const results = await ctx.sessionQuery.readTitleSnapshots( + [second.id, first.id, second.id, missing], + signal, + ) + + expect(results.map(result => [result.sessionId, result.status])).toEqual([ + [second.id, 'fulfilled'], + [first.id, 'fulfilled'], + [missing, 'rejected'], + ]) + expect(results[0]).toMatchObject({ value: { session: second, title: { title: 'Second title' } } }) + expect(results[1]).toMatchObject({ value: { session: first, title: { title: 'First title' } } }) + expect(TestPersistence.listCalls).toBe(1) + expect(TestPersistence.inspectCalls).toEqual([second.id, first.id]) + expect(TestPersistence.listSignals).toEqual([signal]) + expect(TestPersistence.inspectSignals).toEqual([signal, signal]) + }) + + it('bounds persisted title inspection concurrency while preserving ordered results', async () => { + const entries = Array.from({ length: 12 }, (_, index) => { + const meta = header(`bounded-title-${index}`, index) + return { meta, events: eventLog(`title-${index}`) } + }) + TestPersistence.reset(entries) + const ctx = await liveContext() + await ctx.plugin(TestPersistence) + let active = 0 + let maximum = 0 + TestPersistence.inspectOverride = async (id) => { + active += 1 + maximum = Math.max(maximum, active) + await new Promise(resolve => setImmediate(resolve)) + active -= 1 + const entry = TestPersistence.entries.get(id) + if (entry === undefined) throw new Error('missing bounded test session') + return structuredClone(entry) + } + + const results = await ctx.sessionQuery.readTitleSnapshots(entries.map(entry => entry.meta.id)) + + expect(maximum).toBe(4) + expect(TestPersistence.listCalls).toBe(1) + expect(TestPersistence.inspectCalls).toEqual(entries.map(entry => entry.meta.id)) + expect(results.map(result => result.sessionId)).toEqual(entries.map(entry => entry.meta.id)) + expect(results.every(result => result.status === 'fulfilled')).toBe(true) + }) + + it('folds and discards each completed log before its worker dequeues another inspection', async () => { + const entries = Array.from({ length: 5 }, (_, index) => ({ + meta: header(`project-title-${index}`, index), + events: [], + })) + TestPersistence.reset(entries) + const ctx = await liveContext() + await ctx.plugin(TestPersistence) + const timeline: string[] = [] + const releases = new Map void>() + TestPersistence.inspectOverride = id => new Promise((resolve) => { + timeline.push(`inspect:${id}`) + releases.set(id, () => { + const marker = `full-log-marker:${id}` + const titleEvent = { + type: 'session/title', + seq: 1, + time: 20, + data: { + title: `Projected ${id}`, + get messageSeqs() { + timeline.push(`project:${id}`) + return [] + }, + source: { kind: 'fallback' }, + }, + } as unknown as SessionEvent + resolve({ + meta: entries.find(entry => entry.meta.id === id)!.meta, + events: [...eventLog(marker), titleEvent], + }) + }) + }) + const release = (id: SessionIdType): void => { + const settle = releases.get(id) + if (settle === undefined) throw new Error(`inspection ${id} has not started`) + settle() + } + const ids = entries.map(entry => entry.meta.id) + + const pending = ctx.sessionQuery.readTitleSnapshots(ids) + await vi.waitFor(() => { expect(TestPersistence.inspectCalls).toHaveLength(4) }) + release(ids[0]!) + await vi.waitFor(() => { expect(TestPersistence.inspectCalls).toHaveLength(5) }) + + // Heap-retention assertions would depend on nondeterministic GC. This ordering + // is the deterministic guard: a retain-all implementation cannot touch the + // observable title getter until every inspection has completed. + expect(timeline.indexOf(`project:${ids[0]}`)) + .toBeLessThan(timeline.indexOf(`inspect:${ids[4]}`)) + for (const id of ids.slice(1)) release(id) + const results = await pending + + expect(results.map(result => result.sessionId)).toEqual(ids) + expect(JSON.stringify(results)).not.toContain('full-log-marker:') + expect(results.every(result => result.status === 'fulfilled')).toBe(true) + }) + + it('passes cancellation into a stalled persisted title batch and rejects with its reason', async () => { + const persisted = header('stalled-title', 1) + TestPersistence.reset([{ meta: persisted, events: [] }]) + const ctx = await liveContext() + await ctx.plugin(TestPersistence) + const controller = new AbortController() + const reason = new Error('title deadline') + let started!: () => void + const inspectStarted = new Promise((resolve) => { started = resolve }) + TestPersistence.inspectOverride = (_id, signal) => new Promise((_resolve, reject) => { + started() + signal?.addEventListener('abort', () => { reject(reason) }, { once: true }) + }) + + const pending = ctx.sessionQuery.readTitleSnapshots([persisted.id], controller.signal) + await inspectStarted + controller.abort(reason) + + await expect(pending).rejects.toBe(reason) + expect(TestPersistence.listSignals).toEqual([controller.signal]) + expect(TestPersistence.inspectSignals).toEqual([controller.signal]) + }) + + it('drains started title inspections after cancellation without starting queued ids', async () => { + const entries = Array.from({ length: 8 }, (_, index) => ({ + meta: header(`cancel-queued-title-${index}`, index), + events: eventLog(`queued-${index}`), + })) + TestPersistence.reset(entries) + const ctx = await liveContext() + await ctx.plugin(TestPersistence) + const controller = new AbortController() + const reason = new Error('cancel queued title batch') + const releases: Array<() => void> = [] + let abortsObserved = 0 + let inspectionsSettled = 0 + TestPersistence.inspectOverride = (_id, signal) => new Promise((_resolve, reject) => { + signal?.addEventListener('abort', () => { abortsObserved += 1 }, { once: true }) + releases.push(() => { + inspectionsSettled += 1 + reject(reason) + }) + }) + + const pending = ctx.sessionQuery.readTitleSnapshots( + entries.map(entry => entry.meta.id), + controller.signal, + ) + let batchSettled = false + void pending.then( + () => { batchSettled = true }, + () => { batchSettled = true }, + ) + await vi.waitFor(() => { expect(TestPersistence.inspectCalls).toHaveLength(4) }) + controller.abort(reason) + await vi.waitFor(() => { expect(abortsObserved).toBe(4) }) + + expect(batchSettled).toBe(false) + expect(TestPersistence.inspectCalls).toEqual(entries.slice(0, 4).map(entry => entry.meta.id)) + for (const release of releases) release() + + await expect(pending).rejects.toBe(reason) + expect(inspectionsSettled).toBe(4) + expect(TestPersistence.inspectCalls).toEqual(entries.slice(0, 4).map(entry => entry.meta.id)) + }) + + it('passes cancellation into a stalled persisted title listing and rejects with its reason', async () => { + const persisted = header('stalled-title-list', 1) + TestPersistence.reset([{ meta: persisted, events: [] }]) + const ctx = await liveContext() + await ctx.plugin(TestPersistence) + const controller = new AbortController() + const reason = new Error('title listing deadline') + let started!: () => void + const listStarted = new Promise((resolve) => { started = resolve }) + TestPersistence.listOverride = signal => new Promise((_resolve, reject) => { + started() + signal?.addEventListener('abort', () => { reject(reason) }, { once: true }) + }) + + const pending = ctx.sessionQuery.readTitleSnapshots([persisted.id], controller.signal) + await listStarted + controller.abort(reason) + + await expect(pending).rejects.toBe(reason) + expect(TestPersistence.listSignals).toEqual([controller.signal]) + expect(TestPersistence.inspectCalls).toEqual([]) + }) + + it('isolates title read and fold failures while preferring a live owner attached during inspection', async () => { + const attached = header('batch-title-attached', 1) + const failed = header('batch-title-failed', 2) + const malformed = header('batch-title-malformed', 3) + const inspectFailure = new Error('one title inspect failed') + const malformedTitle = { + type: 'session/title', + seq: 0, + time: 30, + data: { + title: 'malformed', + source: { kind: 'fallback' }, + }, + } as unknown as SessionEvent + TestPersistence.reset([ + { meta: attached, events: eventLog('stale persisted') }, + { meta: failed, events: [] }, + { meta: malformed, events: [malformedTitle] }, + ]) + const ctx = await liveContext() + await ctx.plugin(TestPersistence) + TestPersistence.inspectOverride = (id) => { + if (id === failed.id) return Promise.reject(inspectFailure) + const entry = TestPersistence.entries.get(id) + if (entry === undefined) return Promise.reject(new Error('missing test session')) + if (id === attached.id) { + const session = ctx.sessions.create(attached.id, { meta: { createdAt: attached.createdAt } }) + session.append('session/title', { + title: 'Attached live title', + messageSeqs: [], + source: { kind: 'fallback' }, + }) + } + return Promise.resolve(structuredClone(entry)) + } + + const results = await ctx.sessionQuery.readTitleSnapshots([ + attached.id, + failed.id, + malformed.id, + ]) + + expect(results[0]).toMatchObject({ + status: 'fulfilled', + value: { session: attached, title: { title: 'Attached live title' } }, + }) + expect(results[1]).toMatchObject({ + sessionId: failed.id, + status: 'rejected', + reason: { + code: 'SESSION_QUERY_PERSISTENCE_FAILED', + cause: inspectFailure, + }, + }) + expect(results[2]).toMatchObject({ sessionId: malformed.id, status: 'rejected' }) + if (results[2]?.status !== 'rejected') throw new Error('expected malformed title rejection') + expect(results[2].reason).toBeInstanceOf(TypeError) + }) + + it('preserves live batch results across missing persistence, listing failure, and late attachment', async () => { + const liveOnly = await liveContext() + const live = liveOnly.sessions.create(SessionId('batch-title-live')) + const missing = SessionId('batch-title-no-persistence') + + await expect(liveOnly.sessionQuery.readTitleSnapshots([live.id, live.id])).resolves.toEqual([{ + sessionId: live.id, + status: 'fulfilled', + value: { session: live.header }, + }]) + await expect(liveOnly.sessionQuery.readTitleSnapshots([live.id, missing])).resolves.toMatchObject([ + { sessionId: live.id, status: 'fulfilled' }, + { sessionId: missing, status: 'rejected' }, + ]) + await expect(liveOnly.sessionQuery.readTitleSnapshot(missing)) + .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND')) + + const persisted = header('batch-title-persisted', 1) + const late = header('batch-title-late', 2) + TestPersistence.reset([{ meta: persisted, events: [] }]) + const mixed = await liveContext() + const mixedLive = mixed.sessions.create(SessionId('batch-title-mixed-live')) + await mixed.plugin(TestPersistence) + TestPersistence.afterList = () => { + mixed.sessions.create(late.id, { meta: { createdAt: late.createdAt } }) + TestPersistence.afterList = undefined + } + + await expect(mixed.sessionQuery.readTitleSnapshots([ + mixedLive.id, + persisted.id, + late.id, + ])).resolves.toMatchObject([ + { sessionId: mixedLive.id, status: 'fulfilled' }, + { sessionId: persisted.id, status: 'fulfilled' }, + { sessionId: late.id, status: 'fulfilled' }, + ]) + + TestPersistence.reset() + TestPersistence.listFailure = new Error('title listing failed') + const failedList = await liveContext() + const survivingLive = failedList.sessions.create(SessionId('batch-title-list-live')) + await failedList.plugin(TestPersistence) + + await expect(failedList.sessionQuery.readTitleSnapshots([survivingLive.id, missing])) + .resolves.toMatchObject([ + { sessionId: survivingLive.id, status: 'fulfilled' }, + { + sessionId: missing, + status: 'rejected', + reason: expectCode('SESSION_QUERY_PERSISTENCE_FAILED'), + }, + ]) + }) + it('lists live sessions deterministically and returns detached headers', async () => { const ctx = await liveContext() const older = ctx.sessions.create(SessionId('older'), { meta: { createdAt: 1 } }) diff --git a/packages/session-query/tool-session-query/src/index.ts b/packages/session-query/tool-session-query/src/index.ts index e7b8b1ef88..999fc1aa40 100644 --- a/packages/session-query/tool-session-query/src/index.ts +++ b/packages/session-query/tool-session-query/src/index.ts @@ -741,8 +741,16 @@ async function readTitles( signal: AbortSignal, ): Promise { const result = new Map() - for (const id of new Set(ids)) { - result.set(id, await readTitle(ctx, caller, id, signal)) + signal.throwIfAborted() + const observations = await ctx.sessionQuery.readTitleSnapshots(ids, signal) + signal.throwIfAborted() + for (const observation of observations) { + if (observation.status === 'rejected') { + result.set(observation.sessionId, unavailableTitle(ctx, observation.sessionId, observation.reason)) + continue + } + assertObservedTargetAuthorized(caller, observation.sessionId, observation.value.session) + result.set(observation.sessionId, { text: observation.value.title?.title ?? 'untitled' }) } return result as CompleteTitleMap } @@ -753,19 +761,18 @@ async function readTitle( id: SessionIdValue, signal: AbortSignal, ): Promise { - signal.throwIfAborted() - try { - const observation = await ctx.sessionQuery.readTitleSnapshot(id) - signal.throwIfAborted() - assertObservedTargetAuthorized(caller, id, observation.session) - return { text: observation.title?.title ?? 'untitled' } - } catch (error: unknown) { - if (signal.aborted) signal.throwIfAborted() - if (error instanceof HarnessError && error.code === 'SESSION_QUERY_TOOL_UNAUTHORIZED') throw error - const code = error instanceof HarnessError ? error.code : 'UNKNOWN' - ctx.logger.warn(`tool-session-query: title read failed for session "${id}": ${fullError(error)}`) - return { text: 'untitled', unavailableCode: code } - } + return (await readTitles(ctx, caller, [id], signal)).get(id) +} + +function unavailableTitle( + ctx: Context, + id: SessionIdValue, + error: unknown, +): TitleView { + if (error instanceof HarnessError && error.code === 'SESSION_QUERY_TOOL_UNAUTHORIZED') throw error + const code = error instanceof HarnessError ? error.code : 'UNKNOWN' + ctx.logger.warn(`tool-session-query: title read failed for session "${id}": ${fullError(error)}`) + return { text: 'untitled', unavailableCode: code } } function fullError(error: unknown): string { diff --git a/packages/session-query/tool-session-query/tests/tool-session-query.spec.ts b/packages/session-query/tool-session-query/tests/tool-session-query.spec.ts index 8aa8376d67..c33b1c9967 100644 --- a/packages/session-query/tool-session-query/tests/tool-session-query.spec.ts +++ b/packages/session-query/tool-session-query/tests/tool-session-query.spec.ts @@ -21,6 +21,7 @@ import SessionQueryService, { type SessionSearchHit, type SessionSearchPage, type SessionSearchRequest, + type SessionTitleObservationResult, } from '@deepseek-ai/dsh-session-query' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry, { type ToolExecutionResult } from '@deepseek-ai/dsh-tools' @@ -155,20 +156,31 @@ class FakeQuery extends SessionQueryService { return FakeQuery.eventSearch(request, exec) } - override async readTitleSnapshot(sessionId: SessionIdValue) { - const value = FakeQuery.titles.get(sessionId) - if (value instanceof Error) throw value - if (value === undefined) return super.readTitleSnapshot(sessionId) - return { - session: (await this.readSurface(sessionId)).session, - title: { - title: value, - messageSeqs: [], - source: { kind: 'fallback' as const }, - eventSeq: 0, - updatedAt: 1, - }, - } + override async readTitleSnapshots( + sessionIds: readonly SessionIdValue[], + signal?: AbortSignal, + ): Promise { + const observations = await super.readTitleSnapshots(sessionIds, signal) + return observations.map((observation): SessionTitleObservationResult => { + const value = FakeQuery.titles.get(observation.sessionId) + if (value instanceof Error) { + return { sessionId: observation.sessionId, status: 'rejected', reason: value } + } + if (value === undefined || observation.status === 'rejected') return observation + return { + ...observation, + value: { + ...observation.value, + title: { + title: value, + messageSeqs: [], + source: { kind: 'fallback' }, + eventSeq: 0, + updatedAt: 1, + }, + }, + } + }) } } @@ -494,9 +506,13 @@ describe('workspace authority and lineage redaction', () => { root: targetRecord, }) const titleReads: SessionIdValue[] = [] - vi.spyOn(mounted.ctx.sessionQuery, 'readTitleSnapshot').mockImplementation((sessionId) => { - titleReads.push(sessionId) - return Promise.resolve({ session: header(sessionId, '/work') }) + vi.spyOn(mounted.ctx.sessionQuery, 'readTitleSnapshots').mockImplementation((sessionIds) => { + titleReads.push(...sessionIds) + return Promise.resolve([...new Set(sessionIds)].map(sessionId => ({ + sessionId, + status: 'fulfilled' as const, + value: { session: header(sessionId, '/work') }, + }))) }) const output = text(await mounted.call('session_trace', { session_id: target.id })) @@ -596,16 +612,20 @@ describe('workspace authority and lineage redaction', () => { FakeQuery.sessionSearch = () => Promise.resolve({ items: [sessionHit(target.id, '/work', 'safe hit')], }) - vi.spyOn(mounted.ctx.sessionQuery, 'readTitleSnapshot').mockResolvedValueOnce({ - session: movedHeader, - title: { - title: 'secret moved title', - messageSeqs: [], - source: { kind: 'fallback' }, - eventSeq: 0, - updatedAt: 1, + vi.spyOn(mounted.ctx.sessionQuery, 'readTitleSnapshots').mockResolvedValueOnce([{ + sessionId: target.id, + status: 'fulfilled', + value: { + session: movedHeader, + title: { + title: 'secret moved title', + messageSeqs: [], + source: { kind: 'fallback' }, + eventSeq: 0, + updatedAt: 1, + }, }, - }) + }]) const titled = await mounted.call('session_search', { query: 'safe' }) expect(errorCode(titled)).toBe('SESSION_QUERY_TOOL_UNAUTHORIZED') expect(text(titled)).not.toContain('secret moved title') @@ -828,9 +848,11 @@ describe('search paging, prior-history bounds, titles, and cancellation', () => const second = createSession(mounted.ctx, 'stackless-title', '/work') const stackless = new Error('stackless') Object.defineProperty(stackless, 'stack', { value: undefined }) - const readTitle = vi.spyOn(mounted.ctx.sessionQuery, 'readTitleSnapshot') - .mockRejectedValueOnce('string failure') - .mockRejectedValueOnce(stackless) + const readTitles = vi.spyOn(mounted.ctx.sessionQuery, 'readTitleSnapshots') + .mockResolvedValueOnce([ + { sessionId: first.id, status: 'rejected', reason: 'string failure' }, + { sessionId: second.id, status: 'rejected', reason: stackless }, + ]) FakeQuery.sessionSearch = () => Promise.resolve({ items: [ sessionHit(first.id, '/work'), @@ -840,7 +862,8 @@ describe('search paging, prior-history bounds, titles, and cancellation', () => const warn = vi.spyOn(mounted.ctx.logger, 'warn').mockImplementation(() => undefined) const result = await mounted.call('session_search', { query: 'needle' }) expect(text(result)).toContain('title unavailable: UNKNOWN') - expect(readTitle).toHaveBeenCalledTimes(2) + expect(readTitles).toHaveBeenCalledTimes(1) + expect(readTitles.mock.calls[0]?.[0]).toEqual([first.id, second.id]) expect(warn).toHaveBeenCalledWith(expect.stringContaining('string failure')) expect(warn).toHaveBeenCalledWith(expect.stringContaining('Error: stackless')) }) @@ -849,14 +872,43 @@ describe('search paging, prior-history bounds, titles, and cancellation', () => const mounted = await mount() const hit = createSession(mounted.ctx, 'abort-title', '/work') const controller = new AbortController() + const cancellation = new Error('cancelled title batch') FakeQuery.sessionSearch = () => Promise.resolve({ items: [sessionHit(hit.id, '/work')] }) - vi.spyOn(mounted.ctx.sessionQuery, 'readTitleSnapshot').mockImplementation(() => { - controller.abort() - return Promise.reject(new Error('cancelled title')) + let started!: () => void + const batchStarted = new Promise((resolve) => { started = resolve }) + const readTitles = vi.spyOn(mounted.ctx.sessionQuery, 'readTitleSnapshots').mockImplementation((_ids, signal) => { + started() + return new Promise((_resolve, reject) => { + signal?.addEventListener('abort', () => { reject(cancellation) }, { once: true }) + }) }) - const result = await mounted.call('session_search', { query: 'needle' }, { signal: controller.signal }) + const pending = mounted.call('session_search', { query: 'needle' }, { signal: controller.signal }) + await batchStarted + controller.abort(cancellation) + const result = await pending expect(result.isError).toBe(true) expect(text(result)).not.toContain('title unavailable') + expect(readTitles.mock.calls[0]?.[1]).toBe(controller.signal) + }) + + it('does not downgrade an authorization failure returned by title observation', async () => { + const mounted = await mount() + const hit = createSession(mounted.ctx, 'unauthorized-title-error', '/work') + const failure = new HarnessError( + 'title observation became unauthorized', + 'SESSION_QUERY_TOOL_UNAUTHORIZED', + ) + FakeQuery.sessionSearch = () => Promise.resolve({ items: [sessionHit(hit.id, '/work')] }) + vi.spyOn(mounted.ctx.sessionQuery, 'readTitleSnapshots').mockResolvedValueOnce([{ + sessionId: hit.id, + status: 'rejected', + reason: failure, + }]) + + const result = await mounted.call('session_search', { query: 'needle' }) + + expect(errorCode(result)).toBe('SESSION_QUERY_TOOL_UNAUTHORIZED') + expect(text(result)).not.toContain('title unavailable') }) it('passes the exact execution signal to every FTS page and stops on cancellation', async () => { @@ -907,9 +959,13 @@ describe('trace and exact read rendering', () => { complete: true, root: targetRecord, }) - vi.spyOn(mounted.ctx.sessionQuery, 'readTitleSnapshot').mockImplementation(sessionId => Promise.resolve({ - session: header(sessionId, '/work'), - })) + vi.spyOn(mounted.ctx.sessionQuery, 'readTitleSnapshots').mockImplementation(sessionIds => Promise.resolve( + [...new Set(sessionIds)].map(sessionId => ({ + sessionId, + status: 'fulfilled' as const, + value: { session: header(sessionId, '/work') }, + })), + )) const output = text(await mounted.call('session_trace', { session_id: target.id })) expect(output).toContain('Descendants:\n- deep-1 —') diff --git a/scripts/gen-cordis-catalog.ts b/scripts/gen-cordis-catalog.ts index a68097864e..3fe3a681de 100644 --- a/scripts/gen-cordis-catalog.ts +++ b/scripts/gen-cordis-catalog.ts @@ -139,6 +139,7 @@ export const LINK_MAP: Record = { SessionSearchPage: 'session-query.md', SessionSearchRequest: 'session-query.md', SessionTitleObservation: 'session-query.md', + SessionTitleObservationResult: 'session-query.md', SessionTitleProvider: 'session-title.md', SessionTitleSnapshot: 'session-title.md', SkillDefinition: 'skills.md', diff --git a/scripts/type-equiv.manifest.json b/scripts/type-equiv.manifest.json index ce9d54c217..2213725f7f 100644 --- a/scripts/type-equiv.manifest.json +++ b/scripts/type-equiv.manifest.json @@ -404,6 +404,11 @@ "symbol": "SessionTitleObservation", "source": "packages/session-query/session-query/src/types.ts" }, + { + "doc": "docs/core-data-structures/session-query.md", + "symbol": "SessionTitleObservationResult", + "source": "packages/session-query/session-query/src/types.ts" + }, { "doc": "docs/core-data-structures/session-query.md", "symbol": "SessionEventRecord",