Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into xtr/agent-loop-message-machine

Browse Source 
# Conflicts:
#	.agents/notes/implemented/feature/2026-06-14-acp-agent-client-protocol.md
#	.agents/notes/implemented/feature/2026-06-14-acp-agent-client-protocol.zh.md
#	.agents/notes/implemented/feature/2026-06-30-hook-bridges.i18n.yaml
#	.agents/notes/implemented/feature/2026-06-30-interception-seams.i18n.yaml
#	.agents/notes/implemented/feature/2026-07-06-sandbox.i18n.yaml
#	.agents/notes/implemented/feature/2026-07-06-sandbox.md
#	.agents/notes/implemented/feature/2026-07-06-sandbox.zh.md
#	.agents/notes/implemented/feature/2026-07-19-model-facing-goal-tools.i18n.yaml
#	.agents/notes/implemented/feature/2026-07-19-same-session-goal-round-driver.i18n.yaml
#	.agents/notes/implemented/feature/2026-07-25-workspace-ui-product-flow.i18n.yaml
#	.agents/notes/implemented/simplification/2026-07-02-remove-stream-chunk-mirror.i18n.yaml
#	docs/architecture.i18n.yaml
#	docs/cookbook/adding-a-tool.i18n.yaml
#	docs/cookbook/extension-cookbook.i18n.yaml
#	docs/core-data-structures/llm-streaming.i18n.yaml
#	docs/core-data-structures/session.i18n.yaml
#	docs/core-data-structures/tools.i18n.yaml
#	docs/event-producer-consumer.md
#	docs/persistence-catalog.md
#	examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/stdout.expected.jsonl
#	examples/acp-agent/tests/snapshots/escalation-approved/session.jsonl
#	examples/acp-agent/tests/snapshots/escalation-rejected/session.jsonl
#	examples/acp-agent/tests/snapshots/fs-escalation-approved/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-cc-pretool-ask/session.jsonl
#	examples/acp-agent/tests/snapshots/permission-switching/session.jsonl
#	examples/acp-agent/tests/snapshots/plan-mode-reject/session.jsonl
#	examples/acp-agent/tests/snapshots/plan-mode/session.jsonl
#	examples/acp-agent/tests/snapshots/session-sandbox-root/session.jsonl
#	packages/context/session-reference/README.md
#	packages/core/agent-loop/tests/agent.spec.ts
#	packages/hooks/hooks-claude/tests/coverage-cases.ts
#	packages/host/runtime/tests/host-runtime.spec.ts
#	packages/llm/llm-retry/tests/retry.spec.ts
#	packages/session-persistence/session-persistence/src/coordinator.ts
#	packages/support/acp-snapshot/README.md
#	packages/support/acp-snapshot/src/normalize.ts
#	packages/ui/acp/acp-feature-support.md
#	packages/ui/acp/src/codec.ts
#	packages/ui/acp/src/index.ts
#	packages/ui/acp/tests/bridge.spec.ts
#	packages/ui/acp/tests/codec.spec.ts
#	packages/ui/acp/tests/config-options.spec.ts
#	packages/ui/acp/tests/dispose.spec.ts
#	packages/ui/acp/tests/edges.spec.ts
#	packages/ui/acp/tests/stream-update.spec.ts
#	packages/ui/acp/tests/turns.spec.ts
This commit is contained in:
_Kerman
2026-07-26 14:05:33 +08:00
parent af3311e1f6 3ccdf4ac46
commit 57d90c343d
1062 changed files with 33621 additions and 30750 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
packages/session-persistence/session-persistence-jsonl/README.md
View File

