Merge current master into feat/mcp-auto-reconnect
Master advanced from 1ac6ee70bb24a566cca74d99bdfe5b1bbc092940 to cb7b5af91e9f23bb300a1b7d4fb64b64abbdea01 while the prior retarget's fresh CI was finishing. Preserve both already validated merge checkpoints and incorporate the new live tip with another forward merge instead of rewriting history. The newer base adds session-log format guards and benchmark documentation. It merges automatically, including the generated config-catalog overlap, so no MCP feature resolution is required; this checkpoint makes GitHub evaluate PR #2187 against the master tip current at the final push.
This commit is contained in:
Tianyi Cui
@@ -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 packages/core/session/README.md
|
||||
README.md: db477d94037d3463870fc8e66ea35d5e607fb6fe
|
||||
README.zh.md: 1ce1e823a7e0fdbcf7b6898764a89c52b74adf6a
|
||||
README.md: 57569e9c0dbfa7cb696e3a561a9ff108c2ac981f
|
||||
README.zh.md: 16629dc70c79ca838ba7088aeafcc5b38b124f87
|
||||
|
||||
@@ -76,10 +76,11 @@ Also defines `TurnEndReasonMap`, the merge-extensible `kind`-tagged sum type for
|
||||
|
||||
An interrupted live turn ends with `{ kind: 'aborted', reason: AgentCancelCause }`, preserving the typed cancellation cause in the durable transcript. Persistence imports the coarse aborted outcome from the supported older format as `{ kind: 'aborted', reason: { kind: 'legacy' } }`, because that record did not retain its caller. A turn failure carries `{ kind: 'error', error }`; crash recovery alone synthesizes `{ kind: 'interrupted' }`.
|
||||
|
||||
Every `SessionEvent` carries two optional top-level fields (structural metadata):
|
||||
Every `SessionEvent` carries three optional top-level fields (structural metadata):
|
||||
|
||||
- `sourceEventSeqs?: number[]` — seq numbers of earlier events cited as sources (e.g., the `assistant/chunk` seqs behind an `assistant/message`, or the shadowed entries behind a compaction replacement entry). On `assistant/message`, a present `[]` records a known empty provider stream, while omission means a legacy or foreign event did not record the source stream; other surface events require a non-empty list when this field is present.
|
||||
- `surfaceOp?: SurfaceOp` — how this event entered the surface. Absent for non-surface events (boundaries, chunks, usage, errors).
|
||||
- `ignorable?: true` — marks an event a reader may safely skip when it does not recognize the type; absent means required, so an unknown-type event refuses session reconstruction ([mechanism](../../../.agents/notes/implemented/architecture/2026-08-10-session-log-version-mechanism.md)).
|
||||
|
||||
### Metadata types (`types.ts`)
|
||||
|
||||
@@ -139,5 +140,5 @@ Logging causes no invalidation, and exact reconstruction preserves request-prefi
|
||||
|
||||
- **Session branching/tree** (pi-style entry tree) — deferred unless needed beyond boundary-based `fork()`.
|
||||
- **`fork()` cuts only at stable boundaries of live sessions** — the selected prefix must end outside an open turn and the source must be in the store; forking a persisted-but-unloaded session is excluded from the [fork API](../../../.agents/notes/implemented/feature/2026-06-30-session-store-fork-api.md).
|
||||
- **`SESSION_FORMAT_VERSION` stays pinned at `0`** — pre-release, no broad compatibility implied: `Session` accepts only current seed shapes and a backend rejects any other version. Narrow storage import upgrades belong to the persistence boundary ([policy](../../../AGENTS.md), [pre-identity message recovery](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)).
|
||||
- **`SESSION_FORMAT_VERSION` stays pinned at `0`** — pre-release, no broad compatibility implied: `Session` accepts only current seed shapes, and a backend refuses any other version naming the direction (newer: "written by a newer harness — upgrade"; older: no upgrade path ships yet). Unknown event types refuse the same way unless marked `ignorable` in the envelope; the versioning mechanism is the [session-log-version-mechanism note](../../../.agents/notes/implemented/architecture/2026-08-10-session-log-version-mechanism.md). Narrow storage import upgrades belong to the persistence boundary ([policy](../../../AGENTS.md), [pre-identity message recovery](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)).
|
||||
- **`TurnEndReasonMap` omits the ACP-named `refusal` / `max_turn_requests` variants** — producer-gated: they land when an adapter or the loop first emits them.
|
||||
|
||||
@@ -76,10 +76,11 @@
|
||||
|
||||
被中断的实时轮次以 `{ kind: 'aborted', reason: AgentCancelCause }` 结束，在持久 transcript（文本记录）中保留类型化取消原因。持久化会将受支持旧格式中的粗粒度中止结果导入为 `{ kind: 'aborted', reason: { kind: 'legacy' } }`，因为该记录没有保留调用方。轮次失败携带 `{ kind: 'error', error }`；只有崩溃恢复会合成 `{ kind: 'interrupted' }`。
|
||||
|
||||
每个 `SessionEvent` 都有两个可选顶层字段（结构元数据）：
|
||||
每个 `SessionEvent` 都有三个可选顶层字段（结构元数据）：
|
||||
|
||||
- `sourceEventSeqs?: number[]`：被引用为来源的较早事件 seq（例如 `assistant/message` 引用的 `assistant/chunk` seq，或压缩替换条目引用的已遮蔽条目）。对于 `assistant/message`，存在的 `[]` 表示已知提供方流为空；省略则表示旧版或外部事件没有记录源流。其他 surface 事件若有此字段，则要求非空列表。
|
||||
- `surfaceOp?: SurfaceOp`：事件进入 surface 的方式。非 surface 事件（边界、分片、用量、错误）不含该字段。
|
||||
- `ignorable?: true`：标记读取器在不认识事件类型时可以安全跳过该事件；缺失表示必需，不认识的事件类型会使会话重建被拒绝（[机制](../../../.agents/notes/implemented/architecture/2026-08-10-session-log-version-mechanism.md)）。
|
||||
|
||||
### 元数据类型（`types.ts`）
|
||||
|
||||
@@ -139,5 +140,5 @@
|
||||
|
||||
- **会话分支／树**（pi 风格条目树）：除非需要超越基于边界的 `fork()` 能力，否则暂缓。
|
||||
- **`fork()` 仅在实时会话的稳定边界处切分**：所选前缀结束时不得有开放轮次，且源会话必须位于存储中；[fork API](../../../.agents/notes/implemented/feature/2026-06-30-session-store-fork-api.md) 不支持对已持久化但未加载的会话进行 fork。
|
||||
- **`SESSION_FORMAT_VERSION` 固定为 `0`**：预发布阶段不承诺广泛兼容性；`Session` 只接受当前 seed 形状，后端会拒绝其他任何版本。范围受限的存储导入升级应由持久化边界负责（[政策](../../../AGENTS.md)、[消息标识机制引入前的消息恢复](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)）。
|
||||
- **`SESSION_FORMAT_VERSION` 固定为 `0`**：预发布阶段不承诺广泛兼容性；`Session` 只接受当前 seed 形状，后端拒绝其他任何版本并说明方向（更新的版本提示"由更新的 harness 写入，请升级"；更旧的版本说明尚无升级路径）。不认识的事件类型同样被拒绝，除非信封带 `ignorable` 标记；版本机制见 [session-log 版本机制 Agent Note](../../../.agents/notes/implemented/architecture/2026-08-10-session-log-version-mechanism.md)。范围受限的存储导入升级应由持久化边界负责（[政策](../../../AGENTS.md)、[消息标识机制引入前的消息恢复](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)）。
|
||||
- **`TurnEndReasonMap` 不含 ACP（Agent Client Protocol）命名的 `refusal`／`max_turn_requests` 变体**：受生产方约束；只有当适配器或循环首次产生这些变体时才加入。
|
||||
|
||||
@@ -32,6 +32,7 @@ export type { ChunkRow, StorageRecord } from './chunk-rows.ts'
|
||||
export type { SessionSurface, SurfaceFoldReplacement, SurfaceFoldResult } from './surface.ts'
|
||||
export { deriveEventMessage, foldSurface, isAppendSurfaceEvent, isReplacementSurfaceEvent, isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
|
||||
export { canonicalHeader, foldRequestHeader, headerEquals } from './request-header.ts'
|
||||
export { KNOWN_SESSION_EVENT_TYPES } from './known-event-types.ts'
|
||||
|
||||
/**
|
||||
* Find the latest closed turn that entered at least one model step, ignoring
|
||||
@@ -243,6 +244,7 @@ function assertSessionEventEnvelope(value: Record<string, unknown>, index: numbe
|
||||
case 'data':
|
||||
case 'surfaceOp':
|
||||
case 'sourceEventSeqs':
|
||||
case 'ignorable':
|
||||
break
|
||||
default:
|
||||
throw new Error(`seed event at index ${index} has an invalid event envelope`)
|
||||
@@ -254,7 +256,8 @@ function assertSessionEventEnvelope(value: Record<string, unknown>, index: numbe
|
||||
if (typeof type !== 'string'
|
||||
|| typeof seq !== 'number' || !Number.isSafeInteger(seq) || seq < 0
|
||||
|| typeof time !== 'number' || !Number.isSafeInteger(time)
|
||||
|| event['data'] === undefined) {
|
||||
|| event['data'] === undefined
|
||||
|| (event['ignorable'] !== undefined && event['ignorable'] !== true)) {
|
||||
throw new Error(`seed event at index ${index} has an invalid event envelope`)
|
||||
}
|
||||
switch (type) {
|
||||
|
||||
59
packages/core/session/src/known-event-types.ts
Normal file
59
packages/core/session/src/known-event-types.ts
Normal file
@@ -0,0 +1,59 @@
|
||||
/**
|
||||
* GENERATED by `scripts/gen-persistence-catalog.ts` — do not edit by hand; run
|
||||
* `pnpm run gen-persistence-catalog` to regenerate (verified fresh by
|
||||
* `pnpm run verify-persistence-catalog`, part of `doc-sync`).
|
||||
* @module @deepseek-ai/dsh-session/known-event-types
|
||||
*/
|
||||
|
||||
/**
|
||||
* Every `SessionEventMap` member declared in this repository — the event
|
||||
* vocabulary this build understands. The persistence read path refuses to
|
||||
* interpret a log containing a type outside this set unless the event
|
||||
* carries the envelope's `ignorable` marker (see `SessionEvent.ignorable`
|
||||
* in `./types.ts`): such a log was likely written by a newer harness, and
|
||||
* silently skipping a required event would reconstruct a wrong session.
|
||||
* Downstream (out-of-repo) plugin events are outside this list by
|
||||
* construction; a registration surface for them is deferred until such a
|
||||
* consumer exists.
|
||||
*/
|
||||
export const KNOWN_SESSION_EVENT_TYPES: ReadonlySet<string> = new Set([
|
||||
'agent-preset/selected',
|
||||
'agent/inbox/spliced',
|
||||
'approval/asked',
|
||||
'approval/decided',
|
||||
'approval/policy',
|
||||
'assistant/chunk',
|
||||
'assistant/message',
|
||||
'command/done',
|
||||
'command/run',
|
||||
'compact/end',
|
||||
'compact/prune',
|
||||
'compact/start',
|
||||
'compact/summary',
|
||||
'feedback/record',
|
||||
'goal/change',
|
||||
'hook/invoked',
|
||||
'hook/result',
|
||||
'llm/retry',
|
||||
'llm/retry-started',
|
||||
'permission/preset',
|
||||
'plan/mode',
|
||||
'request/context',
|
||||
'request/header',
|
||||
'sandbox/mode',
|
||||
'session/end-seed',
|
||||
'session/title',
|
||||
'session/title-llm-request',
|
||||
'step/end',
|
||||
'step/start',
|
||||
'subagent/descriptor',
|
||||
'todo/write',
|
||||
'tool/call',
|
||||
'tool/code-dispatch',
|
||||
'tool/code-dispatch-start',
|
||||
'tool/result',
|
||||
'turn/end',
|
||||
'turn/start',
|
||||
'user/message',
|
||||
'web/deepseek-search-llm-request',
|
||||
])
|
||||
@@ -30,8 +30,23 @@ export function SessionId(id: string): SessionId {
|
||||
* and enforced by every persistence backend on load. The single source of truth for the
|
||||
* version — write sites and the load-time check all read it.
|
||||
* While the harness is unreleased it is pinned at `0`: no compatibility is
|
||||
* implied, incompatible logs are rejected, and no migration is provided. A
|
||||
* monotonic version policy starts with the first tagged release.
|
||||
* implied, incompatible logs are rejected, and no migration is provided.
|
||||
*
|
||||
* The version is a single monotonic integer with no major/minor split. Whether
|
||||
* a bump is needed is decided by what the WRITER emits, never by what a newer
|
||||
* reader can accept: bump exactly when an older runtime could no longer handle
|
||||
* a new log with full semantic correctness ("parses without error" is not
|
||||
* correctness — silently skipping content that shapes reconstruction is a
|
||||
* wrong read). Only structural changes reach that bar: the header shape, the
|
||||
* {@link SessionEvent} envelope, core event semantics, or the surface
|
||||
* mechanism (the {@link SurfaceEventType} set and {@link SurfaceOp} variants).
|
||||
* Adding an ordinary event type does not bump — the per-event
|
||||
* {@link SessionEvent.ignorable} guard covers vocabulary growth instead. When
|
||||
* in doubt, bump: a near-identity upgrade step is almost free, a missed bump
|
||||
* makes older runtimes read new logs wrong silently. The full mechanism
|
||||
* (upgrade-step chain, in-memory view conversion, migrate-on-continue) is
|
||||
* recorded in the session-log-version-mechanism Agent Note
|
||||
* (`.agents/notes/implemented/architecture/2026-08-10-session-log-version-mechanism.md`).
|
||||
*/
|
||||
export const SESSION_FORMAT_VERSION = 0
|
||||
|
||||
@@ -389,6 +404,17 @@ export type SessionEvent<T extends SessionEventType = SessionEventType> = {
|
||||
/** Unix epoch milliseconds. */
|
||||
time: number
|
||||
data: SessionEventMap[K]
|
||||
/**
|
||||
* Marks an event a reader may safely skip when it does not recognize
|
||||
* `type`. Absent means required: a reader meeting an unrecognized type
|
||||
* without this marker MUST refuse to reconstruct the session instead of
|
||||
* silently dropping the event, because an unrecognized required event may
|
||||
* change how the rest of the log is interpreted. A writer sets `true` only
|
||||
* on purely informational records whose loss cannot affect reconstruction;
|
||||
* defaulting to required means a forgotten marker over-refuses (an
|
||||
* inconvenience) rather than silently resuming a gutted session.
|
||||
*/
|
||||
ignorable?: true
|
||||
} & (K extends SurfaceEventType ? {
|
||||
/**
|
||||
* Seq numbers of earlier events that this event cites as sources
|
||||
|
||||
@@ -1090,12 +1090,20 @@ describe('Session', () => {
|
||||
{ ...base, time: '1' },
|
||||
{ ...base, time: 0.5 },
|
||||
{ type: base.type, seq: base.seq, time: base.time },
|
||||
{ ...base, ignorable: false },
|
||||
{ ...base, ignorable: 'yes' },
|
||||
]
|
||||
|
||||
for (const [index, event] of cases.entries()) {
|
||||
expect(() => Session.create(SessionId(`bad-envelope-${index}`), [event as SessionEvent]))
|
||||
.toThrow(/invalid event envelope/)
|
||||
}
|
||||
|
||||
// `ignorable: true` is the one accepted marker value (unknown-type skip contract).
|
||||
const marked = Session.create(SessionId('ignorable-envelope'), [
|
||||
{ ...base, ignorable: true } as SessionEvent,
|
||||
])
|
||||
expect(marked.events[0]?.ignorable).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@@ -45,6 +45,7 @@ export const sessionEventSchema = z.object({
|
||||
data: z.unknown(),
|
||||
sourceEventSeqs: z.array(z.number()).optional(),
|
||||
surfaceOp: z.unknown().optional(),
|
||||
ignorable: z.literal(true).optional(),
|
||||
}) as unknown as z.ZodType<SessionEvent>
|
||||
|
||||
/** SessionSummary row of session.list (`projections` reuses the history block's shape and schema). */
|
||||
|
||||
@@ -2633,7 +2633,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
},
|
||||
{
|
||||
name: 'SessionEvent',
|
||||
declaration: 'export type SessionEvent<T extends SessionEventType = SessionEventType> = {\n [K in SessionEventType]: {\n type: K;\n seq: number;\n time: number;\n data: SessionEventMap[K];\n } & (K extends SurfaceEventType ? {\n sourceEventSeqs?: number[];\n surfaceOp?: SurfaceOp;\n } : object);\n}[T];',
|
||||
declaration: 'export type SessionEvent<T extends SessionEventType = SessionEventType> = {\n [K in SessionEventType]: {\n type: K;\n seq: number;\n time: number;\n data: SessionEventMap[K];\n ignorable?: true;\n } & (K extends SurfaceEventType ? {\n sourceEventSeqs?: number[];\n surfaceOp?: SurfaceOp;\n } : object);\n}[T];',
|
||||
},
|
||||
{
|
||||
name: 'SessionEventMap',
|
||||
|
||||
@@ -9,8 +9,9 @@
|
||||
*/
|
||||
|
||||
import { join } from 'node:path'
|
||||
import { decodeStorageRecord, packChunkRuns } from '@deepseek-ai/dsh-session'
|
||||
import { decodeStorageRecord, packChunkRuns, SESSION_FORMAT_VERSION } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionEvent, SessionHeader, SessionId, StorageRecord } from '@deepseek-ai/dsh-session'
|
||||
import { SessionFormatUnsupportedError, sessionFormatVersionRefusal } from '@deepseek-ai/dsh-session-persistence'
|
||||
|
||||
/** Physical encoding selected for JSONL session artifacts. */
|
||||
export type JsonlCompression = 'zstd' | 'none'
|
||||
@@ -229,6 +230,22 @@ interface SessionLogScan {
|
||||
}
|
||||
|
||||
/** Parse one complete header record supplied independently from event rows. */
|
||||
/**
|
||||
* Refuse a header carrying a format version this build does not read BEFORE
|
||||
* validating the current header shape or decoding any event row: a future
|
||||
* format need not satisfy today's structural checks at all, and its user must
|
||||
* see "upgrade the harness", never "corrupt session log".
|
||||
* @param parsed - the JSON-parsed first line of a session artifact.
|
||||
*/
|
||||
function refuseForeignFormatVersion(parsed: unknown): void {
|
||||
if (typeof parsed !== 'object' || parsed === null) return
|
||||
const { version, id } = parsed as { version?: unknown; id?: unknown }
|
||||
if (typeof version !== 'number' || version === SESSION_FORMAT_VERSION) return
|
||||
throw new SessionFormatUnsupportedError(
|
||||
sessionFormatVersionRefusal(typeof id === 'string' ? id : String(id), version),
|
||||
)
|
||||
}
|
||||
|
||||
function parseHeaderRecord(record: Buffer): SessionHeader {
|
||||
if (record.length === 0 || record.at(-1) !== 0x0A || record.indexOf(0x0A) !== record.length - 1) {
|
||||
throw new Error('empty or header-less session log')
|
||||
@@ -239,6 +256,7 @@ function parseHeaderRecord(record: Buffer): SessionHeader {
|
||||
} catch {
|
||||
throw new Error('corrupt session log: header line is not valid JSON')
|
||||
}
|
||||
refuseForeignFormatVersion(parsed)
|
||||
if (!isHeaderLine(parsed)) {
|
||||
throw new Error('corrupt session log: first line is not a session header')
|
||||
}
|
||||
|
||||
@@ -16,7 +16,7 @@ import { scheduler } from 'node:timers/promises'
|
||||
import { randomBytes } from 'node:crypto'
|
||||
import {
|
||||
DEFAULT_PREPARED_SESSION_CACHE_SIZE, DEFAULT_WRITE_BATCH_MAX_DELAY_MS, MAX_WRITE_BATCH_DELAY_MS,
|
||||
SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
|
||||
SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, SessionFormatUnsupportedError,
|
||||
type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot,
|
||||
type SessionInspection, type SessionPersistenceRevision as PersistenceRevision, type StoredPrefix,
|
||||
} from '@deepseek-ai/dsh-session-persistence'
|
||||
@@ -256,19 +256,29 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
}
|
||||
}
|
||||
let prefix: Omit<StoredPrefix<JsonlTornMarker>, 'revision'>
|
||||
if (this.compression === 'zstd') {
|
||||
prefix = await this.readZstdPrefix(buffer, signal)
|
||||
} else {
|
||||
signal?.throwIfAborted()
|
||||
const { meta, events, committedBytes } = scanLog(buffer)
|
||||
signal?.throwIfAborted()
|
||||
prefix = {
|
||||
meta,
|
||||
events,
|
||||
...committedBytes < buffer.byteLength
|
||||
? { tornMarker: { truncateTo: committedBytes, recoveredEvents: [] } }
|
||||
: {},
|
||||
try {
|
||||
if (this.compression === 'zstd') {
|
||||
prefix = await this.readZstdPrefix(buffer, signal)
|
||||
} else {
|
||||
signal?.throwIfAborted()
|
||||
const { meta, events, committedBytes } = scanLog(buffer)
|
||||
signal?.throwIfAborted()
|
||||
prefix = {
|
||||
meta,
|
||||
events,
|
||||
...committedBytes < buffer.byteLength
|
||||
? { tornMarker: { truncateTo: committedBytes, recoveredEvents: [] } }
|
||||
: {},
|
||||
}
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
// A parse-time format refusal predates any SessionHeader, so the
|
||||
// coordinator's locate-based enrichment cannot run; attach the artifact
|
||||
// this read actually refused.
|
||||
if (error instanceof SessionFormatUnsupportedError && error.location === undefined) {
|
||||
throw new SessionFormatUnsupportedError(`${error.message} (raw log: ${path})`, { kind: 'jsonl', path })
|
||||
}
|
||||
throw error
|
||||
}
|
||||
signal?.throwIfAborted()
|
||||
await this.assertStoredIdentity(path, prefix.meta, expectedId, signal)
|
||||
|
||||
@@ -3,7 +3,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat, symlink } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { isAbsolute, join, relative, resolve } from 'node:path'
|
||||
import { dirname, isAbsolute, join, relative, resolve } from 'node:path'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
@@ -186,6 +186,76 @@ describe('SessionPersistenceJsonl: format helpers', () => {
|
||||
})
|
||||
await fiber.dispose()
|
||||
})
|
||||
|
||||
it('refuses a structurally foreign future header as unsupported, not corrupt', async () => {
|
||||
const absoluteRoot = await freshRoot()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: absoluteRoot, compression: 'none' })
|
||||
// A future format need not satisfy today's header shape at all (no
|
||||
// createdAt, unknown fields): the version must be refused before shape
|
||||
// validation, so the user sees the upgrade direction.
|
||||
const id = SessionId('future-shape')
|
||||
const path = rawLogPath(resolve(absoluteRoot), '/work', id)
|
||||
await mkdir(dirname(path), { recursive: true })
|
||||
await writeFile(path, `${JSON.stringify({ type: 'session', version: 42, id, futureOnly: true })}\n{"future":"row"}\n`)
|
||||
const failure = await ctx.sessionPersistence.load(id).then(() => undefined, (error: unknown) => error as Error)
|
||||
expect(failure?.name).toBe('SessionFormatUnsupportedError')
|
||||
expect(failure?.message).toMatch(/written by a newer harness.*upgrade the harness/)
|
||||
expect(failure?.message).toContain(`(raw log: ${path})`)
|
||||
await fiber.dispose()
|
||||
})
|
||||
|
||||
it('keeps a non-object header line a corruption, not a format refusal', async () => {
|
||||
const absoluteRoot = await freshRoot()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: absoluteRoot, compression: 'none' })
|
||||
// Valid JSON that is no object carries no version to compare, so the
|
||||
// version guard must pass it through to the corruption diagnostics.
|
||||
const id = SessionId('scalar-header')
|
||||
const path = rawLogPath(resolve(absoluteRoot), '/work', id)
|
||||
await mkdir(dirname(path), { recursive: true })
|
||||
await writeFile(path, '42\n')
|
||||
const failure = await ctx.sessionPersistence.load(id).then(() => undefined, (error: unknown) => error as Error)
|
||||
expect(failure?.name).not.toBe('SessionFormatUnsupportedError')
|
||||
expect(failure?.message).toContain('first line is not a session header')
|
||||
await fiber.dispose()
|
||||
})
|
||||
|
||||
it('names a foreign-version header by its stringified non-string id', async () => {
|
||||
const absoluteRoot = await freshRoot()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: absoluteRoot, compression: 'none' })
|
||||
// A future header's id field is as untrusted as the rest of its shape:
|
||||
// the refusal must still name the session it read, not crash on the type.
|
||||
const id = SessionId('numeric-id')
|
||||
const path = rawLogPath(resolve(absoluteRoot), '/work', id)
|
||||
await mkdir(dirname(path), { recursive: true })
|
||||
await writeFile(path, `${JSON.stringify({ type: 'session', version: 42, id: 123 })}\n`)
|
||||
const failure = await ctx.sessionPersistence.load(id).then(() => undefined, (error: unknown) => error as Error)
|
||||
expect(failure?.name).toBe('SessionFormatUnsupportedError')
|
||||
expect(failure?.message).toContain('session "123" uses log format v42')
|
||||
await fiber.dispose()
|
||||
})
|
||||
|
||||
it('points a format refusal at the raw log path', async () => {
|
||||
const absoluteRoot = await freshRoot()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: absoluteRoot, compression: 'none' })
|
||||
const m = { ...meta('newer-format', '/work'), version: 7 }
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
|
||||
{ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
])
|
||||
const failure = await ctx.sessionPersistence.load(m.id).then(() => undefined, (error: unknown) => error as Error)
|
||||
expect(failure?.name).toBe('SessionFormatUnsupportedError')
|
||||
expect(failure?.message).toContain(`(raw log: ${rawLogPath(resolve(absoluteRoot), '/work', m.id)})`)
|
||||
await fiber.dispose()
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
|
||||
@@ -28,15 +28,18 @@ import {
|
||||
export { SCHEMA_VERSION } from './schema.ts'
|
||||
|
||||
/**
|
||||
* Serialize an event's surface-metadata fields for SQL binding. Both fields are
|
||||
* nullable TEXT columns — null when the event has no surface metadata (non-surface
|
||||
* events, events written before surface support).
|
||||
* Serialize an event's optional envelope fields for SQL binding. The surface
|
||||
* fields are nullable TEXT columns — null when the event has no surface
|
||||
* metadata (non-surface events, events written before surface support); the
|
||||
* ignorable marker is a nullable INTEGER column — `1` iff the envelope carries
|
||||
* `ignorable: true`.
|
||||
*/
|
||||
function surfaceBindings(event: SessionEvent): [string | null, string | null] {
|
||||
function envelopeBindings(event: SessionEvent): [string | null, string | null, number | null] {
|
||||
const se = event as SessionEvent<SurfaceEventType>
|
||||
return [
|
||||
se.sourceEventSeqs ? JSON.stringify(se.sourceEventSeqs) : null,
|
||||
se.surfaceOp !== undefined ? JSON.stringify(se.surfaceOp) : null,
|
||||
event.ignorable === true ? 1 : null,
|
||||
]
|
||||
}
|
||||
|
||||
@@ -225,7 +228,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
|
||||
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 = ? AND seq >= ? ORDER BY seq')
|
||||
.prepare('SELECT seq, type, time, data, source_event_seqs, surface_op, ignorable FROM events WHERE session_id = ? AND seq >= ? ORDER BY seq')
|
||||
.all(id, fromSeq) as unknown as EventRow[]
|
||||
signal?.throwIfAborted()
|
||||
const { preserved } = scanRows(eventRows, fromSeq)
|
||||
@@ -247,7 +250,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
|
||||
const row = this.rowFor(id)
|
||||
if (row !== undefined) {
|
||||
const eventRows = this.db
|
||||
.prepare('SELECT seq, type, time, data, source_event_seqs, surface_op FROM events WHERE session_id = ? ORDER BY seq')
|
||||
.prepare('SELECT seq, type, time, data, source_event_seqs, surface_op, ignorable FROM events WHERE session_id = ? ORDER BY seq')
|
||||
.all(id) as unknown as EventRow[]
|
||||
snapshot = { row, eventRows }
|
||||
}
|
||||
@@ -279,14 +282,14 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
|
||||
async appendBatch(meta: SessionHeader, events: readonly SessionEvent[], isMaterialized: boolean): Promise<void> {
|
||||
await this.ready
|
||||
const insertEvent = this.db.prepare(
|
||||
'INSERT INTO events (session_id, seq, type, time, data, source_event_seqs, surface_op) VALUES (?, ?, ?, ?, ?, ?, ?)',
|
||||
'INSERT INTO events (session_id, seq, type, time, data, source_event_seqs, surface_op, ignorable) VALUES (?, ?, ?, ?, ?, ?, ?, ?)',
|
||||
)
|
||||
this.db.exec('BEGIN')
|
||||
try {
|
||||
if (!isMaterialized) this.writeRow(meta)
|
||||
for (const event of events) {
|
||||
const [surfaceSeqs, surfaceOp] = surfaceBindings(event)
|
||||
insertEvent.run(meta.id, event.seq, event.type, event.time, JSON.stringify(event.data), surfaceSeqs, surfaceOp)
|
||||
const [surfaceSeqs, surfaceOp, ignorable] = envelopeBindings(event)
|
||||
insertEvent.run(meta.id, event.seq, event.type, event.time, JSON.stringify(event.data), surfaceSeqs, surfaceOp, ignorable)
|
||||
}
|
||||
this.db.prepare('UPDATE sessions SET revision = revision + 1 WHERE id = ?').run(meta.id)
|
||||
this.db.exec('COMMIT')
|
||||
@@ -310,11 +313,11 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
|
||||
}
|
||||
if (closers.length > 0) {
|
||||
const insertEvent = this.db.prepare(
|
||||
'INSERT INTO events (session_id, seq, type, time, data, source_event_seqs, surface_op) VALUES (?, ?, ?, ?, ?, ?, ?)',
|
||||
'INSERT INTO events (session_id, seq, type, time, data, source_event_seqs, surface_op, ignorable) VALUES (?, ?, ?, ?, ?, ?, ?, ?)',
|
||||
)
|
||||
for (const event of closers) {
|
||||
const [surfaceSeqs, surfaceOp] = surfaceBindings(event)
|
||||
insertEvent.run(meta.id, event.seq, event.type, event.time, JSON.stringify(event.data), surfaceSeqs, surfaceOp)
|
||||
const [surfaceSeqs, surfaceOp, ignorable] = envelopeBindings(event)
|
||||
insertEvent.run(meta.id, event.seq, event.type, event.time, JSON.stringify(event.data), surfaceSeqs, surfaceOp, ignorable)
|
||||
}
|
||||
}
|
||||
if (tornMarker !== undefined || closers.length > 0) {
|
||||
|
||||
@@ -17,7 +17,7 @@ import type { SessionEvent, SessionId, SessionHeader, SurfaceOp } from '@deepsee
|
||||
* layout; orthogonal to a session's own `version` (which versions the EVENT
|
||||
* vocabulary, stored per session in the `sessions` row).
|
||||
*/
|
||||
export const SCHEMA_VERSION = 14
|
||||
export const SCHEMA_VERSION = 15
|
||||
|
||||
/** SQLite application id protecting unrelated databases from persistence writes. */
|
||||
export const SESSION_PERSISTENCE_SQLITE_APPLICATION_ID = 0x44534850
|
||||
@@ -55,6 +55,8 @@ export interface EventRow {
|
||||
source_event_seqs: string | null
|
||||
/** JSON-encoded `SurfaceOp` — how the event entered the surface, or null. */
|
||||
surface_op: string | null
|
||||
/** `1` iff the event carries the envelope's `ignorable: true` marker, else null. */
|
||||
ignorable: number | null
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -139,6 +141,7 @@ function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalM
|
||||
data TEXT NOT NULL,
|
||||
source_event_seqs TEXT,
|
||||
surface_op TEXT,
|
||||
ignorable INTEGER,
|
||||
PRIMARY KEY (session_id, seq)
|
||||
) STRICT
|
||||
`)
|
||||
@@ -203,12 +206,14 @@ export function rowToEvent(row: EventRow): SessionEvent {
|
||||
...row.source_event_seqs !== null ? { sourceEventSeqs: JSON.parse(row.source_event_seqs) as number[] } : {},
|
||||
...row.surface_op !== null ? { surfaceOp: JSON.parse(row.surface_op) as SurfaceOp } : {},
|
||||
}
|
||||
const ignorableField = row.ignorable === 1 ? { ignorable: true as const } : {}
|
||||
return {
|
||||
type: row.type as SessionEvent['type'],
|
||||
seq: row.seq,
|
||||
time: row.time,
|
||||
data: JSON.parse(row.data) as SessionEvent['data'],
|
||||
...surfaceFields,
|
||||
...ignorableField,
|
||||
} as SessionEvent
|
||||
}
|
||||
|
||||
|
||||
@@ -92,6 +92,7 @@ describe('scanRows', () => {
|
||||
seq: e.seq, type: e.type, time: e.time, data: JSON.stringify(e.data),
|
||||
source_event_seqs: se.sourceEventSeqs !== undefined ? JSON.stringify(se.sourceEventSeqs) : null,
|
||||
surface_op: se.surfaceOp !== undefined ? JSON.stringify(se.surfaceOp) : null,
|
||||
ignorable: e.ignorable === true ? 1 : null,
|
||||
}
|
||||
})
|
||||
|
||||
@@ -142,8 +143,8 @@ describe('scanRows', () => {
|
||||
|
||||
it('throws on an unparsable row inside the committed region', () => {
|
||||
const withCorruptCommitted: EventRow[] = [
|
||||
{ seq: 0, type: 'turn/start', time: 1, data: '{not json', source_event_seqs: null, surface_op: null }, // corrupt, sits before a turn/end
|
||||
{ seq: 1, type: 'turn/end', time: 2, data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }), source_event_seqs: null, surface_op: null },
|
||||
{ seq: 0, type: 'turn/start', time: 1, data: '{not json', source_event_seqs: null, surface_op: null, ignorable: null }, // corrupt, sits before a turn/end
|
||||
{ seq: 1, type: 'turn/end', time: 2, data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }), source_event_seqs: null, surface_op: null, ignorable: null },
|
||||
]
|
||||
expect(() => scanRows(withCorruptCommitted)).toThrow(/unparsable committed event/)
|
||||
})
|
||||
@@ -151,7 +152,7 @@ describe('scanRows', () => {
|
||||
it('tolerates an unparsable torn-tail row after the last turn/end', () => {
|
||||
const withCorruptTail: EventRow[] = [
|
||||
...rows(oneTurnLog()),
|
||||
{ seq: 6, type: 'turn/start', time: 7, data: '{not json', source_event_seqs: null, surface_op: null }, // torn fragment, no committed turn/end after
|
||||
{ seq: 6, type: 'turn/start', time: 7, data: '{not json', source_event_seqs: null, surface_op: null, ignorable: null }, // torn fragment, no committed turn/end after
|
||||
]
|
||||
const { preserved, tornFrom } = scanRows(withCorruptTail)
|
||||
expect(preserved).toEqual(oneTurnLog())
|
||||
@@ -658,7 +659,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
})
|
||||
|
||||
it('exposes the schema version constant', () => {
|
||||
expect(SCHEMA_VERSION).toBe(14)
|
||||
expect(SCHEMA_VERSION).toBe(15)
|
||||
})
|
||||
|
||||
it('keeps the revision stable for an empty repair hook', async () => {
|
||||
@@ -857,6 +858,7 @@ describe('surface field round-trip', () => {
|
||||
data: JSON.stringify({ turn: 1, step: 1, content: [] }),
|
||||
source_event_seqs: JSON.stringify([3, 5]),
|
||||
surface_op: JSON.stringify('append'),
|
||||
ignorable: null,
|
||||
}
|
||||
const event = rowToEvent(row)
|
||||
expect((event as SurfaceEvent).sourceEventSeqs).toEqual([3, 5])
|
||||
@@ -869,6 +871,7 @@ describe('surface field round-trip', () => {
|
||||
data: JSON.stringify({ turn: 1, step: 1, content: [] }),
|
||||
source_event_seqs: JSON.stringify([0, 1]),
|
||||
surface_op: JSON.stringify({ op: 'replace', start: 0, end: 1 }),
|
||||
ignorable: null,
|
||||
}
|
||||
const event = rowToEvent(row)
|
||||
expect((event as SurfaceEvent).sourceEventSeqs).toEqual([0, 1])
|
||||
@@ -879,10 +882,10 @@ describe('surface field round-trip', () => {
|
||||
const rows: EventRow[] = [
|
||||
{ seq: 0, type: 'user/message', time: 1,
|
||||
data: JSON.stringify({ content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }),
|
||||
source_event_seqs: null, surface_op: '{"op":"replace","start":0,"end":0}' },
|
||||
source_event_seqs: null, surface_op: '{"op":"replace","start":0,"end":0}', ignorable: null },
|
||||
{ seq: 1, type: 'turn/end', time: 2,
|
||||
data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }),
|
||||
source_event_seqs: null, surface_op: null },
|
||||
source_event_seqs: null, surface_op: null, ignorable: 1 },
|
||||
]
|
||||
const { preserved } = scanRows(rows)
|
||||
expect(preserved).toHaveLength(2)
|
||||
|
||||
@@ -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 packages/session/session-persistence/README.md
|
||||
README.md: 391548b1b896dca14cbe4f4ae55cf4180c4e0ac2
|
||||
README.zh.md: 7213e1ee71ba418ffacc3685df371dcba33588a7
|
||||
README.md: 324c00b3202bd136566137e1bd398b29d2ea4b82
|
||||
README.zh.md: 2ef5e9a90f0323f8edf8fdc4f936c41ca7e08c70
|
||||
|
||||
@@ -14,9 +14,9 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
|
||||
| `create(meta): Promise<void>` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). |
|
||||
| `append(id, events): Promise<void>` | Durably persist a batch. Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. |
|
||||
| `prepare(id, signal?): Promise<SessionPreparation>` | Reserve the exact unpublished Session used by resume. A coordinator reuses an earlier inspection when available, commits pending recovery, and releases an unpublished reservation back to its bounded cache on disposal. |
|
||||
| `load(id): Promise<{ meta; events }>` | Return an immutable balanced logical log after converting supported older records from the same format version and committing cold recovery. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and durably closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption, malformed records, and unknown `version` reject. |
|
||||
| `load(id): Promise<{ meta; events }>` | Return an immutable balanced logical log after converting supported older records from the same format version and committing cold recovery. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and durably closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption and malformed records reject as `SessionPersistenceCorruptionError`, while an unsupported format `version` or an event type unknown to this build (without the envelope's `ignorable` marker) refuses as `SessionFormatUnsupportedError`, naming the refusal direction and the raw log path when the backend keeps one artifact per session. |
|
||||
| `inspect(id, signal?): Promise<{ meta; events }>` | Return an upgraded, validated, deeply frozen logical view without committing recovery or publishing a Session. A cold view receives in-memory synthetic recovery closers while its physical torn tail remains untouched; an already-live view is its current immutable snapshot and may contain an open turn. Coordinator-backed implementations retain the exact cold unpublished Session in a bounded LRU for later `prepare`, but discard and reload it when the stored revision changes. Same-id inspections share an in-flight read. |
|
||||
| `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | Return valid stored events with `seq >= fromSeq` without preparation caching, truncation, closers, or coordinator state. A `fromSeq` at or past the stored end returns an empty event list; a negative or non-safe-integer `fromSeq` rejects. Seek-capable backends (SQLite) read only the suffix unless converting a supported older record requires earlier records; sequential backends (JSONL) parse the whole artifact and skip forward. Intended for checkpoint consumers that apply only events after a stored sequence number. |
|
||||
| `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | Return valid stored events with `seq >= fromSeq` without preparation caching, truncation, closers, or coordinator state. A `fromSeq` at or past the stored end returns an empty event list; a negative or non-safe-integer `fromSeq` rejects. Seek-capable backends (SQLite) read only the suffix unless converting a supported older record requires earlier records; sequential backends (JSONL) parse the whole artifact and skip forward. Unknown-type refusal follows that access pattern: a seek read checks only the returned suffix, while the sequential fallback also refuses on an unknown required event below the window. Intended for checkpoint consumers that apply only events after a stored sequence number. |
|
||||
| `list(signal?): Promise<SessionHeader[]>` | 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(signal?): Promise<SessionPersistenceSnapshot[]>` | 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. The optional signal requests cancellation of backend discovery work; first-party backends settle any started listing work before rejecting so an awaited call is quiescent. |
|
||||
|
||||
|
||||
@@ -14,9 +14,9 @@
|
||||
| `create(meta): Promise<void>` | 注册新会话元数据。可以将物理写入延迟到第一次 `append`（延迟实体化）。 |
|
||||
| `append(id, events): Promise<void>` | 持久保存一个批次。仅追加；任何修复后，第一个事件 `seq` == 已存储 next-seq；非 JSON 可序列化数据会被拒绝，并命名违规类型。 |
|
||||
| `prepare(id, signal?): Promise<SessionPreparation>` | 预留恢复所使用的那个未发布 Session。协调器会尽可能复用之前的检查结果、提交待处理恢复，并在 dispose 时将未发布 reservation 释放回有界缓存。 |
|
||||
| `load(id): Promise<{ meta; events }>` | 转换同一格式版本中受支持的旧记录后，返回不可变、平衡的逻辑日志，并提交冷恢复。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件持久关闭它。只丢弃撕裂尾部碎片；已提交损坏、格式错误的记录和未知 `version` 会被拒绝。 |
|
||||
| `load(id): Promise<{ meta; events }>` | 转换同一格式版本中受支持的旧记录后，返回不可变、平衡的逻辑日志，并提交冷恢复。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件持久关闭它。只丢弃撕裂尾部碎片；已提交损坏和格式错误的记录以 `SessionPersistenceCorruptionError` 拒绝，不支持的格式 `version` 或本构建不认识且信封未带 `ignorable` 标记的事件类型以 `SessionFormatUnsupportedError` 拒绝，消息说明拒绝方向，并在后端为每个会话保留独立文件时给出原始日志路径。 |
|
||||
| `inspect(id, signal?): Promise<{ meta; events }>` | 返回已经升级、验证和深度冻结的逻辑视图，但不提交恢复或发布 Session。冷视图会获得仅存在于内存的合成恢复 closer，物理撕裂尾部保持不变；实时状态下的视图则是当前不可变快照，可能包含开放的轮次。基于协调器的实现会在有界 LRU 中保留该冷状态下未发布的 Session 本身，供后续 `prepare` 使用，但已存储修订值变化后会丢弃并重新读取。同 id 检查共享进行中的读取。 |
|
||||
| `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | 返回 `seq >= fromSeq` 的有效已存储事件，不进入 preparation 缓存、不截断、不合成 closer，也不发布协调器状态。`fromSeq` 达到或超过已存储末尾时返回空事件列表；负数或非安全整数 `fromSeq` 会被拒绝。可寻址后端（SQLite）只读后缀，除非转换受支持的旧记录需要读取更早的记录；顺序后端（JSONL）解析整个产物并向前跳过。供 checkpoint 消费方只应用已存序号之后的事件。 |
|
||||
| `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | 返回 `seq >= fromSeq` 的有效已存储事件，不进入 preparation 缓存、不截断、不合成 closer，也不发布协调器状态。`fromSeq` 达到或超过已存储末尾时返回空事件列表；负数或非安全整数 `fromSeq` 会被拒绝。可寻址后端（SQLite）只读后缀，除非转换受支持的旧记录需要读取更早的记录；顺序后端（JSONL）解析整个产物并向前跳过。未知类型拒绝遵循同一读取方式：寻址读取只检查返回的后缀，顺序回退路径还会拒绝窗口以下的未知必需事件。供 checkpoint 消费方只应用已存序号之后的事件。 |
|
||||
| `list(signal?): Promise<SessionHeader[]>` | 从元数据轻量列出，不解析完整日志。可选信号取消后端列表工作。零事件延迟实体化会话不在 `list` 中。 |
|
||||
| `listSnapshots(signal?): Promise<SessionPersistenceSnapshot[]>` | 返回轻量元数据和每份日志一个不透明、带品牌类型的修订值，不加载事件日志。日志及其后端存储不变时，修订保持相等；append 或变更性 load 修复后会改变；不会仅因两个存储使用相同本地计数器而冲突。可选信号请求取消后端发现工作；第一方后端会先等待所有已启动的列出工作结束，再予以拒绝，因此调用返回拒绝时，相关工作已完全停稳。 |
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@ import { Context } from '@deepseek-ai/cordis'
|
||||
import {
|
||||
adoptSessionEvent,
|
||||
interruptedTurnClosers,
|
||||
KNOWN_SESSION_EVENT_TYPES,
|
||||
SESSION_FORMAT_VERSION,
|
||||
SessionPreparation,
|
||||
snapshotJsonValue,
|
||||
@@ -16,7 +17,7 @@ import {
|
||||
} from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
|
||||
import type { SessionInspection } from './index.ts'
|
||||
import type { SessionInspection, SessionLocation } from './index.ts'
|
||||
import type { SessionPersistenceRevision } from './revision.ts'
|
||||
import { observeQueuedAbort, SessionPreparations } from './preparations.ts'
|
||||
import type { SessionPreparationReservation } from './preparations.ts'
|
||||
@@ -43,6 +44,42 @@ export class SessionPersistenceCorruptionError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The stored log is intact but this runtime cannot faithfully interpret it:
|
||||
* the header carries an unsupported format version, or an event's type is
|
||||
* unknown to this build and the event is not marked ignorable. Distinct from
|
||||
* {@link SessionPersistenceCorruptionError} — nothing is damaged; the raw log
|
||||
* remains readable at {@link location} when the backend keeps one artifact
|
||||
* per session.
|
||||
*/
|
||||
export class SessionFormatUnsupportedError extends Error {
|
||||
/**
|
||||
* @param message - stable reason the log cannot be interpreted, already
|
||||
* including the raw-log path when one exists.
|
||||
* @param location - the backend's artifact location, when one exists.
|
||||
*/
|
||||
constructor(message: string, readonly location?: SessionLocation) {
|
||||
super(message)
|
||||
this.name = 'SessionFormatUnsupportedError'
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Direction-aware refusal text for a stored session whose format version this
|
||||
* build does not read. Shared by the coordinator's load-time check and by
|
||||
* backends that must refuse BEFORE decoding version-dependent structure (a
|
||||
* future format may not satisfy today's structural checks at all, and the
|
||||
* user must see "upgrade the harness", never "corrupt").
|
||||
* @param id - the stored session id, for message context.
|
||||
* @param version - the stored format version.
|
||||
* @returns the stable refusal text, without a raw-log path suffix.
|
||||
*/
|
||||
export function sessionFormatVersionRefusal(id: string, version: number): string {
|
||||
return version > SESSION_FORMAT_VERSION
|
||||
? `session "${id}" uses log format v${version}, but this harness reads only v${SESSION_FORMAT_VERSION}: the log was written by a newer harness — upgrade the harness to open it`
|
||||
: `session "${id}" uses log format v${version}, older than the supported v${SESSION_FORMAT_VERSION}, and this build ships no upgrade path for it`
|
||||
}
|
||||
|
||||
/** Coordinator policy supplied by a concrete persistence backend. */
|
||||
export interface PersistenceCoordinatorOptions {
|
||||
/** Maximum completed unpublished preparations retained for reuse. */
|
||||
@@ -126,6 +163,11 @@ export interface PersistenceBackend<TornMarker = unknown> {
|
||||
* contains a supported legacy shape whose normalization needs earlier
|
||||
* message-identity facts, in which case the coordinator falls back
|
||||
* to the complete stored prefix.
|
||||
* Unknown-type refusal follows the same suffix scope: a seek-capable
|
||||
* backend's `readFrom` checks only the returned suffix, while the
|
||||
* sequential fallback parses the whole artifact and refuses on an unknown
|
||||
* required event anywhere in it — over-refusal on the sequential side is
|
||||
* accepted rather than widening the seek read.
|
||||
* @param id - persisted session id to resolve.
|
||||
* @param fromSeq - first event seq to include (non-negative safe integer,
|
||||
* validated by the coordinator before this hook runs).
|
||||
@@ -156,6 +198,14 @@ export interface PersistenceBackend<TornMarker = unknown> {
|
||||
*/
|
||||
list(signal?: AbortSignal): Promise<SessionHeader[]>
|
||||
|
||||
/**
|
||||
* Optional side-effect-free artifact locator, used to point refusal
|
||||
* diagnostics ({@link SessionFormatUnsupportedError}) at the raw log.
|
||||
* Backends without one artifact per session omit it or return `undefined`.
|
||||
* @param meta - the header whose artifact is requested.
|
||||
*/
|
||||
locate?(meta: SessionHeader): SessionLocation | undefined
|
||||
|
||||
/**
|
||||
* Optional lifecycle teardown (e.g. close a database handle). Awaited by the
|
||||
* coordinator's dispose effect AFTER the quiescence drain. A stateless file
|
||||
@@ -631,9 +681,13 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
|
||||
private async appendCore(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
|
||||
// Every append route converges here: the public service, live write-behind
|
||||
// drains, and HMR seed/suffix adoption. Keep vocabulary rejection at that
|
||||
// shared boundary so a stale JavaScript plugin cannot persist an event that
|
||||
// this same backend will refuse to load.
|
||||
// drains, and HMR seed/suffix adoption. Legacy-shape rejection stays at
|
||||
// this shared boundary so a stale JavaScript plugin cannot persist a
|
||||
// retired shape this backend refuses to load. The unknown-type guard is
|
||||
// deliberately read-side only: an append-time refusal would stall a live
|
||||
// session's durability mid-flight, which costs more than a loud refusal at
|
||||
// the log's next load (trade-off owned by the session-log-version-mechanism
|
||||
// Agent Note).
|
||||
assertSupportedEvents(events, id)
|
||||
if (events.length === 0) return
|
||||
this.preparations.assertWritable(id)
|
||||
@@ -806,7 +860,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
const whole = await this.readStoredPrefix(id, signal)
|
||||
return { meta: whole.meta, events: whole.events.filter(event => event.seq >= fromSeq) }
|
||||
}
|
||||
return { meta: structuredClone(suffix.meta), events: snapshotStoredEvents(suffix.events, id) }
|
||||
const events = snapshotStoredEvents(suffix.events, id)
|
||||
this.assertEventsSupported(suffix.meta, events)
|
||||
return { meta: structuredClone(suffix.meta), events }
|
||||
}
|
||||
const whole = await this.readStoredPrefix(id, signal)
|
||||
// Sequential fallback: contiguous seqs from 0 make the suffix an index slice.
|
||||
@@ -824,9 +880,11 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
if (stored === undefined) throw new Error(`session "${id}" not found`)
|
||||
this.assertStoredId(id, stored.meta)
|
||||
this.assertVersion(stored.meta)
|
||||
const events = snapshotStoredEvents(stored.events, id)
|
||||
this.assertEventsSupported(stored.meta, events)
|
||||
return {
|
||||
meta: structuredClone(stored.meta),
|
||||
events: snapshotStoredEvents(stored.events, id),
|
||||
events,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -839,6 +897,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
this.assertStoredId(id, meta)
|
||||
this.assertVersion(meta)
|
||||
const storedEvents = adoptStoredEvents(events, id)
|
||||
this.assertEventsSupported(meta, storedEvents)
|
||||
|
||||
// Preserve complete interrupted events and synthesize only missing closers.
|
||||
const closers = interruptedTurnClosers(storedEvents).map(adoptSessionEvent)
|
||||
@@ -861,6 +920,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
closers,
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
// An unsupported format is a refusal over an intact log, not damage —
|
||||
// surface it unwrapped so callers can point at the raw artifact.
|
||||
if (error instanceof SessionFormatUnsupportedError) throw error
|
||||
throw new SessionPersistenceCorruptionError(
|
||||
`stored session "${id}" failed validation: ${String(error)}`,
|
||||
{ cause: error },
|
||||
@@ -982,11 +1044,36 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
}
|
||||
|
||||
private assertVersion(meta: SessionHeader): void {
|
||||
if (meta.version !== SESSION_FORMAT_VERSION) {
|
||||
throw new Error(`unsupported session format version ${meta.version} for "${meta.id}" (only v${SESSION_FORMAT_VERSION} is supported)`)
|
||||
if (meta.version === SESSION_FORMAT_VERSION) return
|
||||
throw this.unsupported(meta, sessionFormatVersionRefusal(meta.id, meta.version))
|
||||
}
|
||||
|
||||
/**
|
||||
* Refuse a log containing an event type this build does not know, unless the
|
||||
* writer marked the event ignorable: an unrecognized required event may
|
||||
* change how the rest of the log must be interpreted, so silently skipping
|
||||
* it would reconstruct a wrong session (the envelope contract on
|
||||
* `SessionEvent.ignorable`). Runs on NORMALIZED events — after
|
||||
* `snapshotStoredEvents`/`adoptStoredEvents` has upgraded the legacy shapes
|
||||
* this build still reads and rejected the ones it does not, so those keep
|
||||
* their specific diagnostics.
|
||||
*/
|
||||
private assertEventsSupported(meta: SessionHeader, events: readonly SessionEvent[]): void {
|
||||
for (const event of events) {
|
||||
if (KNOWN_SESSION_EVENT_TYPES.has(event.type) || event.ignorable === true) continue
|
||||
throw this.unsupported(meta, `session "${meta.id}" contains event type "${event.type}" (seq ${event.seq}) unknown to this harness and not marked ignorable; refusing to interpret the log — it was likely written by a newer harness`)
|
||||
}
|
||||
}
|
||||
|
||||
/** Build a format refusal that points at the raw artifact when the backend has one. */
|
||||
private unsupported(meta: SessionHeader, reason: string): SessionFormatUnsupportedError {
|
||||
const location = this.backend.locate?.(meta)
|
||||
return new SessionFormatUnsupportedError(
|
||||
location === undefined ? reason : `${reason} (raw log: ${location.path})`,
|
||||
location,
|
||||
)
|
||||
}
|
||||
|
||||
/** Reject backend metadata that is not bound to the requested session id. */
|
||||
private assertStoredId(id: SessionId, meta: SessionHeader): void {
|
||||
if (meta.id !== id) {
|
||||
@@ -1219,6 +1306,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
}
|
||||
this.assertVersion(meta)
|
||||
const storedEvents = snapshotStoredEvents(events, session.header.id)
|
||||
this.assertEventsSupported(meta, storedEvents)
|
||||
if (!seedCoversPrefix(seed, storedEvents)) {
|
||||
throw new Error(`session "${session.header.id}" already has a persisted log on disk that does not match this live session (id collision)`)
|
||||
}
|
||||
|
||||
@@ -36,7 +36,9 @@ export {
|
||||
DEFAULT_WRITE_BATCH_MAX_DELAY_MS,
|
||||
MAX_WRITE_BATCH_DELAY_MS,
|
||||
PersistenceCoordinator,
|
||||
SessionFormatUnsupportedError,
|
||||
SessionPersistenceCorruptionError,
|
||||
sessionFormatVersionRefusal,
|
||||
} from './coordinator.ts'
|
||||
export type {
|
||||
PersistenceBackend,
|
||||
|
||||
@@ -706,6 +706,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
|
||||
.rejects.toThrow('lacks an identified message')
|
||||
}
|
||||
|
||||
// An out-of-repo event type passes only with the envelope's ignorable
|
||||
// marker (unknown-type refusal otherwise), and its non-object data is
|
||||
// not message-validated.
|
||||
const pluginId = SessionId('non-object-plugin-event')
|
||||
await ctx.sessionPersistence.create(meta(pluginId, WORK))
|
||||
await ctx.sessionPersistence.append(pluginId, [{
|
||||
@@ -713,11 +716,12 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
|
||||
seq: 0,
|
||||
time: 1,
|
||||
data: null,
|
||||
ignorable: true,
|
||||
} as unknown as SessionEvent])
|
||||
await expect(ctx.sessionPersistence.inspect(pluginId))
|
||||
.resolves.toMatchObject({ events: [{ type: 'plugin/test', data: null }] })
|
||||
.resolves.toMatchObject({ events: [{ type: 'plugin/test', data: null, ignorable: true }] })
|
||||
await expect(ctx.sessionPersistence.readFrom(pluginId, 0))
|
||||
.resolves.toMatchObject({ events: [{ type: 'plugin/test', data: null }] })
|
||||
.resolves.toMatchObject({ events: [{ type: 'plugin/test', data: null, ignorable: true }] })
|
||||
|
||||
for (const type of ['user/message', 'assistant/message'] as const) {
|
||||
const missingContentId = SessionId(`invalid-${type}-without-content`)
|
||||
@@ -1321,14 +1325,60 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects an unknown format version on load (assertVersion)', async () => {
|
||||
it('rejects a newer format version on load, naming the upgrade direction', async () => {
|
||||
const fix = await makeFixture()
|
||||
const { ctx, fiber } = await freshCtx(fix)
|
||||
try {
|
||||
const m = { version: 99, id: SessionId('v99'), createdAt: 1, cwd: WORK }
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
await expect(ctx.sessionPersistence.load(m.id)).rejects.toThrow(/version/)
|
||||
const failure = await ctx.sessionPersistence.load(m.id).then(() => undefined, (error: unknown) => error as Error)
|
||||
expect(failure?.name).toBe('SessionFormatUnsupportedError')
|
||||
expect(failure?.message).toMatch(/written by a newer harness.*upgrade the harness/)
|
||||
} finally {
|
||||
await fiber.dispose()
|
||||
await fix.cleanup()
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects an older format version on load without claiming an upgrade path', async () => {
|
||||
const fix = await makeFixture()
|
||||
const { ctx, fiber } = await freshCtx(fix)
|
||||
try {
|
||||
const m = { version: -1, id: SessionId('v-older'), createdAt: 1, cwd: WORK }
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
const failure = await ctx.sessionPersistence.load(m.id).then(() => undefined, (error: unknown) => error as Error)
|
||||
expect(failure?.name).toBe('SessionFormatUnsupportedError')
|
||||
expect(failure?.message).toMatch(/older than the supported v0.*no upgrade path/)
|
||||
} finally {
|
||||
await fiber.dispose()
|
||||
await fix.cleanup()
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects an unknown event type on load unless the event is marked ignorable', async () => {
|
||||
const fix = await makeFixture()
|
||||
const { ctx, fiber } = await freshCtx(fix)
|
||||
try {
|
||||
const required = meta('unknown-required', WORK)
|
||||
await ctx.sessionPersistence.create(required)
|
||||
await ctx.sessionPersistence.append(required.id, [
|
||||
...oneTurnLog(),
|
||||
{ type: 'future/event', seq: oneTurnLog().length, time: 99, data: { payload: 1 } } as unknown as SessionEvent,
|
||||
])
|
||||
const failure = await ctx.sessionPersistence.load(required.id).then(() => undefined, (error: unknown) => error as Error)
|
||||
expect(failure?.name).toBe('SessionFormatUnsupportedError')
|
||||
expect(failure?.message).toMatch(/event type "future\/event".*not marked ignorable/)
|
||||
|
||||
const skippable = meta('unknown-ignorable', WORK)
|
||||
await ctx.sessionPersistence.create(skippable)
|
||||
await ctx.sessionPersistence.append(skippable.id, [
|
||||
...oneTurnLog(),
|
||||
{ type: 'future/event', seq: oneTurnLog().length, time: 99, data: { payload: 1 }, ignorable: true } as unknown as SessionEvent,
|
||||
])
|
||||
const loaded = await ctx.sessionPersistence.load(skippable.id)
|
||||
expect(loaded.events.some(event => (event.type as string) === 'future/event')).toBe(true)
|
||||
} finally {
|
||||
await fiber.dispose()
|
||||
await fix.cleanup()
|
||||
|
||||
Reference in New Issue
Block a user