Merge remote-tracking branch 'origin/master' into codex/agent-session-jsonl-location
# Conflicts: # docs/config-catalog.md # docs/cordis-catalog/services.md # docs/core-data-structures/bash.md # docs/module-graph.md # docs/rfc/INDEX.md # docs/rfc/implemented/architecture/2026-06-30-bash-stdin-env-trusted-plugin-surface.md # docs/rfc/implemented/feature/2026-06-30-hook-bridges.md # examples/acp-agent/tests/snapshots/advanced-toolchain/session.1.jsonl # examples/acp-agent/tests/snapshots/advanced-toolchain/session.2.jsonl # examples/acp-agent/tests/snapshots/advanced-toolchain/session.jsonl # examples/acp-agent/tests/snapshots/advanced-toolchain/system-prompt.golden.md # examples/acp-agent/tests/snapshots/both-mode-turn/session.jsonl # examples/acp-agent/tests/snapshots/both-mode-turn/system-prompt.golden.md # examples/acp-agent/tests/snapshots/code-mode-turn/system-prompt.golden.md # examples/acp-agent/tests/snapshots/skill-load/session.jsonl # examples/acp-agent/tests/snapshots/text-turn/session.jsonl # examples/sandbox-acp-agent/tests/snapshots/mode-switching/session.jsonl # packages/bash/bash-local/README.md # packages/bash/bash-local/src/run.ts # packages/bash/bash-local/tests/run.spec.ts # packages/bash/bash/README.md # packages/bash/tool-bash/README.md # packages/bash/tool-bash/src/index.ts # packages/bash/tool-bash/tests/tools.spec.ts # packages/cordis/tool-cordis/src/api-catalog.ts # packages/core/agent-core/src/index.ts # packages/session-persistence/session-persistence-jsonl/src/index.ts # packages/session-persistence/session-persistence-sqlite/src/index.ts # packages/session-persistence/session-persistence/README.md # packages/session-persistence/session-persistence/src/index.ts # packages/ui/acp-agent/README.md # packages/ui/stdio-agent/README.md
This commit is contained in:
@@ -23,12 +23,26 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
|
||||
|
||||
## Durability and crash semantics
|
||||
|
||||
- **Lazy materialization.** `create(meta)` writes nothing; the `.jsonl` (header + first batch) is written atomically (temp-write + `fsync` + rename) on the first `append`. A created-but-never-appended session leaves nothing on disk and is absent from `list`.
|
||||
- **Lazy materialization.** `create(meta)` writes nothing; on the first `append`, the backend writes and `fsync`s a temporary file, publishes it without overwrite via a hard link, then `fsync`s the directory. A created-but-never-appended session leaves nothing on disk and is absent from `list`.
|
||||
- **Append-only.** Committed events (at or below a flushed `turn/end`) are never rewritten. Subsequent appends are line appends at EOF + `fsync`.
|
||||
- **Crash recovery — close, don't truncate.** A crash can leave a log whose final turn never closed (real events after the last `turn/end`). `load` PRESERVES those events (a turn can be huge — they are real work) and closes the orphaned turn by durably appending synthetic boundary events: an error `tool/result` for every `tool-call` the crash left unanswered (the loop logs the assistant message before running the tools, so a mid-tool crash leaves dangling calls — and `deriveMessages()` would replay an assistant tool-call with no result, which providers reject), then a `step/end` if a step was open, then `turn/end {kind:'interrupted'}`, returning a balanced log. Only a never-fully-written **torn tail fragment** (a final line with no newline / unparseable) is `ftruncate`d away before the closers are written. See [session persistence](../../../docs/rfc/implemented/architecture/2026-06-14-session-persistence.md).
|
||||
- **Contiguous-seq.** `load` rejects a mid-log parse error or `seq` gap (unloadable); `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.
|
||||
- **Format version.** Only the current `SESSION_FORMAT_VERSION` (v0) is supported; `load` rejects any other version. While the harness is unreleased the on-disk format is pre-release/unstable: a breaking format change is absorbed at v0 (no bump until the first tagged release) and non-current logs are rejected — there is no migration (no persisted user data to preserve).
|
||||
- **Crash recovery — preserve valid tail work.** `load` keeps the contiguous valid prefix of an interrupted final turn. It truncates from the first unparsable or sequence-gapped uncommitted record, then appends the synthetic tool, step, and turn closers required by the shared [persistence contract](../../../docs/rfc/implemented/architecture/2026-06-14-session-persistence.md); the same defect at or before the last committed `turn/end` rejects.
|
||||
- **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.
|
||||
|
||||
## Write path
|
||||
|
||||
The plugin generalizes the example `session-jsonl.ts`: it subscribes to `session/created` (capture the header; persist a fork's seed once), `session/event` (copy each already-frozen event into the persistence-owned write-behind buffer), and `session/flush`/dispose (drain that buffer through `append`). A per-session write cursor means a resumed session never re-appends already-stored events. Existing live sessions are seeded on plugin apply (HMR does not replay `session/created`). All backend operations for one session are serialized, and disposal awaits quiescence (every init + final drain) before returning, so no write lands after teardown.
|
||||
The plugin buffers frozen session events and drains them on flush or disposal. A per-session cursor prevents resumed sessions from re-appending stored events, and live sessions are seeded when the plugin loads. Operations for one session are serialized; disposal waits for initialization and the final drain so no write lands after teardown.
|
||||
|
||||
## Model Experience
|
||||
|
||||
### Resumed conversation history
|
||||
|
||||
**What the model sees**: JSONL storage contributes no live prompt or schema. Loading restores stored surface history and preserves prior request headers for reconstruction; the new loop composes its current envelope. Each unanswered call in an interrupted tail is balanced with the exact error text `Tool call interrupted by a crash; no result was recorded.` Raw `assistant/chunk` records do not duplicate messages.
|
||||
|
||||
**Token effect**: Zero live-request tokens. A resumed agent pays for retained history and its current envelope, plus the quoted repair result for each interrupted call.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **Only the current `SESSION_FORMAT_VERSION` (v0) loads** — the on-disk format is pre-release/unstable: a breaking format change is absorbed at v0 and non-current logs are rejected; there is no migration.
|
||||
- **Nothing deletes session files** — logs accumulate under `root` until removed externally (the seam has no deletion surface).
|
||||
- **Single-process assumption** — per-session serialization and the write cursor live in this process; two processes appending to the same `root` are not coordinated.
|
||||
- **Initial materialization requires hard-link support** — first append uses `link()` so same-id races fail instead of overwriting a committed log; a filesystem that cannot create hard links cannot host this backend.
|
||||
|
||||
@@ -72,19 +72,13 @@ function isHeaderLine(value: unknown): value is HeaderLine {
|
||||
}
|
||||
|
||||
/**
|
||||
* Encode an arbitrary string as a single safe path segment, injectively over
|
||||
* ALL JS (UTF-16) strings — including lone surrogates. A {@link SessionId} is
|
||||
* an unvalidated branded string, so this neutralizes `../`, absolute paths,
|
||||
* NUL, and separators before any filesystem use.
|
||||
* Encode an arbitrary string as a single safe path segment, injectively over ALL JS (UTF-16)
|
||||
* strings — including lone surrogates. A {@link SessionId} is an unvalidated branded string,
|
||||
* so this neutralizes `../`, absolute paths, NUL, and separators before any filesystem use.
|
||||
* Safe code units remain literal; every other unit, including `~`, becomes
|
||||
* `~XXXX`. Operating on code units preserves lone surrogates, while special-
|
||||
* casing `.` and `..` prevents traversal by an otherwise safe whole segment.
|
||||
*
|
||||
* Each UTF-16 code unit is either kept literal (the safe set `[A-Za-z0-9_-]`)
|
||||
* or escaped as `~XXXX` (its 4-hex-digit code unit). `~` is itself escaped, so
|
||||
* the mapping is injective and reversible: distinct inputs never collide. We
|
||||
* iterate code UNITS (`charCodeAt`), not code points, so a lone surrogate
|
||||
* escapes to a distinct `~XXXX` instead of being normalized to U+FFFD (which
|
||||
* `Buffer.from(…, 'utf8')` would do, breaking injectivity). `.` is in the safe
|
||||
* set for readability but the whole-segment tokens `.`/`..` are escaped so they
|
||||
* can never traverse.
|
||||
* @param raw - the string to encode; must be non-empty (throws on `''`).
|
||||
* @returns the escaped single path segment, decodable back to `raw`.
|
||||
*/
|
||||
@@ -142,38 +136,18 @@ export function eventLine(event: SessionEvent): string {
|
||||
|
||||
/**
|
||||
* Parse a JSONL log buffer into its preserved event prefix (the header is line
|
||||
* 0). Returns the longest prefix of complete, seq-contiguous events plus the
|
||||
* byte offset of the end of the last preserved line (`committedBytes`).
|
||||
* 0). Fully written events in an interrupted final turn remain part of the
|
||||
* prefix. The first unparsable record or seq gap after the last `turn/end`
|
||||
* marks a tolerated torn tail; the same hole in the committed region rejects.
|
||||
*
|
||||
* A crash can leave a durable log whose final turn never closed: real,
|
||||
* fully-written events sit after the last `turn/end`. Those are PRESERVED (a
|
||||
* single turn can be huge in a long-horizon task — truncating it would destroy
|
||||
* real work); the backend closes the orphaned open turn with a synthetic
|
||||
* `turn/end {kind:'interrupted'}` on reload (the session-persistence RFC). Only a TORN trailing
|
||||
* fragment — a final line never fully flushed (no newline, unparseable, or a
|
||||
* seq gap) — is excluded; it bounds the preserved region. A parse error or seq
|
||||
* gap AT OR BEFORE the last committed `turn/end` is committed-data corruption
|
||||
* and makes the session unloadable (throws).
|
||||
*
|
||||
* This relies on the session-log invariant that every event lives inside a turn
|
||||
* (`Session.append` enforces it): only the final turn can be open, so the
|
||||
* preserved tail is at most one unclosed turn.
|
||||
* @param buffer - the raw bytes of the log file (header line first).
|
||||
* @returns the header, the preserved event prefix, and `committedBytes` — the
|
||||
* byte offset the next append truncates any torn tail to.
|
||||
*/
|
||||
export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionEvent[]; committedBytes: number } {
|
||||
const text = buffer.toString('utf8')
|
||||
// Split into complete (newline-terminated) lines, tracking the byte offset of
|
||||
// each line's end so the truncation point is exact (multi-byte chars make the
|
||||
// char offset differ from the byte offset). A trailing line with no newline is
|
||||
// an uncommitted crash fragment and is ignored — it is below the last
|
||||
// turn/end by construction (the loop only flushes whole lines).
|
||||
//
|
||||
// Track the byte offset with a RUNNING accumulator (`endByte`), adding each
|
||||
// line's byte length as we go. Recomputing `Buffer.byteLength(text.slice(0, i))`
|
||||
// per newline would rescan the whole prefix every time — O(n²) over a long
|
||||
// log (one assistant/chunk line per token makes that pathological).
|
||||
// Track complete lines by byte offset: a non-newline tail is torn and ignored,
|
||||
// and a running counter avoids rescanning a long multi-byte log.
|
||||
const lines: { text: string; endByte: number }[] = []
|
||||
let start = 0
|
||||
let byteOffset = 0
|
||||
@@ -201,14 +175,8 @@ export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionE
|
||||
}
|
||||
const headerLine = parsedHeader
|
||||
|
||||
// Find the committed region: the prefix up to and including the LAST complete
|
||||
// `turn/end` in the WHOLE log. Two passes so a crash tail after the last
|
||||
// turn/end is tolerated, but corruption/gaps AT OR BEFORE the last committed
|
||||
// turn/end make the log unloadable (committed data must never be silently
|
||||
// dropped).
|
||||
//
|
||||
// Pass 1: parse every line that parses, recording (parsedOk, seq, isTurnEnd,
|
||||
// endByte) per line index. Lines that fail to parse are holes.
|
||||
// Parse every complete record first so the last valid `turn/end` determines
|
||||
// whether an earlier hole is committed corruption or an uncommitted tail.
|
||||
interface Parsed { ok: boolean; event?: SessionEvent; endByte: number }
|
||||
const parsed: Parsed[] = eventEntries.map((entry) => {
|
||||
try {
|
||||
@@ -226,18 +194,8 @@ export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionE
|
||||
if (p?.ok && p.event?.type === 'turn/end') { lastTurnEnd = i; break }
|
||||
}
|
||||
|
||||
// Walk the longest PREFIX of complete, seq-contiguous, parseable event lines
|
||||
// (line i is a parsed event with seq === i). This is the preservable region:
|
||||
// it includes any fully-written events of an interrupted final turn AFTER the
|
||||
// last turn/end — those are real, durably-written work and must NOT be
|
||||
// truncated (a single turn can be huge in a long-horizon task; the orphaned
|
||||
// open turn is closed with a synthetic turn/end on reload, not discarded —
|
||||
// the session-persistence RFC). The walk stops at the first hole (unparseable line or seq gap):
|
||||
// - if that hole is AT OR BEFORE the last committed turn/end, committed data
|
||||
// was damaged → the session is unloadable (throw);
|
||||
// - if it is AFTER (or there is no committed turn/end yet), it is the
|
||||
// tolerated crash boundary — a torn final line never fully flushed — and
|
||||
// it simply bounds the preserved tail.
|
||||
// Preserve the contiguous prefix, including a complete interrupted turn;
|
||||
// holes through the last committed boundary throw, while later holes stop.
|
||||
const preserved: SessionEvent[] = []
|
||||
for (let i = 0; i < parsed.length; i++) {
|
||||
const p = parsed[i]
|
||||
|
||||
@@ -1,20 +1,8 @@
|
||||
/**
|
||||
* JSONL durable session-persistence backend (`@deepseek-ai/dsh-session-persistence-jsonl`).
|
||||
*
|
||||
* One append-only `.jsonl` event log per session (a header line then one
|
||||
* `SessionEvent` per line, verbatim including `assistant/chunk` so `seq` stays
|
||||
* contiguous), with lazy materialization (no file until the first `append`),
|
||||
* atomic first write, and load-time repair of a never-committed crash tail.
|
||||
*
|
||||
* The backend supplies ONLY the file-bytes storage primitives (the
|
||||
* {@link PersistenceBackend} hooks below); all the write-path orchestration
|
||||
* (the `session/event` → buffer → `session/flush` drain, per-session
|
||||
* serialization, write cursors, fork-seed persistence, HMR live-adoption,
|
||||
* crash-repair sequencing, dispose quiescence) lives in the backend-agnostic
|
||||
* {@link PersistenceCoordinator} this class composes. The four stateful public
|
||||
* {@link SessionPersistence} methods delegate to the coordinator; the pure
|
||||
* locator remains backend-owned.
|
||||
*
|
||||
* JSONL durable session-persistence backend. It stores a header and contiguous
|
||||
* events in one append-only file per session, and delegates orchestration to
|
||||
* {@link PersistenceCoordinator}. Its side-effect-free locator returns the
|
||||
* absolute per-session log target before materialization.
|
||||
* @module @deepseek-ai/dsh-session-persistence-jsonl
|
||||
*/
|
||||
|
||||
@@ -42,13 +30,7 @@ export interface Config {
|
||||
root: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether `error` is a "no such file/directory" (`ENOENT`) failure — the ONLY
|
||||
* filesystem error that legitimately means "this session/root is absent" for a
|
||||
* durable backend. Any OTHER error (`EACCES`, `ENOTDIR`, transient I/O) must
|
||||
* surface rather than be silently reported as absence. (A NodeJS filesystem
|
||||
* rejection carries a string `code`.)
|
||||
*/
|
||||
/** Whether a filesystem error means absence; every non-ENOENT failure must surface. */
|
||||
function isENOENT(error: unknown): boolean {
|
||||
return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT'
|
||||
}
|
||||
@@ -66,13 +48,9 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
})
|
||||
|
||||
/**
|
||||
* Backend label for the coordinator's dispose-failure AggregateError and
|
||||
* effect name. NOTE: this intentionally shadows cordis `Service.name` (which
|
||||
* the base sets to `'sessionPersistence'`). The service is registered under the
|
||||
* fixed key the Service constructor captured (`reflect.provide('sessionPersistence', …)`),
|
||||
* not via `this.name`, so overwriting the instance field with the backend label
|
||||
* does not affect `ctx.sessionPersistence` resolution — it only relabels the
|
||||
* dispose diagnostics, which is exactly what {@link PersistenceBackend.name} is for.
|
||||
* Backend label for coordinator diagnostics and effects. It shadows
|
||||
* `Service.name` without changing the service key captured by the base
|
||||
* constructor.
|
||||
*/
|
||||
override readonly name = 'session-persistence-jsonl'
|
||||
|
||||
@@ -81,10 +59,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
|
||||
constructor(ctx: Context, public config: Config) {
|
||||
super(ctx)
|
||||
// Resolve the configured root to an ABSOLUTE path ONCE, here. A relative root
|
||||
// would otherwise re-resolve against `process.cwd()` at every later
|
||||
// readdir/open — so if any plugin or test changed cwd between create, append,
|
||||
// and load, one session's files could split across directories.
|
||||
// Resolve once so later process.cwd() changes cannot split one backend across roots.
|
||||
this.root = resolve(config.root)
|
||||
this.coordinator = new PersistenceCoordinator<number>(this.ctx, this)
|
||||
}
|
||||
@@ -111,11 +86,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
return this.coordinator.load(id)
|
||||
}
|
||||
|
||||
// `list` is BOTH the public service method and the PersistenceBackend hook —
|
||||
// one method, the bucket walk below. The coordinator adds no orchestration for
|
||||
// listing (no per-id serialization, no cursor), so it would just call back into
|
||||
// this same method; routing it through the coordinator would recurse. Defined
|
||||
// once, in the "PersistenceBackend hooks" section.
|
||||
// One method serves both public `list` and the backend hook; delegating it to
|
||||
// the coordinator would call this hook recursively.
|
||||
|
||||
/**
|
||||
* The per-session init promises, exposed for white-box tests that await a
|
||||
@@ -128,7 +100,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
|
||||
// --- PersistenceBackend hooks (the file-bytes storage primitives) ---
|
||||
|
||||
/** Read a stored prefix by id across ALL cwd buckets (cwd unknown). */
|
||||
/** Read a stored prefix by id across all cwd buckets when cwd is unknown. */
|
||||
async loadStored(id: SessionId): Promise<StoredPrefix<number> | undefined> {
|
||||
const file = await this.findLog(id)
|
||||
if (file === undefined) return undefined
|
||||
@@ -136,11 +108,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a stored prefix SCOPED to `cwd` (HMR live-adoption must not cross cwd).
|
||||
* `undefined` is the DEFINITE "no-cwd" bucket, NOT "unknown" — a live session
|
||||
* with no cwd may only adopt a persisted no-cwd log, never a same-id log that
|
||||
* lives in some other cwd bucket. So this looks at exactly `logPath(cwd)`
|
||||
* (which maps `undefined` → the `_no-cwd` bucket), never the all-buckets scan.
|
||||
* Read a stored prefix within one cwd for HMR adoption. `undefined` names the
|
||||
* no-cwd bucket rather than an unknown cwd, so this never scans other buckets.
|
||||
*/
|
||||
async loadLive(id: SessionId, cwd: string | undefined): Promise<StoredPrefix<number> | undefined> {
|
||||
const path = logPath(this.root, cwd, id)
|
||||
@@ -149,10 +118,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
}
|
||||
|
||||
/**
|
||||
* Read and scan a session's log file into a {@link StoredPrefix}. Folds the
|
||||
* torn-tail comparison HERE so the `tornMarker` is the byte offset to truncate
|
||||
* to (or `undefined` when nothing is torn) — the coordinator never sees the
|
||||
* raw byteLength.
|
||||
* Read a stored prefix and convert torn-tail state to the byte offset the
|
||||
* coordinator can round-trip without knowing the file format.
|
||||
*/
|
||||
private async readPrefix(path: string): Promise<StoredPrefix<number>> {
|
||||
const buffer = await readFile(path)
|
||||
@@ -188,9 +155,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
const metas: SessionHeader[] = []
|
||||
for (const dir of await this.listCwdDirs()) {
|
||||
for (const name of await this.listJsonl(dir)) {
|
||||
// Read ONLY the header line, not the whole log: a session picker must
|
||||
// scale with the number of sessions, not the total size of every
|
||||
// conversation (the log persists every assistant/chunk verbatim).
|
||||
// Read only headers so listing scales with session count, not log size.
|
||||
const first = await this.readFirstLine(`${dir}/${name}`)
|
||||
if (first === undefined) continue // empty/half-written file
|
||||
const meta = parseHeaderMeta(first)
|
||||
@@ -203,7 +168,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
|
||||
// --- materialization / append / repair (file mechanics) ---
|
||||
|
||||
/** Atomically write the header line + first batch (temp-write, fsync, rename). */
|
||||
/** Atomically write the header line + first batch (temp-write, fsync, collision-safe hard-link publish). */
|
||||
private async materialize(meta: SessionHeader, events: readonly SessionEvent[]): Promise<void> {
|
||||
const dir = sessionDir(this.root, meta.cwd)
|
||||
await mkdir(this.root, { recursive: true, mode: 0o700 })
|
||||
@@ -211,10 +176,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
await mkdir(dir, { recursive: true, mode: 0o700 })
|
||||
await this.syncDir(this.root)
|
||||
const finalPath = logPath(this.root, meta.cwd, meta.id)
|
||||
// Never rename over an existing committed log: materialize is the FIRST write
|
||||
// of a session the backend believes is new. A file here means a different
|
||||
// session shares this id on disk — reject loudly. (createCore already guards
|
||||
// the create path, so this is unreachable-in-practice TOCTOU defense.)
|
||||
// Materialization is the first write; an existing log is an id collision.
|
||||
/* v8 ignore next 3 -- createCore guards collisions before materialize; this is a TOCTOU backstop */
|
||||
if (await this.exists(finalPath)) {
|
||||
throw new Error(`refusing to materialize "${meta.id}": a log already exists on disk (load/resume it instead)`)
|
||||
@@ -231,28 +193,22 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
// Publish via link()+unlink(), NOT rename(): link fails with EEXIST if the
|
||||
// final path already exists, so two processes materializing the same id
|
||||
// concurrently cannot clobber each other. rename() would silently overwrite.
|
||||
// Publish with link()+unlink(): unlike rename(), link fails if another
|
||||
// process materialized the same id first.
|
||||
let linked = false
|
||||
try {
|
||||
await link(tmp, finalPath)
|
||||
linked = true
|
||||
} finally {
|
||||
// If link FAILED, the temp is the only reference and must be removed before
|
||||
// the original error propagates. If it SUCCEEDED, defer temp cleanup to
|
||||
// AFTER the publish is durable (below) so a temp-rm failure can never reject
|
||||
// a session whose log already published.
|
||||
// Remove an unpublished temp on failure. After publication, defer cleanup
|
||||
// until the directory entry is durable so cleanup cannot reject a live log.
|
||||
/* v8 ignore next -- link failure is the TOCTOU/IO race guarded above; not reachable in test */
|
||||
if (!linked) await rm(tmp, { force: true })
|
||||
}
|
||||
// link() succeeded — the log is published. fsync the directory so the new
|
||||
// entry survives a power loss: the new link is not crash-durable until the
|
||||
// parent directory's metadata is synced.
|
||||
// The published link becomes crash-durable only after its directory fsync.
|
||||
await this.syncDir(dir)
|
||||
// Best-effort temp cleanup: the log is already published and durable, so a
|
||||
// failure to remove the (now-redundant) temp hard link must NOT reject the
|
||||
// append. Swallow only the rm failure; nothing else of consequence runs here.
|
||||
// Best-effort temp cleanup: the log is already published and durable, so a failure to
|
||||
// remove the (now-redundant) temp hard link must not reject the append.
|
||||
try {
|
||||
await rm(tmp, { force: true })
|
||||
} catch {
|
||||
@@ -260,7 +216,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
}
|
||||
}
|
||||
|
||||
/** fsync a directory so a just-created/renamed entry inside it is crash-durable. */
|
||||
/** fsync a directory so a just-created or published entry inside it is crash-durable. */
|
||||
private async syncDir(dir: string): Promise<void> {
|
||||
const handle = await open(dir, 'r')
|
||||
try {
|
||||
@@ -271,11 +227,9 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
}
|
||||
|
||||
/**
|
||||
* Append event lines at EOF and fsync. On a write/sync failure AFTER the kernel
|
||||
* accepted some bytes (ENOSPC, an fsync error), truncate the file back to its
|
||||
* pre-append size before rethrowing: the cursor is unchanged, so the batch will
|
||||
* be retried, and without this rollback the retry would append AFTER the partial
|
||||
* bytes — producing duplicate seqs that make `scanLog` see a gap.
|
||||
* Append and fsync event lines. On a partial write or sync failure, restore the
|
||||
* previous size before rethrowing because the unchanged cursor will retry the
|
||||
* batch; leaving partial bytes would create duplicate sequence numbers.
|
||||
*/
|
||||
private async appendLines(meta: SessionHeader, events: readonly SessionEvent[]): Promise<void> {
|
||||
const path = logPath(this.root, meta.cwd, meta.id)
|
||||
@@ -337,10 +291,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
}
|
||||
|
||||
/**
|
||||
* Find a session's log file by id across ALL cwd buckets — the any-cwd scan
|
||||
* for `loadStored` (resume identifies a session by id alone). The cwd-scoped
|
||||
* lookup (`loadLive`) does NOT use this; it goes straight to `logPath(cwd)` so
|
||||
* a no-cwd session can't match a real-cwd bucket.
|
||||
* Find a session by id across cwd buckets for resume. Cwd-scoped HMR adoption
|
||||
* bypasses this scan so a no-cwd session cannot claim another bucket.
|
||||
*/
|
||||
private async findLog(id: SessionId): Promise<{ path: string; cwd: string | undefined } | undefined> {
|
||||
const target = encodeSegment(id) + '.jsonl'
|
||||
@@ -361,9 +313,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
const entries = await readdir(this.root, { withFileTypes: true })
|
||||
return entries.filter(e => e.isDirectory()).map(e => `${this.root}/${e.name}`)
|
||||
} catch (error) {
|
||||
// ENOENT = the root has not been created yet → genuinely no sessions. Any
|
||||
// other error (EACCES, ENOTDIR, transient I/O) must NOT be reported as "no
|
||||
// sessions" — a durable backend cannot silently pretend state is absent.
|
||||
// Only an absent root means no sessions; rethrow every other I/O failure.
|
||||
if (isENOENT(error)) return []
|
||||
throw error
|
||||
}
|
||||
|
||||
@@ -54,12 +54,8 @@ runPersistenceContract('jsonl', async () => {
|
||||
}
|
||||
})
|
||||
|
||||
// Run the shared coordinator orchestration suite against the real JSONL backend.
|
||||
// One temp root is the shared storage scope (two mounted instances over the same
|
||||
// root = HMR/reload). `corruptTail` appends a partial, newline-less fragment to
|
||||
// the session's .jsonl past the committed region — a never-committed torn tail
|
||||
// that drives the coordinator's commitRepair-with-tornMarker branch over real
|
||||
// file bytes.
|
||||
// Two mounts share this temp root to exercise reload. `corruptTail` appends a partial,
|
||||
// newline-less fragment past the committed region so coordinator repair runs on real file bytes.
|
||||
runCoordinatorContract('jsonl', async (): Promise<CoordinatorFixture> => {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-coord-'))
|
||||
return {
|
||||
@@ -385,10 +381,9 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }), // gap: missing seq 1
|
||||
].join('\n') + '\n'
|
||||
// No committed turn/end, so the gap is a tolerated crash boundary: scanLog
|
||||
// PRESERVES the contiguous prefix (turn/start seq 0) — real interrupted-turn
|
||||
// work, not discarded — and stops at the gap. The orphaned open turn is
|
||||
// closed by loadCore's synthetic turn/end, not here.
|
||||
// No committed turn/end, so the gap is a tolerated crash boundary: scanLog PRESERVES the
|
||||
// contiguous prefix (turn/start seq 0) — real interrupted-turn work, not discarded — and
|
||||
// stops at the gap. `loadCore`, not this scanner, later closes the orphaned turn.
|
||||
expect(scanLog(Buffer.from(log)).events.map(e => e.seq)).toEqual([0])
|
||||
})
|
||||
|
||||
@@ -508,9 +503,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
})
|
||||
|
||||
it('list reads a header line longer than the 8KB read chunk', async () => {
|
||||
// readFirstLine accumulates across reads when the first line exceeds its
|
||||
// buffer. Plant a valid header whose line is > 8192 bytes (a long extra
|
||||
// field is tolerated by the header type guard) and confirm list() reads it.
|
||||
// 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 bigHeader = JSON.stringify({ type: 'session', version: 0, id: 'big', createdAt: 1, pad: 'x'.repeat(9000) })
|
||||
@@ -530,10 +524,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
for (const s of ctx.sessions.list()) await ctx.parallel('session/flush', s)
|
||||
await sessFiberA.dispose()
|
||||
|
||||
// A NEW live Session object reuses id "reuse". The init cache is keyed by
|
||||
// the Session OBJECT, so this gets its OWN onCreated (not A's stale promise)
|
||||
// — which detects the on-disk collision and rejects, rather than silently
|
||||
// appending the new session's events onto A's log under a stale cursor.
|
||||
// A new Session object reuses the id. Object-keyed initialization must run independently,
|
||||
// detect the disk collision, and reject instead of appending through session A's stale cursor.
|
||||
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
|
||||
let b!: Session
|
||||
await ctx.plugin(Object.assign((inner: Context) => {
|
||||
@@ -551,13 +543,9 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
await ctx.sessionPersistence.append(SessionId('x'), oneTurnLog())
|
||||
await ctx.fiber.dispose()
|
||||
|
||||
// Backend 2 over the SAME root. A live no-cwd session reuses id "x". Because
|
||||
// loadLive(id, undefined) is the DEFINITE no-cwd bucket (NOT an all-buckets
|
||||
// scan), case-2 adoption does NOT match the "/w" log — so it would NOT
|
||||
// silently graft the no-cwd events onto the "/w" log with a mismatched cwd
|
||||
// (the bug a non-scope-exact loadLive caused). It falls through to the
|
||||
// new-session path, where createCore's any-cwd collision probe (loadStored)
|
||||
// catches the duplicate id and REJECTS — the id is taken in another bucket.
|
||||
// Backend 2 creates a no-cwd session whose id exists only in `/w`. Exact `loadLive(id,
|
||||
// undefined)` must not adopt across buckets; the any-cwd collision check then rejects instead
|
||||
// of grafting no-cwd events onto a log with mismatched cwd.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
@@ -625,9 +613,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
})
|
||||
|
||||
it('list surfaces a non-ENOENT root error (ENOTDIR) instead of reporting no sessions', async () => {
|
||||
// A durable backend must NOT collapse a storage fault to "no sessions". Point
|
||||
// the root at a regular FILE: readdir then fails with ENOTDIR, which must
|
||||
// propagate rather than be swallowed as an empty listing.
|
||||
// A durable backend must not collapse a storage fault to "no sessions". Making the root a
|
||||
// regular file forces ENOTDIR from `readdir`, which must propagate.
|
||||
const filePath = join(root, 'not-a-dir')
|
||||
await writeFile(filePath, 'x')
|
||||
const ctx2 = new Context()
|
||||
@@ -638,11 +625,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
})
|
||||
|
||||
it('loadLive surfaces a non-ENOENT lookup error (ENOTDIR) instead of reporting absent', async () => {
|
||||
// A non-ENOENT error from the per-id open() must surface, not be collapsed to
|
||||
// "not found" (which would let live-adoption proceed under a false absence
|
||||
// assumption). A live session's onCreated reaches loadLive(id, cwd) →
|
||||
// exists(logPath). Make that cwd's bucket DIRECTORY a regular file: open()ing
|
||||
// `bucket/<id>.jsonl` under it then fails ENOTDIR.
|
||||
// A non-ENOENT per-id open error must surface rather than become "not found" and permit false
|
||||
// live adoption. Making the cwd bucket a regular file forces ENOTDIR for its child log path.
|
||||
const cwd = '/x'
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
@@ -761,10 +745,9 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
|
||||
it('Session.append rejects a non-serializable event at the source (never enters the log)', () => {
|
||||
const session = ctx.sessions.create(SessionId('reject-bad'))
|
||||
// Serializability is enforced at the source: Session.append throws on a
|
||||
// BigInt-bearing event BEFORE it enters session.events, so the durable log
|
||||
// can never diverge from the live log. The throw surfaces at the caller's
|
||||
// append site, not asynchronously in a backend flush.
|
||||
// Serializability is enforced at the source: Session.append throws on a BigInt-bearing
|
||||
// event before it enters session.events, so the durable log can never diverge from the live
|
||||
// log. The error therefore surfaces synchronously at append, not later during backend flush.
|
||||
expect(() => {
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'bad' }], source: { kind: 'user' }, bad: 1n } as never, { surfaceOp: 'append' })
|
||||
}).toThrow(/non-JSON-serializable/)
|
||||
|
||||
Reference in New Issue
Block a user