@@ -6,14 +6,16 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
```
<root>/
cwd-<sha256(cwd)[:12]>/ # per-project bucket (or _no-cwd/ when no cwd)
<encoded-id>.jsonl.zstd # default: checksummed header frame + append frames
<encoded-id>.jsonl # only with compression: 'none'
--<normalized-cwd>--/ # readable project directory (or _no-cwd/)
<encoded-id>/ # session-owned directory
session.jsonl.zstd # default: checksummed header frame + append frames
session.jsonl # only with compression: 'none'
```
- The first logical line is the immutable `SessionHeader` tagged `{ type: 'session', version, id, cwd?, createdAt, parentSession?, seedLength?, delegationDepth }`. `delegationDepth` is required on disk and is `0` for a top-level session; a missing or invalid value rejects the log. Every subsequent logical line is one storage record; `assistant/chunk` events are never dropped, and `seq` stays contiguous across the decoded log (`events[i].seq === i`).
- A storage record is a `SessionEvent` JSON verbatim, or — written only under `packChunks` — a **packed chunk row** (`text-chunks` / `reasoning-chunks` / `tool-call-chunks`; bare slash-less tags like the header's `session`, so row tags cannot be confused with event types): one line holding a run of ≥3 consecutive same-block `assistant/chunk` delta events, `seq0`/`time0` plus per-member `dt` gaps reconstructing every member's `seq`/`time` exactly. The lossless codec lives in `@deepseek-ai/dsh-session` (`packChunkRuns`/`decodeStorageRecord`) and whitelists exact shapes — anything unrecognized stores verbatim. Reading is layout-blind: `load` always decodes rows, so packed, unpacked, and mixed files load identically.
- Session ids are unvalidated branded strings, so they are injectively escaped to a single safe path segment before use (no traversal, no collision).
- The project directory keeps the normalized cwd readable for navigation and is bounded for filesystem component limits. Separator replacement and truncation are intentionally lossy, so cwd strings that normalize alike share a project directory; session ids still select distinct session directories. On a case-insensitive filesystem, identity validation accepts an alternate path spelling only when filesystem canonicalization resolves both spellings to the same transcript. The configured root remains deployment-controlled: it may be project-local, shared, temporary, or centralized. The [project-session directory decision](../../../.agents/notes/implemented/architecture/2026-07-24-project-session-directories.md) records this tradeoff.
- Session ids are unvalidated branded strings, so they are injectively escaped to a single safe path segment before use (no traversal, no collision). The resulting directory is reserved for additional session-owned artifacts; discovery reads only the fixed transcript filename.
## Config
@@ -23,23 +25,23 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
| `packChunks` | `boolean` (default `false`) | Write delta-chunk runs as packed rows (~60% smaller logical logs measured on a real coding session). Off, the written logical layout is byte-identical to the pre-packing format; reading packed rows works regardless of this switch. Off by default while the snapshot goldens stay one-event-per-line — recording with packing on rewrites every fixture `session.jsonl`. |
| `compression` | `'zstd' \| 'none'` | Defaults to `'zstd'`; `'none'` retains newline-delimited UTF-8 text. |
`locate(meta)` returns `{ kind: 'jsonl', path }` using the resolved absolute root and the same cwd-bucket/id encoding as materialization. It performs no filesystem I/O: the target can be returned before the file exists, and an existing file contains only the last flushed prefix.
`locate(meta)` returns `{ kind: 'jsonl', path }` for the fixed transcript inside the resolved project/session directories. It performs no filesystem I/O: the target can be returned before the directory or file exists, and an existing file contains only the last flushed prefix.
## Physical encoding
The default artifact is a standard concatenation of independent [Zstandard frames](../../../.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.md): one checksummed frame containing only the header line, followed by one checksummed frame per durable append batch. The backend uses Node's built-in Zstandard API with its default compression level and exposes no level knob. Listing reads and validates only the header frame. `compression: 'none'` keeps the same logical lines in the original raw representation.
A root belongs to one encoding. Startup discovery and targeted lookup reject the opposite suffix with an error naming the incompatible artifact and instructing the caller to select the matching mode or a separate root. There is no migration, mixed-root fallback, or dual write.
A root belongs to one encoding. Startup discovery and targeted lookup reject the opposite suffix with an error naming the incompatible artifact and instructing the caller to select the matching mode or a separate root. Flat `<project>/<id>.jsonl*` artifacts are also rejected instead of ignored. There is no migration, mixed-root fallback, or dual write.
## Durability and crash semantics
- **Bound storage identity.** Lookup requires one matching encoded filename across the cwd buckets, then verifies that the header id equals the requested id and that the header's id/cwd derive the selected path. Listing applies the same path check and rejects duplicate ids. Identity failures occur before repair or append.
- **Bound storage identity.** Lookup requires one matching session directory across the readable project directories, then verifies that the header id equals the requested id and that the header's id/cwd derive the selected transcript path. Listing applies the same path check and rejects duplicate ids. Identity failures occur before repair or append.
- **Lazy materialization.** `create(meta)` writes nothing; on the first `append`, the backend writes and `fsync`s the encoded header and first batch in a temporary file. POSIX publishes it without overwrite via a hard link and `fsync`s the parent directory. Windows publishes it without overwrite via `MoveFileExW(..., MOVEFILE_WRITE_THROUGH)` and creates missing directories through the same write-through pattern. A created-but-never-appended session leaves nothing on disk and is absent from `list`.
- **Append-only.** Flushed events are never rewritten. Subsequent raw batches append lines; compressed batches append one frame. Both paths `fsync`, and a caught write or sync failure rolls the file back to its prior byte length.
- **Crash recovery — preserve valid tail work.** `load` validates every complete compressed frame and scans their decompressed JSONL. If the last frame is structurally incomplete, the reader keeps its complete decoded records, truncates from that frame's start, and re-encodes those records with the synthetic tool, step, and turn closers required by the shared [persistence contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md). Raw mode truncates from its first incomplete line. A checksum/decompression failure in a complete frame, or a defect at or before the last committed `turn/end`, is corruption and rejects.
- **Non-mutating inspection.** `inspect()` returns the detached valid prefix without truncating an incomplete tail or closing an interrupted turn, and leaves the lightweight revision unchanged.
- **Contiguous-seq.** `append` rejects a batch whose first `seq` does not continue the stored log, and rejects non-JSON-serializable `event.data` naming the offending event type.
- **Lightweight revisions.** `listSnapshots()` identifies a log by its device, inode, size, and nanosecond timestamps, avoiding a full-log parse while changing after append, repair, replacement, or store changes.
- **Lightweight revisions.** `listSnapshots(signal?)` identifies a log by its device, inode, size, and nanosecond timestamps, avoiding a full-log parse while changing after append, repair, replacement, or store changes. It forwards the exact signal through artifact discovery and checks cancellation around every `stat`; because filesystem `stat` is not interruptible, cancellation waits for the active call to settle, then rejects without starting another.
## Write path
@@ -64,6 +66,7 @@ JSONL storage does not mutate live request prefixes. A resumed loop can reuse pr
## Known Limitations and Deferred Work
- **Only the configured encoding and current `SESSION_FORMAT_VERSION` (v0) load** — changing compression requires a separate/fresh root or selecting the legacy raw mode; the pre-release format has no migration.
- **The flat-file storage layout does not load** — use a separate root or move pre-release artifacts into the project/session directory layout before loading.
- **Compressed files are not directly line-readable** — use the backend to load them, or select `compression: 'none'` before writing a fresh root when text fixtures or external line readers are required.
- **Nothing deletes session files** — logs accumulate under `root` until removed externally (the seam has no deletion surface).
- **One live writer per session** — append and repair are coordinated only inside the owning backend instance. Another backend instance or process must not write the same session until that owner reaches quiescent disposal; initial same-id publication remains collision-safe through the POSIX no-overwrite hard link or Windows write-through rename without replacement.

67
packages/session-persistence/session-persistence-jsonl/src/format.ts
View File

@@ -2,13 +2,12 @@
* On-disk format helpers for the JSONL session-persistence backend: path
* sanitization (a {@link SessionId} is an unvalidated branded string, so it
* MUST be encoded before use in a path — no traversal, no collision), the
* per-cwd directory layout, header-line (de)serialization, and the
* per-project/session directory layout, header-line (de)serialization, and the
* truncation-repair offset computation.
*
* @module dsh-session-persistence-jsonl/format
*/
import { createHash } from 'node:crypto'
import { join } from 'node:path'
import { decodeStorageRecord, packChunkRuns } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader, SessionId, StorageRecord } from '@deepseek-ai/dsh-session'
@@ -123,24 +122,64 @@ export function encodeSegment(raw: string): string {
}
/**
* The directory a session's files live in: the configured root, then a per-cwd
* subdirectory so sessions group by project. The cwd subdir is a stable hash of
* the cwd (short, collision-resistant, filesystem-safe); sessions without a
* cwd go in a shared `_no-cwd` bucket.
* @param root - the backend's session root directory.
* @param cwd - the session's project directory; `undefined` selects the shared `_no-cwd` bucket.
* @returns the per-cwd bucket directory path under `root`.
* Build the readable directory key for a project path.
* Filesystem separators and drive separators become `-`; unsafe code units use
* the same `~XXXX` escape as session ids. The key is bounded for filesystem
* component limits. Separator replacement and truncation are intentionally
* lossy, following the common human-navigable project-directory convention.
* @param cwd - the session's project directory.
* @returns a single filesystem-safe project directory name.
*/
export function sessionDir(root: string, cwd: string | undefined): string {
export function projectKey(cwd: string): string {
if (cwd.length === 0) throw new Error('cannot encode an empty project path')
let readable = ''
let separatorRun = false
for (let i = 0; i < cwd.length; i++) {
const code = cwd.charCodeAt(i)
const ch = String.fromCharCode(code)
if (ch === '/' || ch === '\\' || ch === ':') {
if (!separatorRun) readable += '-'
separatorRun = true
} else if (ch !== '~' && /^[A-Za-z0-9._-]$/.test(ch)) {
readable += ch
separatorRun = false
} else {
readable += '~' + code.toString(16).toUpperCase().padStart(4, '0')
separatorRun = false
}
}
const slug = readable.replace(/^-+/, '') || 'root'
return `--${slug.slice(0, 251)}--`
}
/**
* The configured root's human-navigable project directory. A configured root
* may be local or shared; this grouping does not prescribe its deployment.
* @param root - the backend's session root directory.
* @param cwd - the session's project directory; `undefined` selects `_no-cwd`.
* @returns the project directory path under `root`.
*/
export function projectDir(root: string, cwd: string | undefined): string {
if (cwd === undefined) return join(root, '_no-cwd')
const hash = createHash('sha256').update(cwd).digest('hex').slice(0, 12)
return join(root, `cwd-${hash}`)
return join(root, projectKey(cwd))
}
/**
* The directory owned by one session and available for future session-local
* artifacts.
* @param root - the backend's session root directory.
* @param cwd - the session's project directory.
* @param id - the session id, encoded to one safe path segment.
* @returns the session directory beneath its project directory.
*/
export function sessionDir(root: string, cwd: string | undefined, id: SessionId): string {
return join(projectDir(root, cwd), encodeSegment(id))
}
/**
* The append-only event-log file path for a session.
* @param root - the backend's session root directory.
* @param cwd - the session's project directory (picks the per-cwd bucket; `undefined` → `_no-cwd`).
* @param cwd - the session's project directory (`undefined` → `_no-cwd`).
* @param id - the session id, path-encoded via {@link encodeSegment} before filesystem use.
* @param compression - physical artifact encoding and filename suffix.
* @returns the session's configured JSONL artifact path.
@@ -151,7 +190,7 @@ export function logPath(
id: SessionId,
compression: JsonlCompression,
): string {
return join(sessionDir(root, cwd), `${encodeSegment(id)}${logSuffix(compression)}`)
return join(sessionDir(root, cwd, id), `session${logSuffix(compression)}`)
}
/**

256
packages/session-persistence/session-persistence-jsonl/src/index.ts
View File

@@ -9,7 +9,7 @@
import { Context } from 'cordis'
import z from 'schemastery'
import { readdirSync } from 'node:fs'
import { open, mkdir, readFile, readdir, link, rm, stat, truncate } from 'node:fs/promises'
import { open, mkdir, readFile, readdir, realpath, link, rm, stat, truncate } from 'node:fs/promises'
import { dirname, join, resolve } from 'node:path'
import { randomBytes } from 'node:crypto'
import {
@@ -19,7 +19,7 @@ import {
} from '@deepseek-ai/dsh-session-persistence'
import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
import {
encodeSegment, eventLines, logPath, logSuffix, parseHeaderMeta, scanLog, sessionDir, toHeaderLine,
encodeSegment, eventLines, logPath, logSuffix, parseHeaderMeta, projectDir, scanLog, sessionDir, toHeaderLine,
type JsonlCompression,
} from './format.ts'
import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from './zstd.ts'
@@ -40,9 +40,9 @@ export interface Config {
/**
* Root directory for all session files. Required (no default): a default of
* `process.cwd()` would scatter session files as the process's cwd changes
* (bash calls, subprocesses). Sessions group under per-cwd subdirectories. An
* existing root must be a readable directory; an absent root is created on
* first materialization.
* (bash calls, subprocesses). Sessions group under human-readable project
* directories, then per-session directories. An existing root must be a
* readable directory; an absent root is created on first materialization.
*/
root: string
/**
@@ -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
@@ -141,25 +141,34 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
/* jscpd:ignore-end */
// --- 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<StoredPrefix<JsonlTornMarker> | undefined> {
/** Read a stored prefix by id across all project directories when cwd is unknown. */
async loadStored(id: SessionId, signal?: AbortSignal): Promise<StoredPrefix<JsonlTornMarker> | 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<StoredPrefix<JsonlTornMarker>> {
const buffer = await readFile(path)
private async readPrefix(
path: string,
expectedId?: SessionId,
signal?: AbortSignal,
): Promise<StoredPrefix<JsonlTornMarker>> {
const buffer = await readFile(path, { signal })
signal?.throwIfAborted()
let prefix: StoredPrefix<JsonlTornMarker>
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,46 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
: {},
}
}
this.assertStoredIdentity(path, prefix.meta, expectedId)
signal?.throwIfAborted()
await this.assertStoredIdentity(path, prefix.meta, expectedId, signal)
signal?.throwIfAborted()
return prefix
}
/** Decode complete frames and retain complete JSONL records from a torn final frame. */
private async readZstdPrefix(buffer: Buffer): Promise<StoredPrefix<JsonlTornMarker>> {
private async readZstdPrefix(
buffer: Buffer,
signal?: AbortSignal,
): Promise<StoredPrefix<JsonlTornMarker>> {
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 +226,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,16 +277,18 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
/** List valid unique stored sessions' metadata (header line only — no full-log parse). */
async list(): Promise<SessionHeader[]> {
return (await this.listArtifacts()).map(artifact => artifact.header)
async list(signal?: AbortSignal): Promise<SessionHeader[]> {
return (await this.listArtifacts(signal)).map(artifact => artifact.header)
}
/** List metadata plus a stat-derived identity for each append-only log. */
async listSnapshots(): Promise<SessionPersistenceSnapshot[]> {
async listSnapshots(signal?: AbortSignal): Promise<SessionPersistenceSnapshot[]> {
const snapshots: SessionPersistenceSnapshot[] = []
for (const artifact of await this.listArtifacts()) {
for (const artifact of await this.listArtifacts(signal)) {
signal?.throwIfAborted()
try {
const identity = await stat(artifact.path, { bigint: true })
signal?.throwIfAborted()
snapshots.push({
header: artifact.header,
revision: SessionPersistenceRevision([
@@ -268,34 +300,50 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
].join(':')),
})
} catch (error: unknown) {
signal?.throwIfAborted()
if (!isENOENT(error)) throw error
}
}
signal?.throwIfAborted()
return snapshots
}
private async listArtifacts(): Promise<Array<{ header: SessionHeader; path: string }>> {
private async listArtifacts(signal?: AbortSignal): Promise<Array<{ header: SessionHeader; path: string }>> {
signal?.throwIfAborted()
await this.ensureRootEncoding()
signal?.throwIfAborted()
const artifacts: Array<{ header: SessionHeader; path: string }> = []
const ids = new Set<SessionId>()
for (const dir of await this.listCwdDirs()) {
for (const name of await this.listArtifactNames(dir)) {
const path = join(dir, name)
for (const project of await this.listProjectDirs(signal)) {
signal?.throwIfAborted()
for (const dir of await this.listSessionDirs(project, signal)) {
signal?.throwIfAborted()
const opposite = join(dir, `session${logSuffix(this.oppositeCompression())}`)
const oppositeExists = await this.exists(opposite)
signal?.throwIfAborted()
if (oppositeExists) throw this.encodingMismatch(opposite)
const path = join(dir, `session${logSuffix(this.compression)}`)
const pathExists = await this.exists(path)
signal?.throwIfAborted()
if (!pathExists) continue
// 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
this.assertStoredIdentity(path, meta)
await this.assertStoredIdentity(path, meta, undefined, signal)
signal?.throwIfAborted()
if (ids.has(meta.id)) {
throw new Error(`duplicate JSONL session id "${meta.id}" appears in multiple cwd buckets`)
throw new Error(`duplicate JSONL session id "${meta.id}" appears in multiple project directories`)
}
ids.add(meta.id)
artifacts.push({ header: meta, path })
}
}
signal?.throwIfAborted()
return artifacts
}
@@ -303,20 +351,22 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
/** Atomically write the header line + first batch (temp-write, fsync, publish). */
private async materialize(meta: SessionHeader, events: readonly SessionEvent[]): Promise<void> {
const dir = sessionDir(this.root, meta.cwd)
const project = projectDir(this.root, meta.cwd)
const dir = sessionDir(this.root, meta.cwd, meta.id)
const finalPath = logPath(this.root, meta.cwd, meta.id, this.compression)
await this.rejectOppositeArtifact(meta.cwd, meta.id)
const content = await this.encodeMaterialization(meta, events)
/* v8 ignore next -- native Windows coverage exercises this platform dispatch; Linux covers the POSIX peer */
if (process.platform === 'win32') {
await this.materializeWin32(dir, finalPath, meta.id, content)
await this.materializeWin32(project, dir, finalPath, meta.id, content)
} else {
await this.materializePosix(dir, finalPath, meta.id, content)
await this.materializePosix(project, dir, finalPath, meta.id, content)
}
}
/* v8 ignore start -- Windows uses the Win32 durable-publish path; POSIX coverage exercises this peer. */
private async materializePosix(
project: string,
dir: string,
finalPath: string,
id: SessionId,
@@ -324,8 +374,10 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
): Promise<void> {
await mkdir(this.root, { recursive: true, mode: 0o700 })
await this.syncDirPosix(dirname(this.root))
await mkdir(dir, { recursive: true, mode: 0o700 })
await mkdir(project, { recursive: true, mode: 0o700 })
await this.syncDirPosix(this.root)
await mkdir(dir, { recursive: true, mode: 0o700 })
await this.syncDirPosix(project)
await this.rejectExistingLog(finalPath, id)
const tmp = await this.writeSyncedTempFile(finalPath, content)
// Publish via link()+unlink(), NOT rename(): link fails with EEXIST if the
@@ -358,12 +410,14 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
/* v8 ignore start -- native Windows coverage exercises this integration path */
private async materializeWin32(
project: string,
dir: string,
finalPath: string,
id: SessionId,
content: Buffer | string,
): Promise<void> {
await ensureDurableDirectoryWin32(this.root)
await ensureDurableDirectoryWin32(project)
await ensureDurableDirectoryWin32(dir)
await this.rejectExistingLog(finalPath, id)
const tmp = await this.writeSyncedTempFile(finalPath, content)
@@ -492,18 +546,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<string | undefined> {
private async readFirstLine(path: string, signal?: AbortSignal): Promise<string | undefined> {
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 +573,34 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
/** Read and validate only the independently compressed header frame. */
private async readFirstZstdLine(path: string): Promise<string | undefined> {
private async readFirstZstdLine(path: string, signal?: AbortSignal): Promise<string | undefined> {
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')
}
@@ -541,20 +611,27 @@ 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<string | undefined> {
const target = encodeSegment(id) + logSuffix(this.compression)
const oppositeTarget = encodeSegment(id) + logSuffix(this.oppositeCompression())
/** Find the unique physical log for an id across every project directory. */
private async findLog(id: SessionId, signal?: AbortSignal): Promise<string | undefined> {
const matches: string[] = []
for (const dir of await this.listCwdDirs()) {
const path = join(dir, target)
const opposite = join(dir, oppositeTarget)
if (await this.exists(opposite)) throw this.encodingMismatch(opposite)
if (await this.exists(path)) matches.push(path)
for (const project of await this.listProjectDirs(signal)) {
signal?.throwIfAborted()
await this.rejectLegacyFlatArtifact(project, id, signal)
signal?.throwIfAborted()
const dir = join(project, encodeSegment(id))
const path = join(dir, `session${logSuffix(this.compression)}`)
const opposite = join(dir, `session${logSuffix(this.oppositeCompression())}`)
const oppositeExists = await this.exists(opposite)
signal?.throwIfAborted()
if (oppositeExists) throw this.encodingMismatch(opposite)
const pathExists = await this.exists(path)
signal?.throwIfAborted()
if (pathExists) matches.push(path)
}
if (matches.length > 1) {
throw new Error(`duplicate JSONL session id "${id}" appears in multiple cwd buckets`)
throw new Error(`duplicate JSONL session id "${id}" appears in multiple project directories`)
}
signal?.throwIfAborted()
return matches[0]
}
@@ -569,7 +646,13 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
/** Reject metadata that does not identify the selected physical log. */
private assertStoredIdentity(path: string, meta: SessionHeader, expectedId?: SessionId): void {
private async assertStoredIdentity(
path: string,
meta: SessionHeader,
expectedId?: SessionId,
signal?: AbortSignal,
): Promise<void> {
signal?.throwIfAborted()
if (expectedId !== undefined && meta.id !== expectedId) {
throw new Error(`corrupt session log "${path}": requested id "${expectedId}" does not match header id "${meta.id}"`)
}
@@ -579,15 +662,38 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
} catch (error) {
throw new Error(`corrupt session log "${path}": header id cannot name a storage path`, { cause: error })
}
if (path !== expectedPath) {
throw new Error(`corrupt session log "${path}": header id "${meta.id}" and cwd belong at "${expectedPath}"`)
if (path !== expectedPath && !await this.sameFile(path, expectedPath, signal)) {
throw new Error(`corrupt session log "${path}": header id "${meta.id}" and cwd identify "${expectedPath}"`)
}
signal?.throwIfAborted()
}
/**
* Whether two path spellings resolve to the same physical file. This admits
* case aliases on case-insensitive filesystems without weakening identity
* checks on case-sensitive stores.
*/
private async sameFile(path: string, expectedPath: string, signal?: AbortSignal): Promise<boolean> {
signal?.throwIfAborted()
try {
const [actual, expected] = await Promise.all([realpath(path), realpath(expectedPath)])
signal?.throwIfAborted()
return actual === expected
} catch (error) {
signal?.throwIfAborted()
/* v8 ignore else -- non-ENOENT realpath failures require an external permission or I/O fault */
if (isENOENT(error)) return false
/* v8 ignore next -- non-ENOENT realpath failures are external I/O faults, propagated unchanged */
throw error
}
}
/** The cwd-bucket directories under the root (absolute paths). */
private async listCwdDirs(): Promise<string[]> {
/** The human-readable project directories under the configured root. */
private async listProjectDirs(signal?: AbortSignal): Promise<string[]> {
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,13 +702,15 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
}
private async listArtifactNames(dir: string): Promise<string[]> {
const entries = await readdir(dir)
const oppositeSuffix = logSuffix(this.oppositeCompression())
const incompatible = entries.find(name => name.endsWith(oppositeSuffix))
if (incompatible !== undefined) throw this.encodingMismatch(`${dir}/${incompatible}`)
const suffix = logSuffix(this.compression)
return entries.filter(name => name.endsWith(suffix))
/** List session-owned directories and reject the obsolete flat-file layout. */
private async listSessionDirs(project: string, signal?: AbortSignal): Promise<string[]> {
signal?.throwIfAborted()
const entries = await readdir(project, { withFileTypes: true })
signal?.throwIfAborted()
const legacy = entries.find(entry =>
entry.isFile() && (entry.name.endsWith('.jsonl') || entry.name.endsWith('.jsonl.zstd')))
if (legacy !== undefined) throw this.legacyLayout(join(project, legacy.name))
return entries.filter(entry => entry.isDirectory()).map(entry => join(project, entry.name))
}
/** Reject a root that already belongs to the other physical encoding. */
@@ -612,11 +720,26 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
private async checkRootEncoding(): Promise<void> {
const oppositeSuffix = logSuffix(this.oppositeCompression())
for (const dir of await this.listCwdDirs()) {
const entries = await readdir(dir)
const incompatible = entries.find(name => name.endsWith(oppositeSuffix))
if (incompatible !== undefined) throw this.encodingMismatch(`${dir}/${incompatible}`)
for (const project of await this.listProjectDirs()) {
for (const dir of await this.listSessionDirs(project)) {
const incompatible = join(dir, `session${logSuffix(this.oppositeCompression())}`)
if (await this.exists(incompatible)) throw this.encodingMismatch(incompatible)
}
}
}
private async rejectLegacyFlatArtifact(
project: string,
id: SessionId,
signal?: AbortSignal,
): Promise<void> {
signal?.throwIfAborted()
const encoded = encodeSegment(id)
for (const compression of ['zstd', 'none'] as const) {
const path = join(project, encoded + logSuffix(compression))
const artifactExists = await this.exists(path)
signal?.throwIfAborted()
if (artifactExists) throw this.legacyLayout(path)
}
}
@@ -637,6 +760,13 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
)
}
private legacyLayout(path: string): Error {
return new Error(
`session artifact ${JSON.stringify(path)} uses the unsupported flat-file layout; `
+ 'use a separate root or move it into a project/session directory before loading',
)
}
private async exists(path: string): Promise<boolean> {
try {
const handle = await open(path, 'r')
@@ -646,7 +776,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
// Only ENOENT means absent. A permission/I/O error must surface rather
// than letting load or collision checks proceed under false absence.
// Windows reports ENOENT, not ENOTDIR, for `regular-file/child`; verify
// the immediate parent so a blocked cwd bucket remains a storage fault.
// the immediate parent so a blocked session directory remains a storage fault.
/* v8 ignore else -- Windows reports file-valued parents as ENOENT; POSIX covers direct ENOTDIR. */
if (isENOENT(error)) {
await this.assertLogParentAllowsAbsence(path)

6
packages/session-persistence/session-persistence-jsonl/src/win32.ts
View File

@@ -12,7 +12,7 @@
*/
import { mkdtemp, rm, stat } from 'node:fs/promises'
import { basename, join, parse, resolve, toNamespacedPath } from 'node:path'
import { join, parse, resolve, toNamespacedPath } from 'node:path'
type MoveFileExW = (existing: string, replacement: string, flags: number) => number
type GetLastError = () => number
@@ -139,7 +139,9 @@ export async function ensureDurableDirectoryWin32(target: string): Promise<void>
}
async function createLeafDirectoryWin32(parent: string, target: string): Promise<void> {
const staging = await mkdtemp(join(parent, `.dsh-mkdir-${basename(target)}-`))
// Keep the staging component independent of the target basename so a legal
// 255-byte target component does not make mkdtemp's sibling name too long.
const staging = await mkdtemp(join(parent, '.dsh-mkdir-'))
try {
await publishNewFileWin32(staging, target)
} catch (error) {

212
packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts
View File

@@ -1,12 +1,14 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat } from 'node:fs/promises'
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 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'
import { encodeSegment, eventLines, logPath, scanLog, sessionDir, toHeaderLine } from '../src/format.ts'
import {
encodeSegment, eventLines, logPath, projectDir, projectKey, scanLog, sessionDir, toHeaderLine,
} from '../src/format.ts'
import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../session-persistence/tests/contract.ts'
import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
@@ -125,6 +127,16 @@ describe('SessionPersistenceJsonl: format helpers', () => {
expect(() => encodeSegment('')).toThrow(/empty/)
})
it('projectKey normalizes project paths into bounded readable names', () => {
expect(projectKey('/Users/qyj/work/deepseek-harness')).toBe('--Users-qyj-work-deepseek-harness--')
expect(projectKey('/a/b-c')).toBe(projectKey('/a-b/c'))
expect(projectKey('C:\\work\\agent')).toBe('--C-work-agent--')
expect(projectKey('/开发/~agent')).toBe('--~5F00~53D1-~007Eagent--')
expect(projectKey('/')).toBe('--root--')
expect(projectKey('/' + 'x'.repeat(1_000))).toHaveLength(255)
expect(() => projectKey('')).toThrow(/empty project path/)
})
it('resolves a relative custom root before locating a session', async () => {
const absoluteRoot = await freshRoot()
const ctx = new Context()
@@ -161,15 +173,15 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
await ctx.sessionPersistence.create(m)
// locate() is a pure target-path calculation: neither it nor create()
// materializes a file before the first append.
const dir = sessionDir(root, '/work')
const dir = sessionDir(root, '/work', m.id)
await expect(stat(rawLogPath(root, '/work', m.id))).rejects.toThrow()
expect((await ctx.sessionPersistence.list()).map(h => h.id)).not.toContain(m.id)
await ctx.sessionPersistence.append(m.id, oneTurnLog())
// now materialized
expect((await stat(dir)).isDirectory()).toBe(true)
expect((await stat(rawLogPath(root, '/work', m.id))).isFile()).toBe(true)
expect((await ctx.sessionPersistence.list()).map(h => h.id)).toContain(m.id)
void dir
})
it('keeps the same location on resume and gives a fork its own location', async () => {
@@ -251,24 +263,72 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
})
it('surfaces non-ENOENT snapshot stat failures after discovery', async () => {
const blocker = join(root, 'snapshot-not-a-directory')
await writeFile(blocker, 'x')
const persistence = ctx.sessionPersistence as unknown as {
listArtifacts(): Promise<Array<{ header: SessionHeader; path: string }>>
}
const discovery = vi.spyOn(persistence, 'listArtifacts').mockResolvedValue([{
header: meta('snapshot-stat-failure'),
path: join(blocker, 'session.jsonl'),
path: `${root}\0snapshot-stat-failure`,
}])
await expect(ctx.sessionPersistence.listSnapshots()).rejects.toThrow(/ENOTDIR/)
await expect(ctx.sessionPersistence.listSnapshots()).rejects.toThrow(/null bytes/)
discovery.mockRestore()
})
it('forwards snapshot-list cancellation and awaits in-flight discovery cleanup', async () => {
const persistence = ctx.sessionPersistence as unknown as {
listArtifacts(signal?: AbortSignal): Promise<Array<{ header: SessionHeader; path: string }>>
}
const started = Promise.withResolvers<AbortSignal>()
const cleanup = Promise.withResolvers<undefined>()
vi.spyOn(persistence, 'listArtifacts').mockImplementation(async (signal) => {
if (signal === undefined) throw new Error('expected snapshot-list signal')
started.resolve(signal)
await cleanup.promise
return []
})
const reason = new Error('JSONL snapshot discovery cancelled')
const controller = new AbortController()
const pending = ctx.sessionPersistence.listSnapshots(controller.signal)
expect(await started.promise).toBe(controller.signal)
let settled = false
void pending.then(
() => { settled = true },
() => { settled = true },
)
controller.abort(reason)
await Promise.resolve()
expect(settled).toBe(false)
cleanup.resolve(undefined)
await expect(pending).rejects.toBe(reason)
})
it('checks cancellation after an uncancellable snapshot stat settles', async () => {
const m = meta('snapshot-stat-cancellation')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, oneTurnLog())
const persistence = ctx.sessionPersistence as unknown as {
listArtifacts(signal?: AbortSignal): Promise<Array<{ header: SessionHeader; path: string }>>
}
const discovery = vi.spyOn(persistence, 'listArtifacts').mockResolvedValue([{
header: m,
path: rawLogPath(root, m.cwd, m.id),
}])
const reason = new Error('JSONL snapshot stat cancelled')
const controller = new AbortController()
const pending = ctx.sessionPersistence.listSnapshots(controller.signal)
queueMicrotask(() => { controller.abort(reason) })
await expect(pending).rejects.toBe(reason)
expect(discovery).toHaveBeenCalledWith(controller.signal)
})
it('rejects a stored v0 log containing a legacy request/header-delta event', async () => {
const m = meta('legacy-header-delta', '/legacy')
const path = rawLogPath(root, m.cwd, m.id)
await mkdir(sessionDir(root, m.cwd), { recursive: true })
await mkdir(sessionDir(root, m.cwd, m.id), { recursive: true })
await writeFile(path, [
JSON.stringify(toHeaderLine(m)),
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
@@ -283,7 +343,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
it('rejects a stored v0 full header carrying the legacy fallback reason', async () => {
const m = meta('legacy-header-fallback', '/legacy')
const path = rawLogPath(root, m.cwd, m.id)
await mkdir(sessionDir(root, m.cwd), { recursive: true })
await mkdir(sessionDir(root, m.cwd, m.id), { recursive: true })
await writeFile(path, [
JSON.stringify(toHeaderLine(m)),
JSON.stringify({
@@ -713,7 +773,7 @@ describe('SessionPersistenceJsonl: packed chunk rows (packChunks: true)', () =>
const log = chunkRunLog()
// First turn written line-per-event by an unpacked-config writer (an old
// file, hand-planted so this packed-config backend adopts it on load).
await mkdir(sessionDir(root, '/work'), { recursive: true })
await mkdir(sessionDir(root, '/work', m.id), { recursive: true })
await writeFile(rawLogPath(root, '/work', m.id), [
JSON.stringify({ type: 'session', version: 0, id: 'mixed', createdAt: 1000, cwd: '/work', delegationDepth: 0 }),
...log.map(e => JSON.stringify(e)),
@@ -809,34 +869,93 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await expect(stat(rawLogPath(root, '/mutated', SessionId('create-snap')))).rejects.toThrow()
})
it('list discovers sessions across multiple cwd buckets', async () => {
it('list discovers sessions across multiple project directories', async () => {
await ctx.sessionPersistence.create(meta('p1', '/projA'))
await ctx.sessionPersistence.append(SessionId('p1'), oneTurnLog())
await ctx.sessionPersistence.create(meta('p2', '/projB'))
await ctx.sessionPersistence.append(SessionId('p2'), oneTurnLog())
await ctx.sessionPersistence.create(meta('p3')) // no cwd → _no-cwd bucket
await ctx.sessionPersistence.create(meta('p3')) // no cwd → _no-cwd project directory
await ctx.sessionPersistence.append(SessionId('p3'), oneTurnLog())
const ids = (await ctx.sessionPersistence.list()).map(x => x.id).sort()
expect(ids).toEqual(['p1', 'p2', 'p3'])
})
it('groups sessions whose cwd paths normalize to the same project directory', async () => {
const first = meta('normalized-first', '/a/b-c')
const second = meta('normalized-second', '/a-b/c')
await ctx.sessionPersistence.create(first)
await ctx.sessionPersistence.append(first.id, oneTurnLog())
await ctx.sessionPersistence.create(second)
await ctx.sessionPersistence.append(second.id, oneTurnLog())
expect(projectDir(root, first.cwd)).toBe(projectDir(root, second.cwd))
expect(await readdir(projectDir(root, first.cwd))).toEqual(expect.arrayContaining([
encodeSegment(first.id),
encodeSegment(second.id),
]))
expect((await ctx.sessionPersistence.list()).map(header => header.id).sort())
.toEqual([first.id, second.id].sort())
})
it('list on an empty root returns nothing', async () => {
expect(await ctx.sessionPersistence.list()).toEqual([])
})
it('list skips empty and non-header .jsonl files (metadata-only read)', async () => {
it('keeps the transcript in an extensible session-owned directory', async () => {
const m = meta('owned-directory', '/project')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, oneTurnLog())
const dir = sessionDir(root, m.cwd, m.id)
await writeFile(join(dir, 'metadata.json'), '{}\n')
await writeFile(join(projectDir(root, m.cwd), 'README'), 'project metadata\n')
await mkdir(join(projectDir(root, m.cwd), 'reserved-session'), { recursive: true })
expect(await readdir(dir)).toEqual(expect.arrayContaining(['metadata.json', 'session.jsonl']))
expect((await ctx.sessionPersistence.list()).map(header => header.id)).toContain(m.id)
expect((await ctx.sessionPersistence.load(m.id)).events).toEqual(oneTurnLog())
})
it('rejects the obsolete flat-file layout instead of ignoring stored sessions', async () => {
const m = meta('legacy-flat', '/legacy')
const project = projectDir(root, m.cwd)
const path = join(project, `${encodeSegment(m.id)}.jsonl`)
await mkdir(project, { recursive: true })
await writeFile(path, [
JSON.stringify(toHeaderLine(m)),
...oneTurnLog().map(event => JSON.stringify(event)),
'',
].join('\n'))
await expect(ctx.sessionPersistence.load(m.id)).rejects.toThrow(/unsupported flat-file layout/)
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/unsupported flat-file layout/)
})
it('rejects a compressed obsolete flat-file artifact during targeted lookup', async () => {
const m = meta('legacy-compressed-flat', '/legacy')
const project = projectDir(root, m.cwd)
expect(await ctx.sessionPersistence.list()).toEqual([])
await mkdir(project, { recursive: true })
await writeFile(join(project, `${encodeSegment(m.id)}.jsonl.zstd`), 'legacy')
await expect(ctx.sessionPersistence.load(m.id)).rejects.toThrow(/unsupported flat-file layout/)
})
it('list skips empty and non-header session logs (metadata-only read)', async () => {
// A real session…
await ctx.sessionPersistence.create(meta('real', '/p'))
await ctx.sessionPersistence.append(SessionId('real'), oneTurnLog())
// …alongside two junk files in the _no-cwd bucket: an EMPTY file (readFirstLine
// returns undefined) and a file whose first line is not a session header
// (parseHeaderMeta returns undefined). Both are skipped, not listed.
const bucket = join(root, '_no-cwd')
await mkdir(bucket, { recursive: true })
await writeFile(join(bucket, 'empty.jsonl'), '')
await writeFile(join(bucket, 'notheader.jsonl'), '{"type":"turn/start"}\n')
await writeFile(join(bucket, 'badjson.jsonl'), 'not json at all\n')
// …alongside junk session directories whose fixed transcript is empty or
// lacks a header. Both remain unmaterialized and are skipped.
for (const [id, content] of [
['empty', ''],
['notheader', '{"type":"turn/start"}\n'],
['badjson', 'not json at all\n'],
] as const) {
const path = rawLogPath(root, undefined, SessionId(id))
await mkdir(sessionDir(root, undefined, SessionId(id)), { recursive: true })
await writeFile(path, content)
}
const ids = (await ctx.sessionPersistence.list()).map(x => x.id).sort()
expect(ids).toEqual(['real'])
@@ -845,10 +964,10 @@ describe('SessionPersistenceJsonl: edge cases', () => {
it('list reads a header line longer than the 8KB read chunk', async () => {
// A tolerated extra field makes this valid header exceed the 8192-byte read buffer, proving
// `readFirstLine` accumulates chunks before `list()` parses it.
const bucket = join(root, '_no-cwd')
await mkdir(bucket, { recursive: true })
const id = SessionId('big')
await mkdir(sessionDir(root, undefined, id), { recursive: true })
const bigHeader = JSON.stringify({ type: 'session', version: 0, id: 'big', createdAt: 1, delegationDepth: 0, pad: 'x'.repeat(9000) })
await writeFile(join(bucket, 'big.jsonl'), bigHeader + '\n')
await writeFile(rawLogPath(root, undefined, id), bigHeader + '\n')
const ids = (await ctx.sessionPersistence.list()).map(x => x.id)
expect(ids).toContain('big')
})
@@ -859,30 +978,47 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await ctx.sessionPersistence.append(m.id, oneTurnLog())
await rewriteHeader(rawLogPath(root, m.cwd, m.id), (header) => { header.cwd = '/elsewhere' })
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/and cwd belong at/)
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/and cwd identify/)
})
it('accepts an alternate project path only when it identifies the same physical log', async () => {
const m = meta('physical-alias', '/stored')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, oneTurnLog())
const path = rawLogPath(root, m.cwd, m.id)
const aliasCwd = '/alias'
await symlink(
projectDir(root, m.cwd),
projectDir(root, aliasCwd),
process.platform === 'win32' ? 'junction' : 'dir',
)
await rewriteHeader(path, (header) => { header.cwd = aliasCwd })
expect((await ctx.sessionPersistence.load(m.id)).meta.cwd).toBe(aliasCwd)
expect((await ctx.sessionPersistence.list()).map(header => header.id)).toContain(m.id)
})
it('list rejects a session header whose id cannot name a storage path', async () => {
const bucket = sessionDir(root, undefined)
await mkdir(bucket, { recursive: true })
await writeFile(join(bucket, 'invalid-id.jsonl'), JSON.stringify({
const dir = join(projectDir(root, undefined), 'invalid-id')
await mkdir(dir, { recursive: true })
await writeFile(join(dir, 'session.jsonl'), JSON.stringify({
type: 'session', version: 0, id: '', createdAt: 1, delegationDepth: 0,
}) + '\n')
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/header id cannot name a storage path/)
})
it('load and list reject one id materialized in multiple cwd buckets', async () => {
it('load and list reject one id materialized in multiple project directories', async () => {
const id = SessionId('duplicate')
for (const cwd of ['/a', '/b']) {
const m = meta(id, cwd)
await mkdir(sessionDir(root, cwd), { recursive: true })
await mkdir(sessionDir(root, cwd, id), { recursive: true })
const content = [JSON.stringify(toHeaderLine(m)), ...oneTurnLog().map(event => JSON.stringify(event))].join('\n') + '\n'
await writeFile(rawLogPath(root, cwd, id), content)
}
await expect(ctx.sessionPersistence.load(id)).rejects.toThrow(/appears in multiple cwd buckets/)
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/appears in multiple cwd buckets/)
await expect(ctx.sessionPersistence.load(id)).rejects.toThrow(/appears in multiple project directories/)
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/appears in multiple project directories/)
})
it('a DIFFERENT live session object reusing a disposed id gets its own init (no stale cache)', async () => {
@@ -1005,12 +1141,12 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await expect(backend.exists(join(blocker, 'child.jsonl'))).rejects.toThrow(/ENOTDIR/)
})
it('materialization surfaces a cwd-bucket storage fault', async () => {
it('materialization surfaces a project-directory storage fault', async () => {
const cwd = '/x'
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
await writeFile(sessionDir(root, cwd), 'x') // bucket path is now a FILE
await writeFile(projectDir(root, cwd), 'x') // project path is now a file
let s!: Session
await ctx2.plugin(Object.assign((inner: Context) => {
s = inner.sessions.create(SessionId('exists-fault'), { meta: { cwd } })
@@ -1058,14 +1194,14 @@ describe('SessionPersistenceJsonl: edge cases', () => {
})
it('createCore rejects an id already on disk under a DIFFERENT cwd bucket', async () => {
it('createCore rejects an id already on disk under a different project directory', async () => {
// Persist the id under cwd A.
const a = meta('dup-id', '/projA')
await ctx.sessionPersistence.create(a)
await ctx.sessionPersistence.append(a.id, oneTurnLog())
// A fresh backend creating the SAME id under cwd B must still refuse: load
// identifies by id across all buckets, so a second log would make resume
// nondeterministic. create scans every bucket, not just meta.cwd's.
// identifies by id across all projects, so a second log would make resume
// nondeterministic. create scans every project, not just meta.cwd's.
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })

9
packages/session-persistence/session-persistence-jsonl/tests/win32.spec.ts
View File

@@ -151,6 +151,15 @@ describe('Windows durable namespace helpers', () => {
expect(existsSync(raced)).toBe(true)
})
it('keeps staging names valid for a maximum-length target component', async () => {
const { ensureDurableDirectoryWin32 } = await importWithFilesystemMove()
const root = await tempRoot()
const target = join(root, 'x'.repeat(255))
await ensureDurableDirectoryWin32(target)
expect(existsSync(target)).toBe(true)
})
it('surfaces directory publication failures other than an existing-target race', async () => {
const { ensureDurableDirectoryWin32 } = await importWithError(ERROR_ACCESS_DENIED)
const root = await tempRoot()

100
packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts
View File

@@ -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<unknown>
}
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<string> {
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)
@@ -391,15 +462,21 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
it('skips empty, incomplete, and non-header compressed artifacts while rejecting malformed header frames', async () => {
const root = await freshRoot()
const bucket = sessionDir(root, undefined)
await mkdir(bucket, { recursive: true })
await writeFile(join(bucket, 'empty.jsonl.zstd'), '')
await writeFile(join(bucket, 'partial.jsonl.zstd'), MAGIC)
await writeFile(join(bucket, 'not-header.jsonl.zstd'), await compressZstdFrame('{"type":"turn/start"}\n'))
for (const [id, content] of [
['empty', Buffer.alloc(0)],
['partial', MAGIC],
['not-header', await compressZstdFrame('{"type":"turn/start"}\n')],
] as const) {
const sessionId = SessionId(id)
await mkdir(sessionDir(root, undefined, sessionId), { recursive: true })
await writeFile(logPath(root, undefined, sessionId, 'zstd'), content)
}
const ctx = await mount(root)
expect(await ctx.sessionPersistence.list()).toEqual([])
await writeFile(join(bucket, 'two-lines.jsonl.zstd'), await compressZstdFrame([
const twoLinesId = SessionId('two-lines')
await mkdir(sessionDir(root, undefined, twoLinesId), { recursive: true })
await writeFile(logPath(root, undefined, twoLinesId, 'zstd'), await compressZstdFrame([
JSON.stringify(toHeaderLine(meta('two-lines'))),
JSON.stringify({ type: 'turn/start' }),
'',
@@ -411,8 +488,9 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
it('rejects missing, empty, and checksum-corrupt header frames on targeted reads', async () => {
const root = await freshRoot()
const bucket = sessionDir(root, undefined)
await mkdir(bucket, { recursive: true })
for (const id of ['partial-only', 'empty-header', 'bad-checksum']) {
await mkdir(sessionDir(root, undefined, SessionId(id)), { recursive: true })
}
await writeFile(logPath(root, undefined, SessionId('partial-only'), 'zstd'), MAGIC)
await writeFile(logPath(root, undefined, SessionId('empty-header'), 'zstd'), await compressZstdFrame(''))
const corruptHeader = Buffer.from(await compressZstdFrame(`${JSON.stringify(toHeaderLine(meta('bad-checksum')))}\n`))
@@ -453,7 +531,7 @@ describe('SessionPersistenceJsonl: encoding selection', () => {
expect(await ctx.sessionPersistence.list()).toEqual([])
const loadHeader = meta('late-raw-load', '/late')
await mkdir(sessionDir(root, loadHeader.cwd), { recursive: true })
await mkdir(sessionDir(root, loadHeader.cwd, loadHeader.id), { recursive: true })
await writeFile(logPath(root, loadHeader.cwd, loadHeader.id, 'none'), [
JSON.stringify(toHeaderLine(loadHeader)),
...oneTurnLog().map(e => JSON.stringify(e)),
@@ -471,13 +549,13 @@ describe('SessionPersistenceJsonl: encoding selection', () => {
await ctx.sessionPersistence.list()
const header = meta('late-raw-materialize', '/late')
await ctx.sessionPersistence.create(header)
await mkdir(sessionDir(root, header.cwd), { recursive: true })
await mkdir(sessionDir(root, header.cwd, header.id), { recursive: true })
await writeFile(logPath(root, header.cwd, header.id, 'none'), [
JSON.stringify(toHeaderLine(header)),
...oneTurnLog().map(e => JSON.stringify(e)),
'',
].join('\n'))
await expect(ctx.sessionPersistence.append(header.id, oneTurnLog())).rejects.toThrow(/uses \.jsonl/)
expect((await readdir(sessionDir(root, header.cwd))).some(name => name.endsWith('.jsonl.zstd'))).toBe(false)
expect((await readdir(sessionDir(root, header.cwd, header.id))).some(name => name.endsWith('.jsonl.zstd'))).toBe(false)
})
})

2
packages/session-persistence/session-persistence-sqlite/README.md
View File

@@ -20,7 +20,7 @@ On filesystems with POSIX modes, the backend requests mode `0700` for missing di
- **Lazy materialization.** `create()` records intent in memory only — no row is written until the first `append`. A created-but-never-appended session has no `sessions` row, so it is absent from `list()` (which reports exactly the sessions that have a row).
- **Interrupted-turn close on load.** `load()` implements the shared [crash-recovery contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md): preserve the valid interrupted turn, append its synthetic closing events in one transaction, and remove only a torn tail row. Committed parse errors or sequence gaps make the session unloadable. Because recovery mutates stored rows, the next append starts from a balanced log and accurate cursor.
- **Non-mutating inspection.** `inspect()` returns the detached valid row prefix without deleting a torn tail row or appending recovery closers, and leaves the lightweight revision unchanged.
- **Lightweight revisions.** `listSnapshots()` combines the immutable store and database-file identity, a per-materialization incarnation id, and a per-session counter incremented in each mutating transaction. This keeps unchanged observations stable without parsing event rows and distinguishes independent stores and recreated same-id logs.
- **Lightweight revisions.** `listSnapshots(signal?)` combines the immutable store and database-file identity, a per-materialization incarnation id, and a per-session counter incremented in each mutating transaction. This keeps unchanged observations stable without parsing event rows and distinguishes independent stores and recreated same-id logs. It checks cancellation before and after shared readiness and the synchronous metadata query; the query itself is non-preemptible.
## Configuration (schemastery)

23
packages/session-persistence/session-persistence-sqlite/src/index.ts
View File

@@ -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<StoredPrefix<number> | undefined> {
return this.readPrefix(id)
loadStored(id: SessionId, signal?: AbortSignal): Promise<StoredPrefix<number> | 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<StoredPrefix<number> | undefined> {
private async readPrefix(id: SessionId, signal?: AbortSignal): Promise<StoredPrefix<number> | 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,18 +254,24 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
}
/** List all materialized sessions' metadata (every row is a materialized session). */
async list(): Promise<SessionHeader[]> {
async list(signal?: AbortSignal): Promise<SessionHeader[]> {
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)
}
/** List metadata with a source-qualified monotonic revision per session. */
async listSnapshots(): Promise<SessionPersistenceSnapshot[]> {
async listSnapshots(signal?: AbortSignal): Promise<SessionPersistenceSnapshot[]> {
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(row => ({
header: rowToMeta(row),
revision: SessionPersistenceRevision(

25
packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts
View File

@@ -580,6 +580,31 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
await second.dispose()
})
it('awaits in-flight readiness before surfacing snapshot-list cancellation', async () => {
const b = await backend()
const internals = b.ctx.sessionPersistence as unknown as { ready: Promise<void> }
const originalReady = internals.ready
const readiness = Promise.withResolvers<undefined>()
internals.ready = readiness.promise
const reason = new Error('SQLite snapshot readiness cancelled')
const controller = new AbortController()
const pending = b.ctx.sessionPersistence.listSnapshots(controller.signal)
let settled = false
void pending.then(
() => { settled = true },
() => { settled = true },
)
controller.abort(reason)
await Promise.resolve()
expect(settled).toBe(false)
readiness.resolve(undefined)
await expect(pending).rejects.toBe(reason)
internals.ready = originalReady
await b.dispose()
})
it('exposes the schema version constant', () => {
expect(SCHEMA_VERSION).toBe(10)
})

12
packages/session-persistence/session-persistence/README.md
View File

@@ -12,9 +12,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. |
| `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<SessionHeader[]>` | Lightweight listing from metadata, no full-log parse. A zero-event lazily-materialized session is absent from `list`. |
| `listSnapshots(): 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. |
| `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<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. |
## Invariants every backend must honor
@@ -33,17 +33,17 @@ Crash repair is cold-only. For a live id, `load(id)` snapshots the authoritative
When a live session emits `session/disposed`, the coordinator waits for its controller, serializes a final drain, then releases state owned by that exact `Session` object. Failed retirement leaves the controller in the live-session map, so backend teardown can retry it. Backend teardown stops event admission first, flushes every remaining controller, awaits per-id operations, and only then closes the storage handle.
The side-effect-free `locate` and lightweight `listSnapshots` queries remain backend-owned because they describe storage topology and revision identity rather than write orchestration.
The side-effect-free `locate` and lightweight `listSnapshots` queries remain backend-owned because they describe storage topology and revision identity rather than write orchestration. `listSnapshots(signal?)` passes the caller's exact signal into backend discovery so observers can cancel that work without detaching it.
The `PersistenceBackend<TornMarker>` hooks (the only seam between the coordinator and storage):
| 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).

94
packages/session-persistence/session-persistence/src/coordinator.ts
View File

@@ -40,8 +40,10 @@ export interface PersistenceBackend<TornMarker = unknown> {
* `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<StoredPrefix<TornMarker> | undefined>
loadStored(id: SessionId, signal?: AbortSignal): Promise<StoredPrefix<TornMarker> | undefined>
/**
* Durably append a CONTIGUOUS batch, lazily materializing the session first
@@ -60,8 +62,11 @@ export interface PersistenceBackend<TornMarker = unknown> {
*/
commitRepair(meta: SessionHeader, tornMarker: TornMarker | undefined, closers: readonly SessionEvent[]): Promise<void>
/** List all stored (materialized) sessions' metadata. */
list(): Promise<SessionHeader[]>
/**
* List all stored (materialized) sessions' metadata.
* @param signal - optional cancellation for backend listing work.
*/
list(signal?: AbortSignal): Promise<SessionHeader[]>
/**
* Optional lifecycle teardown (e.g. close a database handle). Awaited by the
@@ -273,15 +278,27 @@ export class PersistenceCoordinator<TornMarker = unknown> {
* 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[] }> {
inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
return Promise.resolve(this.retirements.get(id))
.then(() => this.serialize(id, () => this.inspectCore(id)))
.then(() => 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<TornMarker> | 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)
@@ -338,9 +355,19 @@ export class PersistenceCoordinator<TornMarker = unknown> {
* public methods must NOT call each other (deadlock); they call the unserialized
* `*Core` helpers instead.
*/
private serialize<T>(id: SessionId, op: () => Promise<T> | T): Promise<T> {
private serialize<T>(
id: SessionId,
op: () => Promise<T> | T,
signal?: AbortSignal,
): Promise<T> {
const prior = this.chains.get(id) ?? Promise.resolve()
const next = prior.then(op, op)
let started = false
const run = (): Promise<T> | 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)
@@ -350,7 +377,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
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. */
@@ -628,3 +655,50 @@ export class PersistenceCoordinator<TornMarker = unknown> {
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<T>(
operation: Promise<T>,
signal: AbortSignal,
started: () => boolean,
): Promise<T> {
return new Promise<T>((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)
}

9
packages/session-persistence/session-persistence/src/index.ts
View File

@@ -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<SessionHeader[]>
abstract list(signal?: AbortSignal): Promise<SessionHeader[]>
/**
* List materialized sessions with cheap per-log change tokens.
@@ -120,9 +122,10 @@ export abstract class SessionPersistence extends Service {
* successful mutating {@link load} repair changes the next listed revision.
* Revisions also distinguish independently backed stores so backend-local
* counters cannot compare equal across different persistence sources.
* @param signal - optional cancellation for backend snapshot-listing work.
* @returns one header and opaque revision per materialized session without loading full logs.
*/
abstract listSnapshots(): Promise<SessionPersistenceSnapshot[]>
abstract listSnapshots(signal?: AbortSignal): Promise<SessionPersistenceSnapshot[]>
}
export default SessionPersistence

17
packages/session-persistence/session-persistence/tests/contract.ts
View File

@@ -238,6 +238,23 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
}
})
it('rejects pre-aborted observation reads with the exact cancellation reason', async () => {
const { persistence, dispose } = await make()
try {
const reason = new Error('persistence observation cancelled')
const controller = new AbortController()
await expect(persistence.listSnapshots(controller.signal)).resolves.toEqual([])
controller.abort(reason)
await expect(persistence.list(controller.signal)).rejects.toBe(reason)
await expect(persistence.listSnapshots(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 {

2
packages/session-persistence/session-persistence/tests/coordinator-contract.ts
View File

@@ -683,7 +683,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
try {
// Ownerless state created WITHOUT a cwd (the no-cwd bucket).
// Ownerless state created WITHOUT a cwd (the `_no-cwd` project directory).
await ctx.sessionPersistence.create(meta('no-cwd-state'))
// A live session reusing the id but WITH cwd WORK is a cwd mismatch
// (undefined vs WORK) and must be rejected.

119
packages/session-persistence/session-persistence/tests/persistence.spec.ts
View File

@@ -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,11 +132,13 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
if (closers.length > 0) entry.events.push(...structuredClone(closers) as SessionEvent[])
}
async list(): Promise<SessionHeader[]> {
async list(signal?: AbortSignal): Promise<SessionHeader[]> {
signal?.throwIfAborted()
return [...this.store.values()].map(e => structuredClone(e.meta))
}
async listSnapshots(): Promise<SessionPersistenceSnapshot[]> {
async listSnapshots(signal?: AbortSignal): Promise<SessionPersistenceSnapshot[]> {
signal?.throwIfAborted()
return [...this.store.values()].map(entry => ({
header: structuredClone(entry.meta),
revision: SessionPersistenceRevision(`events:${entry.events.length}`),
@@ -153,10 +155,10 @@ class ControlledBackend implements PersistenceBackend<never> {
loadAttempts = 0
repairAttempts = 0
beforeAppend?: (attempt: number) => Promise<void>
beforeLoadStored?: (attempt: number) => Promise<void>
beforeLoadStored?: (attempt: number, signal?: AbortSignal) => Promise<void>
async loadStored(id: SessionId): Promise<StoredPrefix<never> | undefined> {
await this.beforeLoadStored?.(++this.loadAttempts)
async loadStored(id: SessionId, signal?: AbortSignal): Promise<StoredPrefix<never> | 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 +350,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<boolean>()
backend.beforeLoadStored = async (attempt) => {
if (attempt === 1) await loadGate.promise
}
let coordinator!: PersistenceCoordinator<never>
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<boolean>()
let cleanupComplete = false
backend.beforeLoadStored = async (_attempt, signal) => {
await new Promise<void>((resolve) => {
signal?.addEventListener('abort', () => {
void cleanupGate.promise.then(() => {
cleanupComplete = true
resolve()
})
}, { once: true })
})
throw new Error('backend cancellation after cleanup')
}
let coordinator!: PersistenceCoordinator<never>
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()