feat: add durable JSONL storage backend

2026-11-05 23:37:25 +00:00 · 2026-09-23 01:33:31 +02:00 · 2026-09-23 01:33:31 +02:00 · 898ab80405
commit 898ab80405
parent 9672462143
21 changed files with 3431 additions and 84 deletions
--- a/packages/durable/README.md
+++ b/packages/durable/README.md
@ -2,7 +2,7 @@

 Durable conversation, task, and document runtime for Pi.

-This package contains the Pico runtime. Its current public API provides the durable record contracts and detached in-memory storage implementation:
+This package contains the Pico runtime. Its current public API provides durable record contracts and memory, JSONL, and SQLite storage implementations:

 ```ts
 import { MemoryStorage, ROOT_CONVERSATION_ID } from "@earendil-works/pi-durable";
@ -14,14 +14,19 @@ The root export is runtime-neutral. Storage implementations also have explicit s
 import { MemoryStorage } from "@earendil-works/pi-durable/storage/memory";
 ```

-Node applications can open file-backed SQLite through its Node-only subpath:
+Node applications can open file-backed JSONL or SQLite storage through Node-only subpaths:

 ```ts
+import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
+import { openNodeJsonlStorage } from "@earendil-works/pi-durable/storage/jsonl/node";
 import { openNodeSqliteStorage } from "@earendil-works/pi-durable/storage/sqlite/node";

-const storage = await openNodeSqliteStorage("./session.sqlite");
+const jsonl = await openNodeJsonlStorage("./session", BACKGROUND_CONTEXT);
+const sqlite = await openNodeSqliteStorage("./session.sqlite");
 ```

+The portable JSONL core accepts the `FileSystem` capability exported from `@earendil-works/pi-durable/env`. Its `fsync` option defaults to `false`; enabling it flushes affected sidecars before appending the main commit marker. One `JsonlStorage` owner must serialize writes to a storage directory; cross-process locking and ID allocation are not supported.
+
 The portable SQLite core, minimal database facade, and ordered schema migrations are exported from `@earendil-works/pi-durable/storage/sqlite`. Adapters for synchronous SQLite environments such as Bun and Cloudflare Durable Objects can implement that facade without importing Node APIs. Remote asynchronous APIs such as Cloudflare D1 cannot implement this synchronous facade; they require a dedicated `Storage` backend.

 The Node adapter uses WAL mode with `synchronous = NORMAL` and checkpoints the WAL on close. Acknowledged commits survive process crashes, but the newest commits may be lost after a power or host failure. One `SqliteStorage` owner must serialize writes to a database file; cross-process ID allocation is not supported.
@ -35,7 +40,7 @@ npm run bench:storage
 npm run bench:storage:memory
 ```

-The timing suite compares memory and file-backed SQLite across representative commits, indexed reads, pagination, fork traversal, document replay, historical reads, and SQLite reopen. The memory suite measures each backend in a separate process at 1k and 10k scales and reports heap, RSS, external memory, and SQLite file/page metrics. These deterministic synthetic workloads are baselines for regression analysis, not production capacity limits or CI pass/fail thresholds.
+The timing suite compares memory, JSONL, and SQLite across representative commits, indexed reads, pagination, fork traversal, document replay, historical reads, and persistent-backend reopen. The footprint suite measures each backend in a separate process at 1k and 10k scales and reports heap, RSS, external memory, file counts, and on-disk JSONL/SQLite size. These deterministic synthetic workloads are baselines for regression analysis, not production capacity limits or CI pass/fail thresholds.

 The normative design and implementation sequence are in:

--- a/packages/durable/docs/pico-v5-handoff.md
+++ b/packages/durable/docs/pico-v5-handoff.md
@ -61,8 +61,9 @@ Copy, rather than import, the current `ExecutionEnv`, `FileSystem`, `Shell`, Nod
 implementation, and their required utility files from `packages/agent/src/harness`
 into `packages/durable/src/env`. Copy only the environment-related slice, not
 agent skills, prompts, telemetry, or tool definitions. Extend the copied
-filesystem contract with exact-byte file truncation and file flushing. Keep the
-portable environment and JSONL entry points free of Node built-ins; expose Node
+filesystem contract with exact-byte file truncation and file flushing. JSONL
+depends only on `FileSystem`, not the broader `ExecutionEnv`. Keep the portable
+environment and JSONL entry points free of Node built-ins; expose Node
 implementations only from `/env/node` and `/storage/jsonl/node`. Do not use the
 Pico3 implementation as source material.

--- a/packages/durable/docs/pico-v5.md
+++ b/packages/durable/docs/pico-v5.md
@ -2096,11 +2096,12 @@ benchmarks.

 ### 11.3 JSONL

-JSONL uses reclaimable sidecars without exposing them to the harness. Persistence
-alone does not provide the ownership boundary: any decoded indexes, materialized
-values, or caches retained in memory must be detached from commit arguments and
-must not be exposed directly by reads. A JSONL backend cannot simply add file
-appends around aliasing memory tables.
+JSONL depends on the portable `FileSystem` capability, not the broader
+`ExecutionEnv`. It uses reclaimable sidecars without exposing them to the
+harness. Persistence alone does not provide the ownership boundary: any decoded
+indexes, materialized values, or caches retained in memory must be detached from
+commit arguments and must not be exposed directly by reads. A JSONL backend
+cannot simply add file appends around aliasing memory tables.

 ```text
 main.jsonl       table writes, document records, and one marker per commit
--- a/packages/durable/package.json
+++ b/packages/durable/package.json
@ -11,11 +11,31 @@
 			"types": "./dist/index.d.ts",
 			"import": "./dist/index.js"
 		},
+		"./env": {
+			"source": "./src/env/index.ts",
+			"types": "./dist/env/index.d.ts",
+			"import": "./dist/env/index.js"
+		},
+		"./env/node": {
+			"source": "./src/env/node.ts",
+			"types": "./dist/env/node.d.ts",
+			"import": "./dist/env/node.js"
+		},
 		"./storage/memory": {
 			"source": "./src/storage/memory.ts",
 			"types": "./dist/storage/memory.d.ts",
 			"import": "./dist/storage/memory.js"
 		},
+		"./storage/jsonl": {
+			"source": "./src/storage/jsonl/index.ts",
+			"types": "./dist/storage/jsonl/index.d.ts",
+			"import": "./dist/storage/jsonl/index.js"
+		},
+		"./storage/jsonl/node": {
+			"source": "./src/storage/jsonl/node.ts",
+			"types": "./dist/storage/jsonl/node.d.ts",
+			"import": "./dist/storage/jsonl/node.js"
+		},
 		"./storage/sqlite": {
 			"source": "./src/storage/sqlite/index.ts",
 			"types": "./dist/storage/sqlite/index.d.ts",
--- a/packages/durable/src/env/index.ts
+++ b/packages/durable/src/env/index.ts
@ -0,0 +1,189 @@
+import type { Context } from "@earendil-works/chord";
+import type { TruncationResult } from "./utils/truncate.ts";
+
+/** Result of a fallible operation. Expected failures are returned instead of thrown. */
+export type Result<TValue, TError> = { ok: true; value: TValue } | { ok: false; error: TError };
+
+export function ok<TValue, TError>(value: TValue): Result<TValue, TError> {
+	return { ok: true, value };
+}
+
+export function err<TValue, TError>(error: TError): Result<TValue, TError> {
+	return { ok: false, error };
+}
+
+export function getOrThrow<TValue, TError>(result: Result<TValue, TError>): TValue {
+	if (!result.ok) throw result.error;
+	return result.value;
+}
+
+export function getOrUndefined<TValue extends object, TError>(result: Result<TValue, TError>): TValue | undefined {
+	return result.ok ? result.value : undefined;
+}
+
+export function toError(error: unknown): Error {
+	if (error instanceof Error) return error;
+	if (typeof error === "string") return new Error(error);
+	try {
+		return new Error(JSON.stringify(error));
+	} catch {
+		return new Error(String(error));
+	}
+}
+
+export type FileKind = "file" | "directory" | "symlink";
+
+export type FileErrorCode =
+	| "aborted"
+	| "not_found"
+	| "permission_denied"
+	| "not_directory"
+	| "is_directory"
+	| "invalid"
+	| "not_supported"
+	| "unknown";
+
+export class FileError extends Error {
+	public code: FileErrorCode;
+	public path?: string;
+
+	constructor(code: FileErrorCode, message: string, path?: string, cause?: Error) {
+		super(message, cause === undefined ? undefined : { cause });
+		this.name = "FileError";
+		this.code = code;
+		this.path = path;
+	}
+}
+
+export type ExecutionErrorCode =
+	| "aborted"
+	| "timeout"
+	| "shell_unavailable"
+	| "spawn_error"
+	| "callback_error"
+	| "unknown";
+
+export class ExecutionError extends Error {
+	public code: ExecutionErrorCode;
+
+	constructor(code: ExecutionErrorCode, message: string, cause?: Error) {
+		super(message, cause === undefined ? undefined : { cause });
+		this.name = "ExecutionError";
+		this.code = code;
+	}
+}
+
+export interface FileInfo {
+	name: string;
+	path: string;
+	kind: FileKind;
+	size: number;
+	mtimeMs: number;
+}
+
+export interface TextLine {
+	text: string;
+	terminated: boolean;
+}
+
+export interface TextLineReader {
+	readLine(context: Context): Promise<Result<TextLine | undefined, FileError>>;
+	close(context: Context): Promise<void>;
+}
+
+/** Portable filesystem capability. Operations return failures rather than throwing. */
+export interface FileSystem {
+	cwd: string;
+	absolutePath(path: string, context: Context): Promise<Result<string, FileError>>;
+	joinPath(parts: string[], context: Context): Promise<Result<string, FileError>>;
+	readTextFile(path: string, context: Context): Promise<Result<string, FileError>>;
+	openTextLineReader(path: string, context: Context): Promise<Result<TextLineReader, FileError>>;
+	readTextLines(
+		path: string,
+		options: { maxLines?: number } | undefined,
+		context: Context,
+	): Promise<Result<string[], FileError>>;
+	readBinaryFile(path: string, context: Context): Promise<Result<Uint8Array, FileError>>;
+	writeFile(path: string, content: string | Uint8Array, context: Context): Promise<Result<void, FileError>>;
+	appendFile(path: string, content: string | Uint8Array, context: Context): Promise<Result<void, FileError>>;
+	/** Truncate or extend a file to exactly `size` bytes. */
+	truncateFile(path: string, size: number, context: Context): Promise<Result<void, FileError>>;
+	/** Flush file contents and metadata needed to retrieve them from an open file handle. */
+	flushFile(path: string, context: Context): Promise<Result<void, FileError>>;
+	renameFile(sourcePath: string, destinationPath: string, context: Context): Promise<Result<void, FileError>>;
+	fileInfo(path: string, context: Context): Promise<Result<FileInfo, FileError>>;
+	listDir(path: string, context: Context): Promise<Result<FileInfo[], FileError>>;
+	canonicalPath(path: string, context: Context): Promise<Result<string, FileError>>;
+	exists(path: string, context: Context): Promise<Result<boolean, FileError>>;
+	createDir(
+		path: string,
+		options: { recursive?: boolean } | undefined,
+		context: Context,
+	): Promise<Result<void, FileError>>;
+	remove(
+		path: string,
+		options: { recursive?: boolean; force?: boolean } | undefined,
+		context: Context,
+	): Promise<Result<void, FileError>>;
+	createTempDir(prefix: string | undefined, context: Context): Promise<Result<string, FileError>>;
+	createTempFile(
+		options: { prefix?: string; suffix?: string } | undefined,
+		context: Context,
+	): Promise<Result<string, FileError>>;
+	cleanup(context: Context): Promise<void>;
+}
+
+export type ShellOutputRetention = "head" | "tail";
+
+export interface ShellOutputLimits {
+	maxBytes: number;
+	maxLines: number;
+	retain?: ShellOutputRetention;
+}
+
+export interface ShellOutputCaptureOptions {
+	limits: ShellOutputLimits;
+	spill?: boolean;
+}
+
+export type ShellOutputTruncation = Omit<TruncationResult, "content">;
+
+export interface ShellOutputMetadata {
+	truncation: ShellOutputTruncation;
+	spillPath?: string;
+	lastLineBytes?: number;
+}
+
+export interface ShellOutputView extends ShellOutputMetadata {
+	text: string;
+}
+
+export type ShellOutputUpdate =
+	| { kind: "replace"; output: ShellOutputView }
+	| { kind: "append"; text: string; metadata: ShellOutputMetadata }
+	| { kind: "slide"; drop: number; text: string; metadata: ShellOutputMetadata }
+	| { kind: "metadata"; metadata: ShellOutputMetadata };
+
+export interface ShellExecResult extends ShellOutputMetadata {
+	exitCode: number;
+}
+
+export interface ShellExecOptions {
+	cwd?: string;
+	env?: Record<string, string>;
+	inheritEnv?: boolean;
+	timeout?: number;
+	capture?: ShellOutputCaptureOptions;
+	onUpdate?: (update: ShellOutputUpdate, context: Context) => void;
+}
+
+export interface Shell {
+	exec(
+		command: string,
+		options: ShellExecOptions | undefined,
+		context: Context,
+	): Promise<Result<ShellExecResult, ExecutionError>>;
+	cleanup(context: Context): Promise<void>;
+}
+
+export interface ExecutionEnv extends FileSystem, Shell {}
--- a/packages/durable/src/env/node.ts
+++ b/packages/durable/src/env/node.ts
@ -0,0 +1,963 @@
+import { type ChildProcess, spawn } from "node:child_process";
+import { randomUUID } from "node:crypto";
+import { constants, createWriteStream, type WriteStream } from "node:fs";
+import {
+	access,
+	appendFile,
+	lstat,
+	mkdir,
+	mkdtemp,
+	open as openFile,
+	readdir,
+	readFile,
+	realpath,
+	rename,
+	rm,
+	writeFile,
+} from "node:fs/promises";
+import { homedir, constants as osConstants, tmpdir } from "node:os";
+import { basename, isAbsolute, join, resolve } from "node:path";
+import { fileURLToPath } from "node:url";
+import type { Context } from "@earendil-works/chord";
+import {
+	type ExecutionEnv,
+	ExecutionError,
+	err,
+	FileError,
+	type FileInfo,
+	type FileKind,
+	ok,
+	type Result,
+	type ShellExecOptions,
+	type ShellExecResult,
+	type TextLine,
+	type TextLineReader,
+	toError,
+} from "./index.ts";
+import { OutputCapture } from "./utils/output-capture.ts";
+
+const MAX_TIMEOUT_MS = 2_147_483_647;
+const MAX_TIMEOUT_SECONDS = MAX_TIMEOUT_MS / 1000;
+const EXIT_STDIO_GRACE_MS = 100;
+const SPILL_HIGH_WATER_MARK = 8 * 1024 * 1024;
+
+type SpillChunk = string | Uint8Array;
+
+function resolveTimeoutMs(timeout: number | undefined): Result<number | undefined, ExecutionError> {
+	if (timeout === undefined) return ok(undefined);
+	if (!Number.isFinite(timeout) || timeout <= 0) {
+		return err(new ExecutionError("timeout", "Invalid timeout: must be a finite number of seconds"));
+	}
+
+	const timeoutMs = timeout * 1000;
+	if (timeoutMs > MAX_TIMEOUT_MS) {
+		return err(new ExecutionError("timeout", `Invalid timeout: maximum is ${MAX_TIMEOUT_SECONDS} seconds`));
+	}
+	return ok(timeoutMs);
+}
+
+function resolvePath(cwd: string, path: string): string {
+	let normalized = path;
+	if (normalized === "~") {
+		normalized = homedir();
+	} else if (normalized.startsWith("~/") || (process.platform === "win32" && normalized.startsWith("~\\"))) {
+		normalized = join(homedir(), normalized.slice(2));
+	} else if (normalized.startsWith("file://")) {
+		try {
+			normalized = fileURLToPath(normalized);
+		} catch {
+			// Keep malformed URLs as ordinary paths so filesystem methods preserve their non-throwing contract.
+		}
+	}
+	return isAbsolute(normalized) ? resolve(normalized) : resolve(cwd, normalized);
+}
+
+function fileKindFromStats(stats: {
+	isFile(): boolean;
+	isDirectory(): boolean;
+	isSymbolicLink(): boolean;
+}): FileKind | undefined {
+	if (stats.isFile()) return "file";
+	if (stats.isDirectory()) return "directory";
+	if (stats.isSymbolicLink()) return "symlink";
+	return undefined;
+}
+
+function fileInfoFromStats(
+	path: string,
+	stats: { isFile(): boolean; isDirectory(): boolean; isSymbolicLink(): boolean; size: number; mtimeMs: number },
+): Result<FileInfo, FileError> {
+	const kind = fileKindFromStats(stats);
+	if (!kind) return err(new FileError("invalid", "Unsupported file type", path));
+	return ok({
+		name: basename(path),
+		path,
+		kind,
+		size: stats.size,
+		mtimeMs: stats.mtimeMs,
+	});
+}
+
+function isNodeError(error: unknown): error is NodeJS.ErrnoException {
+	return error instanceof Error && "code" in error;
+}
+
+function toFileError(error: unknown, fallbackPath?: string): FileError {
+	if (error instanceof FileError) return error;
+	const cause = toError(error);
+	const nodeError = isNodeError(error) ? error : undefined;
+	const path = typeof nodeError?.path === "string" ? nodeError.path : fallbackPath;
+	if (nodeError) {
+		const message = nodeError.message;
+		switch (nodeError.code) {
+			case "ABORT_ERR":
+				return new FileError("aborted", message, path, cause);
+			case "ENOENT":
+				return new FileError("not_found", message, path, cause);
+			case "EACCES":
+			case "EPERM":
+				return new FileError("permission_denied", message, path, cause);
+			case "ENOTDIR":
+				return new FileError("not_directory", message, path, cause);
+			case "EISDIR":
+				return new FileError("is_directory", message, path, cause);
+			case "EINVAL":
+				return new FileError("invalid", message, path, cause);
+		}
+	}
+	return new FileError("unknown", cause.message, path, cause);
+}
+
+function abortResult<TValue>(signal: AbortSignal | undefined, path?: string): Result<TValue, FileError> | undefined {
+	return signal?.aborted ? err(new FileError("aborted", "aborted", path)) : undefined;
+}
+
+async function pathExists(path: string): Promise<boolean> {
+	try {
+		await access(path, constants.F_OK);
+		return true;
+	} catch {
+		return false;
+	}
+}
+
+async function runCommand(
+	command: string,
+	args: string[],
+	timeoutMs: number,
+): Promise<{ stdout: string; status: number | null }> {
+	return await new Promise((resolve) => {
+		let stdout = "";
+		let child: ReturnType<typeof spawn>;
+		try {
+			child = spawn(command, args, {
+				stdio: ["ignore", "pipe", "ignore"],
+				windowsHide: true,
+			});
+		} catch {
+			resolve({ stdout: "", status: null });
+			return;
+		}
+		const timeout = setTimeout(() => {
+			if (child.pid) killProcessTree(child.pid);
+		}, timeoutMs);
+		child.stdout?.setEncoding("utf8");
+		child.stdout?.on("data", (chunk: string) => {
+			stdout += chunk;
+		});
+		child.on("error", () => {
+			clearTimeout(timeout);
+			resolve({ stdout: "", status: null });
+		});
+		child.on("close", (status) => {
+			clearTimeout(timeout);
+			resolve({ stdout, status });
+		});
+	});
+}
+
+async function findBashOnPath(): Promise<string | null> {
+	const result =
+		process.platform === "win32"
+			? await runCommand("where", ["bash.exe"], 5000)
+			: await runCommand("which", ["bash"], 5000);
+	if (result.status !== 0 || !result.stdout) return null;
+	const firstMatch = result.stdout.trim().split(/\r?\n/)[0];
+	return firstMatch && (await pathExists(firstMatch)) ? firstMatch : null;
+}
+
+interface ShellConfig {
+	shell: string;
+	args: string[];
+	commandTransport?: "argv" | "stdin";
+}
+
+function isLegacyWslBashPath(path: string): boolean {
+	const normalized = path.replace(/\//g, "\\").toLowerCase();
+	return /^[a-z]:\\windows\\(?:system32|sysnative)\\bash\.exe$/.test(normalized);
+}
+
+function getBashShellConfig(shell: string): ShellConfig {
+	return isLegacyWslBashPath(shell) ? { shell, args: ["-s"], commandTransport: "stdin" } : { shell, args: ["-c"] };
+}
+
+async function getShellConfig(customShellPath?: string): Promise<Result<ShellConfig, ExecutionError>> {
+	if (customShellPath) {
+		if (await pathExists(customShellPath)) {
+			return ok(getBashShellConfig(customShellPath));
+		}
+		return err(new ExecutionError("shell_unavailable", `Custom shell path not found: ${customShellPath}`));
+	}
+	if (process.platform === "win32") {
+		const candidates: string[] = [];
+		const programFiles = process.env.ProgramFiles;
+		if (programFiles) candidates.push(`${programFiles}\\Git\\bin\\bash.exe`);
+		const programFilesX86 = process.env["ProgramFiles(x86)"];
+		if (programFilesX86) candidates.push(`${programFilesX86}\\Git\\bin\\bash.exe`);
+		for (const candidate of candidates) {
+			if (await pathExists(candidate)) {
+				return ok(getBashShellConfig(candidate));
+			}
+		}
+		const bashOnPath = await findBashOnPath();
+		if (bashOnPath) {
+			return ok(getBashShellConfig(bashOnPath));
+		}
+		return err(
+			new ExecutionError(
+				"shell_unavailable",
+				`No bash shell found. Options:\n` +
+					`  1. Install Git for Windows: https://git-scm.com/download/win\n` +
+					`  2. Add your bash to PATH (Cygwin, MSYS2, etc.)\n` +
+					"  3. Configure an explicit shellPath\n\n" +
+					`Searched Git Bash in:\n${candidates.map((path) => `  ${path}`).join("\n")}`,
+			),
+		);
+	}
+
+	if (await pathExists("/bin/bash")) {
+		return ok(getBashShellConfig("/bin/bash"));
+	}
+	const bashOnPath = await findBashOnPath();
+	if (bashOnPath) {
+		return ok(getBashShellConfig(bashOnPath));
+	}
+	return ok({ shell: "sh", args: ["-c"] });
+}
+
+function getShellEnv(
+	baseEnv?: NodeJS.ProcessEnv,
+	extraEnv?: Record<string, string>,
+	inheritEnv = true,
+): NodeJS.ProcessEnv {
+	if (!inheritEnv) return { ...extraEnv };
+	return {
+		...process.env,
+		...baseEnv,
+		...extraEnv,
+	};
+}
+
+function killProcessTree(pid: number): void {
+	if (process.platform === "win32") {
+		try {
+			const child = spawn(
+				join(process.env.SystemRoot ?? "C:\\Windows", "System32", "taskkill.exe"),
+				["/F", "/T", "/PID", String(pid)],
+				{
+					stdio: "ignore",
+					detached: true,
+					windowsHide: true,
+				},
+			);
+			// A failed spawn emits "error" asynchronously; consume it to avoid crashing Node.
+			child.once("error", () => {});
+		} catch {
+			// Ignore errors.
+		}
+		return;
+	}
+
+	try {
+		process.kill(-pid, "SIGKILL");
+	} catch {
+		try {
+			process.kill(pid, "SIGKILL");
+		} catch {
+			// Process already dead.
+		}
+	}
+}
+
+function waitForChildProcess(
+	child: ChildProcess,
+	spillIsDraining: () => boolean,
+): Promise<{ code: number | null; signal: NodeJS.Signals | null }> {
+	return new Promise((resolvePromise, reject) => {
+		let settled = false;
+		let exited = false;
+		let exitCode: number | null = null;
+		let exitSignal: NodeJS.Signals | null = null;
+		let postExitTimer: ReturnType<typeof setTimeout> | undefined;
+		let stdoutEnded = child.stdout === null;
+		let stderrEnded = child.stderr === null;
+
+		const cleanup = (): void => {
+			if (postExitTimer) clearTimeout(postExitTimer);
+			child.removeListener("error", onError);
+			child.removeListener("exit", onExit);
+			child.removeListener("close", onClose);
+			child.stdout?.removeListener("end", onStdoutEnd);
+			child.stderr?.removeListener("end", onStderrEnd);
+			child.stdout?.removeListener("data", onData);
+			child.stderr?.removeListener("data", onData);
+		};
+		const finalize = (): void => {
+			if (settled) return;
+			settled = true;
+			cleanup();
+			child.stdout?.destroy();
+			child.stderr?.destroy();
+			resolvePromise({ code: exitCode, signal: exitSignal });
+		};
+		const maybeFinalizeAfterExit = (): void => {
+			if (exited && stdoutEnded && stderrEnded) finalize();
+		};
+		const armIdleTimer = (): void => {
+			if (postExitTimer) clearTimeout(postExitTimer);
+			postExitTimer = setTimeout(() => {
+				if (spillIsDraining()) armIdleTimer();
+				else finalize();
+			}, EXIT_STDIO_GRACE_MS);
+		};
+		const onData = (): void => {
+			if (exited && !settled) armIdleTimer();
+		};
+		const onStdoutEnd = (): void => {
+			stdoutEnded = true;
+			maybeFinalizeAfterExit();
+		};
+		const onStderrEnd = (): void => {
+			stderrEnded = true;
+			maybeFinalizeAfterExit();
+		};
+		const onError = (error: Error): void => {
+			if (settled) return;
+			settled = true;
+			cleanup();
+			reject(error);
+		};
+		const onExit = (code: number | null, signal: NodeJS.Signals | null): void => {
+			exited = true;
+			exitCode = code;
+			exitSignal = signal;
+			maybeFinalizeAfterExit();
+			if (!settled) armIdleTimer();
+		};
+		const onClose = (code: number | null, signal: NodeJS.Signals | null): void => {
+			exitCode = code;
+			exitSignal = signal;
+			finalize();
+		};
+
+		child.stdout?.once("end", onStdoutEnd);
+		child.stderr?.once("end", onStderrEnd);
+		child.stdout?.on("data", onData);
+		child.stderr?.on("data", onData);
+		child.once("error", onError);
+		child.once("exit", onExit);
+		child.once("close", onClose);
+	});
+}
+
+/** Strict LF reader; Node readline does not report whether its final line was newline-terminated. */
+class NodeTextLineReader implements TextLineReader {
+	private readonly file: Awaited<ReturnType<typeof openFile>>;
+	private readonly path: string;
+	private readonly decoder = new TextDecoder();
+	private readonly chunk = new Uint8Array(64 * 1024);
+	private byteOffset = 0;
+	private buffered = "";
+	private ended = false;
+	private closed = false;
+
+	constructor(file: Awaited<ReturnType<typeof openFile>>, path: string) {
+		this.file = file;
+		this.path = path;
+	}
+
+	async readLine(context: Context): Promise<Result<TextLine | undefined, FileError>> {
+		const aborted = abortResult<TextLine | undefined>(context.abortSignal, this.path);
+		if (aborted) return aborted;
+		if (this.closed) return err(new FileError("invalid", "Text line reader is closed", this.path));
+
+		try {
+			while (true) {
+				const newline = this.buffered.indexOf("\n");
+				if (newline !== -1) {
+					const text = this.buffered.slice(0, newline);
+					this.buffered = this.buffered.slice(newline + 1);
+					return ok({ text, terminated: true });
+				}
+				if (this.ended) {
+					if (this.buffered.length === 0) return ok(undefined);
+					const text = this.buffered;
+					this.buffered = "";
+					return ok({ text, terminated: false });
+				}
+
+				// Explicit positions allow an aborted read to be retried without skipping bytes.
+				const { bytesRead } = await this.file.read(this.chunk, 0, this.chunk.length, this.byteOffset);
+				const afterReadAbort = abortResult<TextLine | undefined>(context.abortSignal, this.path);
+				if (afterReadAbort) return afterReadAbort;
+				this.byteOffset += bytesRead;
+				if (bytesRead === 0) {
+					this.buffered += this.decoder.decode();
+					this.ended = true;
+				} else {
+					this.buffered += this.decoder.decode(this.chunk.subarray(0, bytesRead), { stream: true });
+				}
+			}
+		} catch (error) {
+			return err(toFileError(error, this.path));
+		}
+	}
+
+	async close(_context: Context): Promise<void> {
+		if (this.closed) return;
+		this.closed = true;
+		this.buffered = "";
+		try {
+			await this.file.close();
+		} catch {
+			// Closing is best-effort, including after cancellation or an earlier I/O failure.
+		}
+	}
+}
+
+export class NodeExecutionEnv implements ExecutionEnv {
+	cwd: string;
+	private shellPath?: string;
+	private shellEnv?: NodeJS.ProcessEnv;
+	private activeChildPids = new Set<number>();
+
+	constructor(options: { cwd: string; shellPath?: string; shellEnv?: NodeJS.ProcessEnv }) {
+		this.cwd = options.cwd;
+		this.shellPath = options.shellPath;
+		this.shellEnv = options.shellEnv;
+	}
+
+	async absolutePath(path: string, _context: Context): Promise<Result<string, FileError>> {
+		return ok(resolvePath(this.cwd, path));
+	}
+
+	async joinPath(parts: string[], _context: Context): Promise<Result<string, FileError>> {
+		return ok(join(...parts));
+	}
+
+	async exec(
+		command: string,
+		options: ShellExecOptions | undefined,
+		context: Context,
+	): Promise<Result<ShellExecResult, ExecutionError>> {
+		const signal = context.abortSignal;
+		if (signal?.aborted) return err(new ExecutionError("aborted", "aborted"));
+		const timeoutMsResult = resolveTimeoutMs(options?.timeout);
+		if (!timeoutMsResult.ok) return err(timeoutMsResult.error);
+		const timeoutMs = timeoutMsResult.value;
+
+		const cwd = options?.cwd ? resolvePath(this.cwd, options.cwd) : this.cwd;
+		const shellConfig = await getShellConfig(this.shellPath);
+		if (!shellConfig.ok) return shellConfig;
+		try {
+			await access(cwd, constants.F_OK);
+		} catch (error) {
+			const cause = toError(error);
+			return err(
+				new ExecutionError(
+					"spawn_error",
+					`Working directory does not exist: ${cwd}\nCannot execute bash commands.`,
+					cause,
+				),
+			);
+		}
+
+		return await new Promise((resolvePromise) => {
+			let settled = false;
+			let timedOut = false;
+			let callbackError: ExecutionError | undefined;
+			let spillError: ExecutionError | undefined;
+			let child: ReturnType<typeof spawn> | undefined;
+			let timeoutId: ReturnType<typeof setTimeout> | undefined;
+			const spillPrefix: SpillChunk[] = [];
+			let spillPath: string | undefined;
+			const spillQueue: SpillChunk[] = [];
+			let spillStart: Promise<void> | undefined;
+			let spillStream: WriteStream | undefined;
+			let spillBackpressured = false;
+
+			const onAbort = () => {
+				if (child?.pid) killProcessTree(child.pid);
+			};
+			const failCallback = (error: unknown) => {
+				if (callbackError !== undefined) return;
+				const cause = toError(error);
+				callbackError = new ExecutionError("callback_error", cause.message, cause);
+				onAbort();
+			};
+			let capture: OutputCapture;
+			try {
+				capture = new OutputCapture(options?.capture, context, {
+					onUpdate: options?.onUpdate,
+					onError: failCallback,
+				});
+			} catch (error) {
+				const cause = toError(error);
+				resolvePromise(err(new ExecutionError("unknown", cause.message, cause)));
+				return;
+			}
+
+			const settle = (result: Result<ShellExecResult, ExecutionError>) => {
+				if (settled) return;
+				settled = true;
+				if (timeoutId) clearTimeout(timeoutId);
+				if (signal) signal.removeEventListener("abort", onAbort);
+				if (child?.pid) this.activeChildPids.delete(child.pid);
+				capture.dispose();
+				resolvePromise(result);
+			};
+			const pauseOutput = () => {
+				child?.stdout?.pause();
+				child?.stderr?.pause();
+			};
+			const resumeOutput = () => {
+				if (callbackError || spillError || timedOut || signal?.aborted || spillBackpressured) return;
+				child?.stdout?.resume();
+				child?.stderr?.resume();
+			};
+			const failSpill = (error: unknown) => {
+				if (spillError !== undefined) return;
+				const cause = toError(error);
+				spillError = new ExecutionError(
+					"unknown",
+					`Failed to preserve complete shell output: ${cause.message}`,
+					cause,
+				);
+				spillBackpressured = false;
+				onAbort();
+			};
+			const writeSpill = (chunk: SpillChunk): void => {
+				if (spillStream === undefined || chunk.length === 0) return;
+				if (spillStream.write(chunk) || spillBackpressured) return;
+				spillBackpressured = true;
+				pauseOutput();
+				spillStream.once("drain", () => {
+					spillBackpressured = false;
+					resumeOutput();
+				});
+			};
+			const startSpill = (chunk: SpillChunk): void => {
+				if (spillStream !== undefined) {
+					writeSpill(chunk);
+					return;
+				}
+				spillQueue.push(chunk);
+				if (spillStart !== undefined) return;
+				pauseOutput();
+				spillStart = (async () => {
+					const created = await this.createTempFile({ prefix: "pi-output-", suffix: ".log" }, context);
+					if (!created.ok) throw created.error;
+					spillPath = created.value;
+					capture.setSpillPath(spillPath);
+					spillStream = createWriteStream(spillPath, { flags: "a", highWaterMark: SPILL_HIGH_WATER_MARK });
+					spillStream.on("error", failSpill);
+					for (const queued of spillQueue) writeSpill(queued);
+					spillQueue.length = 0;
+				})()
+					.catch(failSpill)
+					.finally(resumeOutput);
+			};
+			const finishSpill = async (): Promise<void> => {
+				await spillStart;
+				const stream = spillStream;
+				if (stream === undefined || spillError !== undefined || stream.destroyed) return;
+				await new Promise<void>((resolveFinish) => {
+					stream.once("error", () => resolveFinish());
+					stream.once("finish", resolveFinish);
+					stream.end();
+				});
+			};
+
+			try {
+				const commandFromStdin = shellConfig.value.commandTransport === "stdin";
+				child = spawn(
+					shellConfig.value.shell,
+					commandFromStdin ? shellConfig.value.args : [...shellConfig.value.args, command],
+					{
+						cwd,
+						detached: process.platform !== "win32",
+						env: getShellEnv(this.shellEnv, options?.env, options?.inheritEnv),
+						stdio: [commandFromStdin ? "pipe" : "ignore", "pipe", "pipe"],
+						windowsHide: true,
+					},
+				);
+				if (child.pid) this.activeChildPids.add(child.pid);
+				if (commandFromStdin) {
+					child.stdin?.on("error", () => {});
+					child.stdin?.end(command);
+				}
+			} catch (error) {
+				const cause = toError(error);
+				settle(err(new ExecutionError("spawn_error", cause.message, cause)));
+				return;
+			}
+
+			timeoutId =
+				timeoutMs === undefined
+					? undefined
+					: setTimeout(() => {
+							timedOut = true;
+							onAbort();
+						}, timeoutMs);
+
+			if (signal) {
+				if (signal.aborted) onAbort();
+				else signal.addEventListener("abort", onAbort, { once: true });
+			}
+
+			const feed = (chunk: Uint8Array) => {
+				try {
+					const wasTruncated = capture.truncated;
+					capture.push(chunk);
+					if (!options?.capture?.spill || chunk.length === 0) return;
+					if (spillPath !== undefined || wasTruncated) {
+						startSpill(chunk);
+					} else if (capture.truncated) {
+						for (const prefix of spillPrefix) startSpill(prefix);
+						spillPrefix.length = 0;
+						startSpill(chunk);
+					} else {
+						spillPrefix.push(chunk);
+					}
+				} catch (error) {
+					failCallback(error);
+				}
+			};
+			child.stdout?.on("data", feed);
+			child.stderr?.on("data", feed);
+
+			void waitForChildProcess(
+				child,
+				() =>
+					spillError === undefined &&
+					spillStart !== undefined &&
+					(spillStream === undefined || spillBackpressured),
+			).then(
+				async ({ code, signal: exitSignal }) => {
+					await finishSpill();
+					try {
+						capture.finish();
+						capture.flush();
+					} catch (error) {
+						failCallback(error);
+					}
+					if (callbackError) {
+						settle(err(callbackError));
+						return;
+					}
+					if (timedOut) {
+						settle(err(new ExecutionError("timeout", `timeout:${options?.timeout}`)));
+						return;
+					}
+					if (signal?.aborted) {
+						settle(err(new ExecutionError("aborted", "aborted")));
+						return;
+					}
+					if (spillError) {
+						settle(err(spillError));
+						return;
+					}
+					const output = capture.snapshot();
+					// A process killed by a signal (e.g. OOM killer) has no exit code; map it
+					// to the conventional 128 + signal number so callers do not mistake it
+					// for a successful exit.
+					const exitCode = code ?? (exitSignal ? 128 + (osConstants.signals[exitSignal] ?? 0) : 1);
+					settle(
+						ok({
+							exitCode,
+							truncation: output.truncation,
+							...(output.spillPath === undefined ? {} : { spillPath: output.spillPath }),
+							...(output.lastLineBytes === undefined ? {} : { lastLineBytes: output.lastLineBytes }),
+						}),
+					);
+				},
+				(error: Error) => settle(err(new ExecutionError("spawn_error", error.message, error))),
+			);
+		});
+	}
+
+	async openTextLineReader(path: string, context: Context): Promise<Result<TextLineReader, FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const aborted = abortResult<TextLineReader>(context.abortSignal, resolved);
+		if (aborted) return aborted;
+		try {
+			const file = await openFile(resolved, "r");
+			const afterOpenAbort = abortResult<TextLineReader>(context.abortSignal, resolved);
+			if (afterOpenAbort) {
+				await file.close().catch(() => undefined);
+				return afterOpenAbort;
+			}
+			return ok(new NodeTextLineReader(file, resolved));
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		}
+	}
+
+	async readTextFile(path: string, context: Context): Promise<Result<string, FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const signal = context.abortSignal;
+		const aborted = abortResult<string>(signal, resolved);
+		if (aborted) return aborted;
+		try {
+			return ok(await readFile(resolved, { encoding: "utf8", signal }));
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		}
+	}
+
+	async readTextLines(
+		path: string,
+		options: { maxLines?: number } | undefined,
+		context: Context,
+	): Promise<Result<string[], FileError>> {
+		if (options?.maxLines !== undefined && options.maxLines <= 0) return ok([]);
+		const opened = await this.openTextLineReader(path, context);
+		if (!opened.ok) return opened;
+		const lines: string[] = [];
+		try {
+			while (options?.maxLines === undefined || lines.length < options.maxLines) {
+				const line = await opened.value.readLine(context);
+				if (!line.ok) return line;
+				if (line.value === undefined) break;
+				lines.push(line.value.text);
+			}
+			return ok(lines);
+		} finally {
+			await opened.value.close(context);
+		}
+	}
+
+	async readBinaryFile(path: string, context: Context): Promise<Result<Uint8Array, FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const signal = context.abortSignal;
+		const aborted = abortResult<Uint8Array>(signal, resolved);
+		if (aborted) return aborted;
+		try {
+			return ok(await readFile(resolved, { signal }));
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		}
+	}
+
+	async writeFile(path: string, content: string | Uint8Array, context: Context): Promise<Result<void, FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const signal = context.abortSignal;
+		const aborted = abortResult<void>(signal, resolved);
+		if (aborted) return aborted;
+		try {
+			await mkdir(resolve(resolved, ".."), { recursive: true });
+			const afterMkdirAbort = abortResult<void>(signal, resolved);
+			if (afterMkdirAbort) return afterMkdirAbort;
+			await writeFile(resolved, content, { signal });
+			return ok(undefined);
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		}
+	}
+
+	async appendFile(path: string, content: string | Uint8Array, context: Context): Promise<Result<void, FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const signal = context.abortSignal;
+		const aborted = abortResult<void>(signal, resolved);
+		if (aborted) return aborted;
+		try {
+			await mkdir(resolve(resolved, ".."), { recursive: true });
+			const afterMkdirAbort = abortResult<void>(signal, resolved);
+			if (afterMkdirAbort) return afterMkdirAbort;
+			await appendFile(resolved, content);
+			const afterAppendAbort = abortResult<void>(signal, resolved);
+			return afterAppendAbort ?? ok(undefined);
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		}
+	}
+
+	async truncateFile(path: string, size: number, context: Context): Promise<Result<void, FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const signal = context.abortSignal;
+		const aborted = abortResult<void>(signal, resolved);
+		if (aborted) return aborted;
+		if (!Number.isSafeInteger(size) || size < 0) {
+			return err(new FileError("invalid", "File size must be a non-negative safe integer", resolved));
+		}
+		let file: Awaited<ReturnType<typeof openFile>> | undefined;
+		try {
+			file = await openFile(resolved, "r+");
+			await file.truncate(size);
+			const afterTruncateAbort = abortResult<void>(signal, resolved);
+			return afterTruncateAbort ?? ok(undefined);
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		} finally {
+			await file?.close().catch(() => undefined);
+		}
+	}
+
+	async flushFile(path: string, context: Context): Promise<Result<void, FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const signal = context.abortSignal;
+		const aborted = abortResult<void>(signal, resolved);
+		if (aborted) return aborted;
+		let file: Awaited<ReturnType<typeof openFile>> | undefined;
+		try {
+			file = await openFile(resolved, "r+");
+			await file.sync();
+			const afterSyncAbort = abortResult<void>(signal, resolved);
+			return afterSyncAbort ?? ok(undefined);
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		} finally {
+			await file?.close().catch(() => undefined);
+		}
+	}
+
+	async renameFile(sourcePath: string, destinationPath: string, context: Context): Promise<Result<void, FileError>> {
+		const source = resolvePath(this.cwd, sourcePath);
+		const destination = resolvePath(this.cwd, destinationPath);
+		const aborted = abortResult<void>(context.abortSignal, destination);
+		if (aborted) return aborted;
+		try {
+			await rename(source, destination);
+			return ok(undefined);
+		} catch (error) {
+			return err(toFileError(error, source));
+		}
+	}
+
+	async fileInfo(path: string, context: Context): Promise<Result<FileInfo, FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const aborted = abortResult<FileInfo>(context.abortSignal, resolved);
+		if (aborted) return aborted;
+		try {
+			return fileInfoFromStats(resolved, await lstat(resolved));
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		}
+	}
+
+	async listDir(path: string, context: Context): Promise<Result<FileInfo[], FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const signal = context.abortSignal;
+		const aborted = abortResult<FileInfo[]>(signal, resolved);
+		if (aborted) return aborted;
+		try {
+			const entries = await readdir(resolved, { withFileTypes: true });
+			const infos: FileInfo[] = [];
+			for (const entry of entries) {
+				const loopAbort = abortResult<FileInfo[]>(signal, resolved);
+				if (loopAbort) return loopAbort;
+				const entryPath = resolve(resolved, entry.name);
+				try {
+					const info = fileInfoFromStats(entryPath, await lstat(entryPath));
+					if (info.ok) infos.push(info.value);
+				} catch (error) {
+					return err(toFileError(error, entryPath));
+				}
+			}
+			return ok(infos);
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		}
+	}
+
+	async canonicalPath(path: string, context: Context): Promise<Result<string, FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const aborted = abortResult<string>(context.abortSignal, resolved);
+		if (aborted) return aborted;
+		try {
+			return ok(await realpath(resolved));
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		}
+	}
+
+	async exists(path: string, context: Context): Promise<Result<boolean, FileError>> {
+		const result = await this.fileInfo(path, context);
+		if (result.ok) return ok(true);
+		if (result.error.code === "not_found") return ok(false);
+		return err(result.error);
+	}
+
+	async createDir(
+		path: string,
+		options: { recursive?: boolean } | undefined,
+		context: Context,
+	): Promise<Result<void, FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const aborted = abortResult<void>(context.abortSignal, resolved);
+		if (aborted) return aborted;
+		try {
+			await mkdir(resolved, { recursive: options?.recursive ?? true });
+			return ok(undefined);
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		}
+	}
+
+	async remove(
+		path: string,
+		options: { recursive?: boolean; force?: boolean } | undefined,
+		context: Context,
+	): Promise<Result<void, FileError>> {
+		const resolved = resolvePath(this.cwd, path);
+		const aborted = abortResult<void>(context.abortSignal, resolved);
+		if (aborted) return aborted;
+		try {
+			await rm(resolved, { recursive: options?.recursive ?? false, force: options?.force ?? false });
+			return ok(undefined);
+		} catch (error) {
+			return err(toFileError(error, resolved));
+		}
+	}
+
+	async createTempDir(prefix: string | undefined, context: Context): Promise<Result<string, FileError>> {
+		const aborted = abortResult<string>(context.abortSignal);
+		if (aborted) return aborted;
+		try {
+			prefix ??= "tmp-";
+			return ok(await mkdtemp(join(tmpdir(), prefix)));
+		} catch (error) {
+			return err(toFileError(error));
+		}
+	}
+
+	async createTempFile(
+		options: { prefix?: string; suffix?: string } | undefined,
+		context: Context,
+	): Promise<Result<string, FileError>> {
+		const dir = await this.createTempDir("tmp-", context);
+		if (!dir.ok) return dir;
+		const filePath = join(dir.value, `${options?.prefix ?? ""}${randomUUID()}${options?.suffix ?? ""}`);
+		try {
+			await writeFile(filePath, "");
+			return ok(filePath);
+		} catch (error) {
+			return err(toFileError(error, filePath));
+		}
+	}
+
+	async cleanup(_context: Context): Promise<void> {
+		for (const pid of this.activeChildPids) killProcessTree(pid);
+		this.activeChildPids.clear();
+	}
+}
--- a/packages/durable/src/env/utils/adaptive-publisher.ts
+++ b/packages/durable/src/env/utils/adaptive-publisher.ts
@ -0,0 +1,87 @@
+export interface AdaptivePublisherOptions<TValue, TUpdate> {
+	snapshot(): TValue;
+	update(previous: TValue | undefined, current: TValue): TUpdate | undefined;
+	measure(update: TUpdate): number;
+	publish(update: TUpdate): void;
+	onError(error: unknown): void;
+	minIntervalMs?: number;
+	targetBytesPerSecond?: number;
+}
+
+/**
+ * Publishes the latest state without queuing intermediate mutations.
+ *
+ * The first dirty state after idle is immediate. Each publication then buys a
+ * delay proportional to its encoded size, with a minimum interval that also
+ * bounds event count. A single trailing timer guarantees eventual publication.
+ */
+export class AdaptivePublisher<TValue, TUpdate> {
+	readonly #options: AdaptivePublisherOptions<TValue, TUpdate>;
+	readonly #minIntervalMs: number;
+	readonly #targetBytesPerSecond: number;
+	#published: TValue | undefined;
+	#dirty = false;
+	#nextEmitAt = 0;
+	#timer: ReturnType<typeof setTimeout> | undefined;
+	#disposed = false;
+
+	constructor(options: AdaptivePublisherOptions<TValue, TUpdate>) {
+		this.#options = options;
+		this.#minIntervalMs = options.minIntervalMs ?? 100;
+		this.#targetBytesPerSecond = options.targetBytesPerSecond ?? 100 * 1024;
+	}
+
+	markDirty(): void {
+		if (this.#disposed) return;
+		this.#dirty = true;
+		const wait = this.#nextEmitAt - Date.now();
+		if (wait <= 0) {
+			this.flush();
+			return;
+		}
+		this.#armTimer(wait);
+	}
+
+	flush(force = false): void {
+		if (this.#disposed || !this.#dirty) return;
+		const now = Date.now();
+		if (!force && now < this.#nextEmitAt) {
+			this.#armTimer(this.#nextEmitAt - now);
+			return;
+		}
+		if (this.#timer !== undefined) clearTimeout(this.#timer);
+		this.#timer = undefined;
+		const current = this.#options.snapshot();
+		const update = this.#options.update(this.#published, current);
+		if (update === undefined) {
+			this.#published = current;
+			this.#dirty = false;
+			return;
+		}
+		const encodedBytes = this.#options.measure(update);
+		this.#published = current;
+		this.#dirty = false;
+		this.#nextEmitAt = now + Math.max(this.#minIntervalMs, (encodedBytes * 1000) / this.#targetBytesPerSecond);
+		// Commit before delivery. A consumer may apply the update and then throw or
+		// reenter the producer; retaining the old baseline would duplicate that delta.
+		this.#options.publish(update);
+	}
+
+	dispose(): void {
+		if (this.#timer !== undefined) clearTimeout(this.#timer);
+		this.#timer = undefined;
+		this.#disposed = true;
+	}
+
+	#armTimer(wait: number): void {
+		if (this.#timer !== undefined) return;
+		this.#timer = setTimeout(() => {
+			this.#timer = undefined;
+			try {
+				this.flush();
+			} catch (error) {
+				this.#options.onError(error);
+			}
+		}, wait);
+	}
+}
--- a/packages/durable/src/env/utils/output-capture.ts
+++ b/packages/durable/src/env/utils/output-capture.ts
@ -0,0 +1,238 @@
+import type { Context } from "@earendil-works/chord";
+import type { ShellOutputCaptureOptions, ShellOutputMetadata, ShellOutputUpdate, ShellOutputView } from "../index.ts";
+import { AdaptivePublisher } from "./adaptive-publisher.ts";
+import { DEFAULT_MAX_BYTES, DEFAULT_MAX_LINES, truncateHead, truncateTail, utf8ByteLength } from "./truncate.ts";
+
+export const OUTPUT_MIN_EMIT_INTERVAL_MS = 100;
+export const OUTPUT_TARGET_BYTES_PER_SECOND = 100 * 1024;
+
+const INVALID_SHELL_OUTPUT = /[\x00-\x08\x0b-\x1f\ufff9-\ufffb]/g;
+const textEncoder = new TextEncoder();
+const textDecoder = new TextDecoder();
+
+interface OutputCaptureHandlers {
+	onUpdate?: (update: ShellOutputUpdate, context: Context) => void;
+	onError(error: unknown): void;
+}
+
+/**
+ * Maintains and publishes one bounded shell-output view.
+ *
+ * Writes received while publication is rate-limited collapse into the latest
+ * view. Small changes remain responsive; complete window turnovers purchase a
+ * proportionally longer delay. The first update after idle and an explicit
+ * final flush are immediate.
+ */
+export class OutputCapture {
+	readonly #maxBytes: number;
+	readonly #maxLines: number;
+	readonly #retain: "head" | "tail";
+	readonly #context: Context;
+	readonly #onUpdate: OutputCaptureHandlers["onUpdate"];
+
+	readonly #decoder = new TextDecoder();
+	#buffer = "";
+	#bufferBytes = 0;
+	#totalBytes = 0;
+	#newlines = 0;
+	#endsWithNewline = true;
+	#currentLineBytes = 0;
+	#spillPath: string | undefined;
+	#disposed = false;
+	readonly #publisher: AdaptivePublisher<ShellOutputView, ShellOutputUpdate>;
+
+	constructor(options: ShellOutputCaptureOptions | undefined, context: Context, handlers: OutputCaptureHandlers) {
+		this.#maxBytes = options?.limits.maxBytes ?? DEFAULT_MAX_BYTES;
+		this.#maxLines = options?.limits.maxLines ?? DEFAULT_MAX_LINES;
+		this.#retain = options?.limits.retain ?? "tail";
+		this.#context = context;
+		this.#onUpdate = handlers.onUpdate;
+		if (!Number.isFinite(this.#maxBytes) || this.#maxBytes <= 0) {
+			throw new TypeError("Output maxBytes must be a positive finite number");
+		}
+		if (!Number.isInteger(this.#maxLines) || this.#maxLines <= 0) {
+			throw new TypeError("Output maxLines must be a positive integer");
+		}
+		this.#publisher = new AdaptivePublisher({
+			snapshot: () => this.snapshot(),
+			update: updateFrom,
+			measure: (update) => utf8ByteLength(JSON.stringify(update)),
+			publish: (update) => this.#onUpdate?.(update, this.#context),
+			onError: handlers.onError,
+			minIntervalMs: OUTPUT_MIN_EMIT_INTERVAL_MS,
+			targetBytesPerSecond: OUTPUT_TARGET_BYTES_PER_SECOND,
+		});
+	}
+
+	get truncated(): boolean {
+		return this.#totalBytes > this.#maxBytes || this.#totalLines() > this.#maxLines;
+	}
+
+	push(chunk: string | Uint8Array): void {
+		if (this.#disposed) return;
+		if (typeof chunk === "string") {
+			this.#appendText(this.#decoder.decode());
+			this.#appendText(chunk);
+			return;
+		}
+		this.#appendText(this.#decoder.decode(chunk, { stream: true }));
+	}
+
+	finish(): void {
+		if (this.#disposed) return;
+		this.#appendText(this.#decoder.decode());
+	}
+
+	setSpillPath(path: string): void {
+		if (this.#disposed || this.#spillPath === path) return;
+		this.#spillPath = path;
+		this.#publisher.markDirty();
+		this.flush();
+	}
+
+	snapshot(): ShellOutputView {
+		const retained =
+			this.#retain === "head"
+				? truncateHead(this.#buffer, { maxBytes: this.#maxBytes, maxLines: this.#maxLines })
+				: truncateTail(this.#buffer, { maxBytes: this.#maxBytes, maxLines: this.#maxLines });
+		const totalLines = this.#totalLines();
+		const truncated = this.truncated;
+		const { content, ...truncation } = retained;
+		return {
+			text: sanitizeShellOutput(content),
+			truncation: {
+				...truncation,
+				truncated,
+				truncatedBy: truncated ? (totalLines > this.#maxLines ? "lines" : "bytes") : null,
+				totalBytes: this.#totalBytes,
+				totalLines,
+			},
+			...(this.#spillPath === undefined ? {} : { spillPath: this.#spillPath }),
+			...(retained.lastLinePartial ? { lastLineBytes: this.#currentLineBytes } : {}),
+		};
+	}
+
+	flush(): void {
+		if (this.#disposed) return;
+		this.#publisher.flush(true);
+	}
+
+	dispose(): void {
+		this.#publisher.dispose();
+		this.#disposed = true;
+	}
+
+	#appendText(text: string): void {
+		if (text === "") return;
+		const textBytes = utf8ByteLength(text);
+		this.#totalBytes += textBytes;
+		this.#newlines += countNewlines(text);
+		this.#endsWithNewline = text.endsWith("\n");
+		const lastNewline = text.lastIndexOf("\n");
+		this.#currentLineBytes =
+			lastNewline === -1 ? this.#currentLineBytes + textBytes : utf8ByteLength(text.slice(lastNewline + 1));
+		this.#buffer += text;
+		this.#bufferBytes += textBytes;
+
+		const guard = this.#maxBytes * 2;
+		if (this.#bufferBytes > guard * 2) {
+			this.#buffer =
+				this.#retain === "tail"
+					? trimToLastUtf8Bytes(this.#buffer, guard)
+					: trimToFirstUtf8Bytes(this.#buffer, guard);
+			this.#bufferBytes = utf8ByteLength(this.#buffer);
+		}
+		this.#publisher.markDirty();
+	}
+
+	#totalLines(): number {
+		return this.#newlines + (this.#endsWithNewline || this.#totalBytes === 0 ? 0 : 1);
+	}
+}
+
+export function applyShellOutputUpdate(
+	current: ShellOutputView | undefined,
+	update: ShellOutputUpdate,
+): ShellOutputView {
+	switch (update.kind) {
+		case "replace":
+			return update.output;
+		case "append":
+			return { text: `${current?.text ?? ""}${update.text}`, ...update.metadata };
+		case "slide":
+			return { text: `${current?.text.slice(update.drop) ?? ""}${update.text}`, ...update.metadata };
+		case "metadata":
+			return { text: current?.text ?? "", ...update.metadata };
+	}
+}
+
+function updateFrom(previous: ShellOutputView | undefined, current: ShellOutputView): ShellOutputUpdate {
+	if (previous === undefined) return { kind: "replace", output: current };
+	const metadata: ShellOutputMetadata = {
+		truncation: current.truncation,
+		...(current.spillPath === undefined ? {} : { spillPath: current.spillPath }),
+		...(current.lastLineBytes === undefined ? {} : { lastLineBytes: current.lastLineBytes }),
+	};
+	if (current.text === previous.text) return { kind: "metadata", metadata };
+	if (current.text.length > previous.text.length && current.text.slice(0, previous.text.length) === previous.text) {
+		return { kind: "append", text: current.text.slice(previous.text.length), metadata };
+	}
+	const shared = suffixPrefixOverlap(
+		previous.text,
+		current.text,
+		Math.min(previous.text.length, current.text.length, current.truncation.maxBytes * 2),
+	);
+	if (shared > 0) {
+		return {
+			kind: "slide",
+			drop: previous.text.length - shared,
+			text: current.text.slice(shared),
+			metadata,
+		};
+	}
+	return { kind: "replace", output: current };
+}
+
+function suffixPrefixOverlap(before: string, after: string, scan: number): number {
+	if (before.length === 0 || after.length === 0 || scan === 0) return 0;
+	const tail = before.length > scan ? before.slice(before.length - scan) : before;
+	for (const probeLength of [Math.min(64, after.length), 1]) {
+		const probe = after.slice(0, probeLength);
+		let candidates = 0;
+		for (let index = tail.indexOf(probe); index !== -1; index = tail.indexOf(probe, index + 1)) {
+			if (++candidates > 8) break;
+			const overlapLength = tail.length - index;
+			if (overlapLength <= after.length && tail.slice(index) === after.slice(0, overlapLength)) {
+				return overlapLength;
+			}
+		}
+		if (probeLength === 1) break;
+	}
+	return 0;
+}
+
+export function sanitizeShellOutput(text: string): string {
+	return text.replace(INVALID_SHELL_OUTPUT, "");
+}
+
+function countNewlines(text: string): number {
+	let count = 0;
+	for (let index = text.indexOf("\n"); index !== -1; index = text.indexOf("\n", index + 1)) count++;
+	return count;
+}
+
+function trimToLastUtf8Bytes(text: string, maxBytes: number): string {
+	const bytes = textEncoder.encode(text);
+	if (bytes.length <= maxBytes) return text;
+	let start = bytes.length - maxBytes;
+	while (start < bytes.length && ((bytes[start] ?? 0) & 0xc0) === 0x80) start++;
+	return textDecoder.decode(bytes.subarray(start));
+}
+
+function trimToFirstUtf8Bytes(text: string, maxBytes: number): string {
+	const bytes = textEncoder.encode(text);
+	if (bytes.length <= maxBytes) return text;
+	let end = maxBytes;
+	while (end > 0 && ((bytes[end] ?? 0) & 0xc0) === 0x80) end--;
+	return textDecoder.decode(bytes.subarray(0, end));
+}
--- a/packages/durable/src/env/utils/truncate.ts
+++ b/packages/durable/src/env/utils/truncate.ts
@ -0,0 +1,350 @@
+/**
+ * Shared truncation utilities for tool outputs.
+ *
+ * Truncation is based on two independent limits - whichever is hit first wins:
+ * - Line limit (default: 2000 lines)
+ * - Byte limit (default: 50KB)
+ *
+ * Never returns partial lines (except bash tail truncation edge case).
+ */
+
+export const DEFAULT_MAX_LINES = 2000;
+export const DEFAULT_MAX_BYTES = 50 * 1024; // 50KB
+export const GREP_MAX_LINE_LENGTH = 500; // Max chars per grep match line
+
+export interface TruncationResult {
+	/** The truncated content */
+	content: string;
+	/** Whether truncation occurred */
+	truncated: boolean;
+	/** Which limit was hit: "lines", "bytes", or null if not truncated */
+	truncatedBy: "lines" | "bytes" | null;
+	/** Total number of lines in the original content */
+	totalLines: number;
+	/** Total number of bytes in the original content */
+	totalBytes: number;
+	/** Number of complete lines in the truncated output */
+	outputLines: number;
+	/** Number of bytes in the truncated output */
+	outputBytes: number;
+	/** Whether the last line was partially truncated (only for tail truncation edge case) */
+	lastLinePartial: boolean;
+	/** Whether the first line exceeded the byte limit (for head truncation) */
+	firstLineExceedsLimit: boolean;
+	/** The max lines limit that was applied */
+	maxLines: number;
+	/** The max bytes limit that was applied */
+	maxBytes: number;
+}
+
+export interface TruncationOptions {
+	/** Maximum number of lines (default: 2000) */
+	maxLines?: number;
+	/** Maximum number of bytes (default: 50KB) */
+	maxBytes?: number;
+}
+
+interface RuntimeBuffer {
+	byteLength(content: string, encoding: "utf8"): number;
+}
+
+const runtimeBuffer = (globalThis as { Buffer?: RuntimeBuffer }).Buffer;
+const nonAsciiPattern = /[^\x00-\x7f]/;
+
+export function utf8ByteLength(content: string): number {
+	if (runtimeBuffer) return runtimeBuffer.byteLength(content, "utf8");
+
+	const firstNonAscii = content.search(nonAsciiPattern);
+	if (firstNonAscii === -1) return content.length;
+
+	let bytes = firstNonAscii;
+	for (let i = firstNonAscii; i < content.length; i++) {
+		const code = content.charCodeAt(i);
+		if (code <= 0x7f) {
+			bytes += 1;
+		} else if (code <= 0x7ff) {
+			bytes += 2;
+		} else if (code >= 0xd800 && code <= 0xdbff && i + 1 < content.length) {
+			const next = content.charCodeAt(i + 1);
+			if (next >= 0xdc00 && next <= 0xdfff) {
+				bytes += 4;
+				i++;
+			} else {
+				bytes += 3;
+			}
+		} else {
+			bytes += 3;
+		}
+	}
+	return bytes;
+}
+
+function splitLinesForCounting(content: string): string[] {
+	if (content.length === 0) return [];
+	const lines = content.split("\n");
+	if (content.endsWith("\n")) lines.pop();
+	return lines;
+}
+
+function replaceUnpairedSurrogates(content: string): string {
+	let output = "";
+	for (let i = 0; i < content.length; i++) {
+		const code = content.charCodeAt(i);
+		if (code >= 0xd800 && code <= 0xdbff) {
+			if (i + 1 < content.length) {
+				const next = content.charCodeAt(i + 1);
+				if (next >= 0xdc00 && next <= 0xdfff) {
+					output += content[i] + content[i + 1];
+					i++;
+					continue;
+				}
+			}
+			output += "<22>";
+		} else if (code >= 0xdc00 && code <= 0xdfff) {
+			output += "<22>";
+		} else {
+			output += content[i];
+		}
+	}
+	return output;
+}
+
+/**
+ * Format bytes as human-readable size.
+ */
+export function formatSize(bytes: number): string {
+	if (bytes < 1024) {
+		return `${bytes}B`;
+	} else if (bytes < 1024 * 1024) {
+		return `${(bytes / 1024).toFixed(1)}KB`;
+	} else {
+		return `${(bytes / (1024 * 1024)).toFixed(1)}MB`;
+	}
+}
+
+/**
+ * Truncate content from the head (keep first N lines/bytes).
+ * Suitable for file reads where you want to see the beginning.
+ *
+ * Never returns partial lines. If first line exceeds byte limit,
+ * returns empty content with firstLineExceedsLimit=true.
+ */
+export function truncateHead(content: string, options: TruncationOptions = {}): TruncationResult {
+	const maxLines = options.maxLines ?? DEFAULT_MAX_LINES;
+	const maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES;
+
+	const totalBytes = utf8ByteLength(content);
+	const lines = splitLinesForCounting(content);
+	const totalLines = lines.length;
+
+	// Check if no truncation needed
+	if (totalLines <= maxLines && totalBytes <= maxBytes) {
+		return {
+			content,
+			truncated: false,
+			truncatedBy: null,
+			totalLines,
+			totalBytes,
+			outputLines: totalLines,
+			outputBytes: totalBytes,
+			lastLinePartial: false,
+			firstLineExceedsLimit: false,
+			maxLines,
+			maxBytes,
+		};
+	}
+
+	// Check if first line alone exceeds byte limit
+	const firstLineBytes = utf8ByteLength(lines[0]);
+	if (firstLineBytes > maxBytes) {
+		return {
+			content: "",
+			truncated: true,
+			truncatedBy: "bytes",
+			totalLines,
+			totalBytes,
+			outputLines: 0,
+			outputBytes: 0,
+			lastLinePartial: false,
+			firstLineExceedsLimit: true,
+			maxLines,
+			maxBytes,
+		};
+	}
+
+	// Collect complete lines that fit
+	const outputLinesArr: string[] = [];
+	let outputBytesCount = 0;
+	let truncatedBy: "lines" | "bytes" = "lines";
+
+	for (let i = 0; i < lines.length && i < maxLines; i++) {
+		const line = lines[i];
+		const lineBytes = utf8ByteLength(line) + (i > 0 ? 1 : 0); // +1 for newline
+
+		if (outputBytesCount + lineBytes > maxBytes) {
+			truncatedBy = "bytes";
+			break;
+		}
+
+		outputLinesArr.push(line);
+		outputBytesCount += lineBytes;
+	}
+
+	// If we exited due to line limit
+	if (outputLinesArr.length >= maxLines && outputBytesCount <= maxBytes) {
+		truncatedBy = "lines";
+	}
+
+	const outputContent = outputLinesArr.join("\n");
+	const finalOutputBytes = utf8ByteLength(outputContent);
+
+	return {
+		content: outputContent,
+		truncated: true,
+		truncatedBy,
+		totalLines,
+		totalBytes,
+		outputLines: outputLinesArr.length,
+		outputBytes: finalOutputBytes,
+		lastLinePartial: false,
+		firstLineExceedsLimit: false,
+		maxLines,
+		maxBytes,
+	};
+}
+
+/**
+ * Truncate content from the tail (keep last N lines/bytes).
+ * Suitable for bash output where you want to see the end (errors, final results).
+ *
+ * May return partial first line if the last line of original content exceeds byte limit.
+ */
+export function truncateTail(content: string, options: TruncationOptions = {}): TruncationResult {
+	const maxLines = options.maxLines ?? DEFAULT_MAX_LINES;
+	const maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES;
+
+	const totalBytes = utf8ByteLength(content);
+	const lines = splitLinesForCounting(content);
+	const totalLines = lines.length;
+
+	// Check if no truncation needed
+	if (totalLines <= maxLines && totalBytes <= maxBytes) {
+		return {
+			content,
+			truncated: false,
+			truncatedBy: null,
+			totalLines,
+			totalBytes,
+			outputLines: totalLines,
+			outputBytes: totalBytes,
+			lastLinePartial: false,
+			firstLineExceedsLimit: false,
+			maxLines,
+			maxBytes,
+		};
+	}
+
+	// Work backwards from the end
+	const outputLinesArr: string[] = [];
+	let outputBytesCount = 0;
+	let truncatedBy: "lines" | "bytes" = "lines";
+	let lastLinePartial = false;
+
+	for (let i = lines.length - 1; i >= 0 && outputLinesArr.length < maxLines; i--) {
+		const line = lines[i];
+		const lineBytes = utf8ByteLength(line) + (outputLinesArr.length > 0 ? 1 : 0); // +1 for newline
+
+		if (outputBytesCount + lineBytes > maxBytes) {
+			truncatedBy = "bytes";
+			// Edge case: if we haven't added ANY lines yet and this line exceeds maxBytes,
+			// take the end of the line (partial)
+			if (outputLinesArr.length === 0) {
+				const truncatedLine = truncateStringToBytesFromEnd(line, maxBytes);
+				outputLinesArr.unshift(truncatedLine);
+				outputBytesCount = utf8ByteLength(truncatedLine);
+				lastLinePartial = true;
+			}
+			break;
+		}
+
+		outputLinesArr.unshift(line);
+		outputBytesCount += lineBytes;
+	}
+
+	// If we exited due to line limit
+	if (outputLinesArr.length >= maxLines && outputBytesCount <= maxBytes) {
+		truncatedBy = "lines";
+	}
+
+	const outputContent = outputLinesArr.join("\n");
+	const finalOutputBytes = utf8ByteLength(outputContent);
+
+	return {
+		content: outputContent,
+		truncated: true,
+		truncatedBy,
+		totalLines,
+		totalBytes,
+		outputLines: outputLinesArr.length,
+		outputBytes: finalOutputBytes,
+		lastLinePartial,
+		firstLineExceedsLimit: false,
+		maxLines,
+		maxBytes,
+	};
+}
+
+/**
+ * Truncate a string to fit within a byte limit (from the end).
+ * Handles multi-byte UTF-8 characters correctly.
+ */
+function truncateStringToBytesFromEnd(str: string, maxBytes: number): string {
+	if (maxBytes <= 0) return "";
+
+	let outputBytes = 0;
+	let start = str.length;
+	let needsReplacement = false;
+	for (let i = str.length; i > 0; ) {
+		let characterStart = i - 1;
+		const code = str.charCodeAt(characterStart);
+		let characterBytes: number;
+		let unpairedSurrogate = false;
+		if (code >= 0xdc00 && code <= 0xdfff && characterStart > 0) {
+			const previous = str.charCodeAt(characterStart - 1);
+			if (previous >= 0xd800 && previous <= 0xdbff) {
+				characterStart--;
+				characterBytes = 4;
+			} else {
+				characterBytes = 3;
+				unpairedSurrogate = true;
+			}
+		} else if (code >= 0xd800 && code <= 0xdfff) {
+			characterBytes = 3;
+			unpairedSurrogate = true;
+		} else {
+			characterBytes = code <= 0x7f ? 1 : code <= 0x7ff ? 2 : 3;
+		}
+		if (outputBytes + characterBytes > maxBytes) break;
+		outputBytes += characterBytes;
+		start = characterStart;
+		needsReplacement ||= unpairedSurrogate;
+		i = characterStart;
+	}
+
+	const output = str.slice(start);
+	return needsReplacement ? replaceUnpairedSurrogates(output) : output;
+}
+
+/**
+ * Truncate a single line to max characters, adding [truncated] suffix.
+ * Used for grep match lines.
+ */
+export function truncateLine(
+	line: string,
+	maxChars: number = GREP_MAX_LINE_LENGTH,
+): { text: string; wasTruncated: boolean } {
+	if (line.length <= maxChars) {
+		return { text: line, wasTruncated: false };
+	}
+	return { text: `${line.slice(0, maxChars)}... [truncated]`, wasTruncated: true };
+}
--- a/packages/durable/src/storage/jsonl/index.ts
+++ b/packages/durable/src/storage/jsonl/index.ts
@ -0,0 +1,6 @@
+export {
+	JsonlCorruptionError,
+	JsonlStorage,
+	type JsonlStorageOptions,
+	JsonlStoragePoisonedError,
+} from "./storage.ts";
--- a/packages/durable/src/storage/jsonl/node.ts
+++ b/packages/durable/src/storage/jsonl/node.ts
@ -0,0 +1,14 @@
+import type { Context } from "@earendil-works/chord";
+import { NodeExecutionEnv } from "../../env/node.ts";
+import { JsonlStorage, type JsonlStorageOptions } from "./storage.ts";
+
+/** Open or create a JSONL storage directory using the local Node filesystem. */
+export async function openNodeJsonlStorage(
+	directory: string,
+	context: Context,
+	options: JsonlStorageOptions = {},
+): Promise<JsonlStorage> {
+	return JsonlStorage.open(directory, new NodeExecutionEnv({ cwd: process.cwd() }), context, options);
+}
+
+export { JsonlStorage, type JsonlStorageOptions } from "./storage.ts";
--- a/packages/durable/src/storage/jsonl/storage.ts
+++ b/packages/durable/src/storage/jsonl/storage.ts
@ -0,0 +1,641 @@
+import type { Context, JsonValue } from "@earendil-works/chord";
+import type { FileError, FileSystem } from "../../env/index.ts";
+import type {
+	Cursor,
+	DocumentAddress,
+	DocumentContent,
+	DocumentCreate,
+	DocumentPoint,
+	DocumentQuery,
+	EntryQuery,
+	Id,
+	Seq,
+	Storage,
+	StorageWrite,
+	TaskQuery,
+	TaskRecord,
+} from "../../types.ts";
+import { MemoryStorage } from "../memory.ts";
+
+const FORMAT_VERSION = 1;
+const MAIN_FILE = "main.jsonl";
+const textDecoder = new TextDecoder("utf-8", { fatal: true });
+
+type StoredTask = TaskRecord<JsonValue, JsonValue, JsonValue>;
+
+type MainOperation =
+	| Extract<StorageWrite, { readonly type: "conversation" | "entry" | "submission" | "document.retire" }>
+	| { readonly type: "task"; readonly value: StoredTask }
+	| { readonly type: "task.sidecar"; readonly id: Id; readonly ordinal: number }
+	| { readonly type: "document.create"; readonly record: DocumentCreate; readonly ordinal: number }
+	| { readonly type: "document.change"; readonly id: Id; readonly ordinal: number };
+
+type MainMarker = {
+	readonly format: typeof FORMAT_VERSION;
+	readonly type: "commit";
+	readonly seq: Seq;
+	readonly writes: readonly MainOperation[];
+};
+
+type SidecarPayload =
+	| { readonly type: "task"; readonly value: StoredTask }
+	| { readonly type: "document"; readonly id: Id; readonly content: DocumentContent };
+
+type SidecarRecord = {
+	readonly format: typeof FORMAT_VERSION;
+	readonly type: "record";
+	readonly seq: Seq;
+	readonly ordinal: number;
+	readonly payload: SidecarPayload;
+};
+
+type ParsedLine<T> = {
+	readonly value: T;
+	readonly start: number;
+};
+
+type ParsedFile<T> = {
+	readonly path: string;
+	readonly lines: readonly ParsedLine<T>[];
+};
+
+type EncodedCommit = {
+	readonly marker: string;
+	readonly sidecars: ReadonlyMap<string, string>;
+};
+
+export type JsonlStorageOptions = {
+	/** Flush every affected sidecar before appending the main marker. Defaults to false. */
+	readonly fsync?: boolean;
+};
+
+export class JsonlCorruptionError extends Error {
+	constructor(message: string, cause?: Error) {
+		super(message, cause === undefined ? undefined : { cause });
+		this.name = "JsonlCorruptionError";
+	}
+}
+
+export class JsonlStoragePoisonedError extends Error {
+	constructor(cause: Error) {
+		super("JSONL storage is poisoned and must be reopened", { cause });
+		this.name = "JsonlStoragePoisonedError";
+	}
+}
+
+const isObject = (value: unknown): value is Record<string, unknown> =>
+	value !== null && typeof value === "object" && !Array.isArray(value);
+
+const isSafeInteger = (value: unknown): value is number => Number.isSafeInteger(value);
+
+const sidecarFileName = (kind: "doc" | "task", id: Id): string => `${kind}-${id}.jsonl`;
+
+const isSidecarFileName = (name: string): boolean => /^(?:doc|task)-(?:0|[1-9]\d*)\.jsonl$/.test(name);
+
+const sidecarKey = (file: string, seq: Seq, ordinal: number): string => JSON.stringify([file, seq, ordinal]);
+
+const jsonLine = (value: unknown): string => {
+	const encoded = JSON.stringify(value);
+	if (encoded === undefined) throw new TypeError("JSONL record is not serializable");
+	return `${encoded}\n`;
+};
+
+const errorFromFile = (action: string, error: FileError): Error =>
+	new Error(`JSONL ${action} failed: ${error.message}`, { cause: error });
+
+const parseJson = (text: string, description: string): unknown => {
+	try {
+		return JSON.parse(text) as unknown;
+	} catch (error) {
+		throw new JsonlCorruptionError(`Malformed complete ${description}`, error instanceof Error ? error : undefined);
+	}
+};
+
+const validateMainOperation = (value: unknown, description: string): MainOperation => {
+	if (!isObject(value) || typeof value.type !== "string") {
+		throw new JsonlCorruptionError(`Invalid write in ${description}`);
+	}
+	switch (value.type) {
+		case "conversation":
+		case "entry":
+		case "submission":
+			if (!isObject(value.value) || !isSafeInteger(value.value.id)) {
+				throw new JsonlCorruptionError(`Invalid ${value.type} write in ${description}`);
+			}
+			return value as MainOperation;
+		case "task":
+			if (
+				!isObject(value.value) ||
+				!isSafeInteger(value.value.id) ||
+				!isObject(value.value.state) ||
+				value.value.state.status !== "terminal"
+			) {
+				throw new JsonlCorruptionError(`Invalid terminal task write in ${description}`);
+			}
+			return value as MainOperation;
+		case "document.retire":
+			if (!isSafeInteger(value.id)) throw new JsonlCorruptionError(`Invalid document retirement in ${description}`);
+			return value as MainOperation;
+		case "task.sidecar":
+			if (!isSafeInteger(value.id) || !isSafeInteger(value.ordinal) || value.ordinal < 0) {
+				throw new JsonlCorruptionError(`Invalid task sidecar write in ${description}`);
+			}
+			return value as MainOperation;
+		case "document.create":
+			if (
+				!isObject(value.record) ||
+				!isSafeInteger(value.record.id) ||
+				!isSafeInteger(value.ordinal) ||
+				value.ordinal < 0
+			) {
+				throw new JsonlCorruptionError(`Invalid document creation in ${description}`);
+			}
+			return value as MainOperation;
+		case "document.change":
+			if (!isSafeInteger(value.id) || !isSafeInteger(value.ordinal) || value.ordinal < 0) {
+				throw new JsonlCorruptionError(`Invalid document change in ${description}`);
+			}
+			return value as MainOperation;
+		default:
+			throw new JsonlCorruptionError(`Unknown write type in ${description}`);
+	}
+};
+
+const parseMainMarker = (text: string, line: number): MainMarker => {
+	const description = `${MAIN_FILE} line ${line}`;
+	const value = parseJson(text, description);
+	if (
+		!isObject(value) ||
+		value.format !== FORMAT_VERSION ||
+		value.type !== "commit" ||
+		!isSafeInteger(value.seq) ||
+		value.seq < 1 ||
+		!Array.isArray(value.writes)
+	) {
+		throw new JsonlCorruptionError(`Invalid commit marker in ${description}`);
+	}
+	return {
+		format: FORMAT_VERSION,
+		type: "commit",
+		seq: value.seq,
+		writes: value.writes.map((write) => validateMainOperation(write, description)),
+	};
+};
+
+const validateDocumentContent = (value: unknown, description: string): DocumentContent => {
+	if (!isObject(value) || !isSafeInteger(value.version) || value.version < 1) {
+		throw new JsonlCorruptionError(`Invalid document content in ${description}`);
+	}
+	if (value.kind === "base" && isObject(value.value)) return value as DocumentContent;
+	if (value.kind === "delta" && Array.isArray(value.ops)) return value as DocumentContent;
+	throw new JsonlCorruptionError(`Invalid document content in ${description}`);
+};
+
+const parseSidecarRecord = (text: string, file: string, line: number): SidecarRecord => {
+	const description = `${file} line ${line}`;
+	const value = parseJson(text, description);
+	if (
+		!isObject(value) ||
+		value.format !== FORMAT_VERSION ||
+		value.type !== "record" ||
+		!isSafeInteger(value.seq) ||
+		value.seq < 1 ||
+		!isSafeInteger(value.ordinal) ||
+		value.ordinal < 0 ||
+		!isObject(value.payload) ||
+		typeof value.payload.type !== "string"
+	) {
+		throw new JsonlCorruptionError(`Invalid sidecar record in ${description}`);
+	}
+	if (value.payload.type === "task") {
+		if (
+			!isObject(value.payload.value) ||
+			!isSafeInteger(value.payload.value.id) ||
+			!isObject(value.payload.value.state) ||
+			(value.payload.value.state.status !== "pending" && value.payload.value.state.status !== "running")
+		) {
+			throw new JsonlCorruptionError(`Invalid live task record in ${description}`);
+		}
+	} else if (value.payload.type === "document") {
+		if (!isSafeInteger(value.payload.id)) {
+			throw new JsonlCorruptionError(`Invalid document record in ${description}`);
+		}
+		validateDocumentContent(value.payload.content, description);
+	} else {
+		throw new JsonlCorruptionError(`Unknown sidecar record type in ${description}`);
+	}
+	return value as SidecarRecord;
+};
+
+/** Portable JSONL implementation of the storage contract. */
+export class JsonlStorage implements Storage {
+	private readonly fs: FileSystem;
+	private readonly directory: string;
+	private readonly mainPath: string;
+	private readonly fsync: boolean;
+	private readonly memory: MemoryStorage;
+	private closed = false;
+	private poisonError: JsonlStoragePoisonedError | undefined;
+
+	private constructor(
+		fs: FileSystem,
+		directory: string,
+		mainPath: string,
+		memory: MemoryStorage,
+		options: JsonlStorageOptions,
+	) {
+		this.fs = fs;
+		this.directory = directory;
+		this.mainPath = mainPath;
+		this.memory = memory;
+		this.fsync = options.fsync ?? false;
+	}
+
+	/** Open or create a JSONL storage directory using the supplied filesystem. */
+	static async open(
+		directory: string,
+		fs: FileSystem,
+		context: Context,
+		options: JsonlStorageOptions = {},
+	): Promise<JsonlStorage> {
+		const absolute = await fs.absolutePath(directory, context);
+		if (!absolute.ok) throw errorFromFile("path resolution", absolute.error);
+		const created = await fs.createDir(absolute.value, { recursive: true }, context);
+		if (!created.ok) throw errorFromFile("directory creation", created.error);
+		const mainPathResult = await fs.joinPath([absolute.value, MAIN_FILE], context);
+		if (!mainPathResult.ok) throw errorFromFile("path join", mainPathResult.error);
+		const memory = new MemoryStorage();
+		await JsonlStorage.recover(fs, absolute.value, mainPathResult.value, memory, context);
+		return new JsonlStorage(fs, absolute.value, mainPathResult.value, memory, options);
+	}
+
+	async commit(writes: readonly StorageWrite[], context: Context): Promise<Seq> {
+		this.assertUsable();
+		const prepared = this.memory.prepareCommit(writes);
+		const encoded = this.encodeCommit(prepared.seq, prepared.writes);
+		const sidecars = await Promise.all(
+			[...encoded.sidecars].map(async ([file, content]) => ({
+				file,
+				content,
+				path: await this.resolveFile(file, context),
+			})),
+		);
+
+		for (const sidecar of sidecars) {
+			const result = await this.fs.appendFile(sidecar.path, sidecar.content, context);
+			if (!result.ok) throw this.poison(errorFromFile(`append to ${sidecar.file}`, result.error));
+		}
+		if (this.fsync) {
+			for (const sidecar of sidecars) {
+				const result = await this.fs.flushFile(sidecar.path, context);
+				if (!result.ok) throw this.poison(errorFromFile(`flush of ${sidecar.file}`, result.error));
+			}
+		}
+		const marker = await this.fs.appendFile(this.mainPath, encoded.marker, context);
+		if (!marker.ok) throw this.poison(errorFromFile(`append to ${MAIN_FILE}`, marker.error));
+		return prepared.apply();
+	}
+
+	async mintId() {
+		return this.store.mintId();
+	}
+
+	async conversation(id: Id, context: Context) {
+		return this.store.conversation(id, context);
+	}
+
+	async scanConversations(cursor: Cursor | undefined, limit: number, context: Context) {
+		return this.store.scanConversations(cursor, limit, context);
+	}
+
+	async entry(id: Id, context: Context) {
+		return this.store.entry(id, context);
+	}
+
+	async findLatestHeadMarker(conversationId: Id, atOrBeforeEntryId: Id | undefined, context: Context) {
+		return this.store.findLatestHeadMarker(conversationId, atOrBeforeEntryId, context);
+	}
+
+	async scanEntries(query: EntryQuery, cursor: Cursor | undefined, limit: number, context: Context) {
+		return this.store.scanEntries(query, cursor, limit, context);
+	}
+
+	async task(id: Id, context: Context) {
+		return this.store.task(id, context);
+	}
+
+	async scanTasks(query: TaskQuery, cursor: Cursor | undefined, limit: number, context: Context) {
+		return this.store.scanTasks(query, cursor, limit, context);
+	}
+
+	async submission(id: Id, context: Context) {
+		return this.store.submission(id, context);
+	}
+
+	async submissionByRequest(conversationId: Id, requestId: string, context: Context) {
+		return this.store.submissionByRequest(conversationId, requestId, context);
+	}
+
+	async findDocument(address: DocumentAddress, at: DocumentPoint, context: Context) {
+		return this.store.findDocument(address, at, context);
+	}
+
+	async document(id: Id, at: DocumentPoint, context: Context) {
+		return this.store.document(id, at, context);
+	}
+
+	async scanDocuments(query: DocumentQuery, cursor: Cursor | undefined, limit: number, context: Context) {
+		return this.store.scanDocuments(query, cursor, limit, context);
+	}
+
+	async close(context: Context): Promise<void> {
+		if (this.closed) return;
+		this.closed = true;
+		await this.memory.close(context);
+	}
+
+	private encodeCommit(seq: Seq, writes: readonly StorageWrite[]): EncodedCommit {
+		const mainWrites: MainOperation[] = [];
+		const records = new Map<string, SidecarRecord[]>();
+		let nextOrdinal = 0;
+		const addSidecar = (file: string, payload: SidecarPayload): number => {
+			const ordinal = nextOrdinal++;
+			let fileRecords = records.get(file);
+			if (fileRecords === undefined) {
+				fileRecords = [];
+				records.set(file, fileRecords);
+			}
+			fileRecords.push({ format: FORMAT_VERSION, type: "record", seq, ordinal, payload });
+			return ordinal;
+		};
+
+		for (const write of writes) {
+			switch (write.type) {
+				case "conversation":
+				case "entry":
+				case "submission":
+				case "document.retire":
+					mainWrites.push(write);
+					break;
+				case "task":
+					if (write.value.state.status === "terminal") mainWrites.push(write);
+					else {
+						const ordinal = addSidecar(sidecarFileName("task", write.value.id), {
+							type: "task",
+							value: write.value,
+						});
+						mainWrites.push({ type: "task.sidecar", id: write.value.id, ordinal });
+					}
+					break;
+				case "document.create": {
+					const ordinal = addSidecar(sidecarFileName("doc", write.record.id), {
+						type: "document",
+						id: write.record.id,
+						content: write.content,
+					});
+					mainWrites.push({ type: "document.create", record: write.record, ordinal });
+					break;
+				}
+				case "document.change": {
+					const ordinal = addSidecar(sidecarFileName("doc", write.id), {
+						type: "document",
+						id: write.id,
+						content: write.content,
+					});
+					mainWrites.push({ type: "document.change", id: write.id, ordinal });
+					break;
+				}
+			}
+		}
+
+		const sidecars = new Map<string, string>();
+		for (const [file, fileRecords] of records) {
+			sidecars.set(file, fileRecords.map((record) => jsonLine(record)).join(""));
+		}
+		const marker: MainMarker = {
+			format: FORMAT_VERSION,
+			type: "commit",
+			seq,
+			writes: mainWrites,
+		};
+		return { marker: jsonLine(marker), sidecars };
+	}
+
+	private static async recover(
+		fs: FileSystem,
+		directory: string,
+		mainPath: string,
+		memory: MemoryStorage,
+		context: Context,
+	): Promise<void> {
+		const main = await JsonlStorage.readLines(fs, mainPath, MAIN_FILE, context, (text, line) =>
+			parseMainMarker(text, line),
+		);
+		let previousSeq = 0;
+		for (const marker of main.lines) {
+			if (marker.value.seq <= previousSeq) {
+				throw new JsonlCorruptionError(`Commit sequence does not strictly increase in ${MAIN_FILE}`);
+			}
+			previousSeq = marker.value.seq;
+		}
+
+		const listed = await fs.listDir(directory, context);
+		if (!listed.ok) throw errorFromFile("directory listing", listed.error);
+		const sidecarFiles = listed.value
+			.filter((info) => info.kind === "file" && isSidecarFileName(info.name))
+			.map((info) => info.name)
+			.sort();
+		const parsedFiles = new Map<string, ParsedFile<SidecarRecord>>();
+		const recordByKey = new Map<string, ParsedLine<SidecarRecord>>();
+		for (const file of sidecarFiles) {
+			const pathResult = await fs.joinPath([directory, file], context);
+			if (!pathResult.ok) throw errorFromFile("path join", pathResult.error);
+			const parsed = await JsonlStorage.readLines(fs, pathResult.value, file, context, (text, line) =>
+				parseSidecarRecord(text, file, line),
+			);
+			parsedFiles.set(file, parsed);
+			let previous: SidecarRecord | undefined;
+			for (const line of parsed.lines) {
+				if (
+					previous !== undefined &&
+					(line.value.seq < previous.seq ||
+						(line.value.seq === previous.seq && line.value.ordinal <= previous.ordinal))
+				) {
+					throw new JsonlCorruptionError(`Sidecar records are out of order in ${file}`);
+				}
+				previous = line.value;
+				recordByKey.set(sidecarKey(file, line.value.seq, line.value.ordinal), line);
+			}
+		}
+
+		const confirmed = new Set<string>();
+		for (const line of main.lines) {
+			const marker = line.value;
+			const writes: StorageWrite[] = [];
+			for (const operation of marker.writes) {
+				switch (operation.type) {
+					case "conversation":
+					case "entry":
+					case "submission":
+					case "task":
+					case "document.retire":
+						writes.push(operation);
+						break;
+					case "task.sidecar": {
+						const record = JsonlStorage.confirmRecord(
+							marker,
+							operation.ordinal,
+							sidecarFileName("task", operation.id),
+							recordByKey,
+							confirmed,
+						);
+						if (record.payload.type !== "task" || record.payload.value.id !== operation.id) {
+							throw new JsonlCorruptionError(`Confirmed task sidecar data does not match commit ${marker.seq}`);
+						}
+						writes.push({ type: "task", value: record.payload.value });
+						break;
+					}
+					case "document.create": {
+						const record = JsonlStorage.confirmRecord(
+							marker,
+							operation.ordinal,
+							sidecarFileName("doc", operation.record.id),
+							recordByKey,
+							confirmed,
+						);
+						if (record.payload.type !== "document" || record.payload.id !== operation.record.id) {
+							throw new JsonlCorruptionError(
+								`Confirmed document sidecar data does not match commit ${marker.seq}`,
+							);
+						}
+						if (record.payload.content.kind !== "base") {
+							throw new JsonlCorruptionError(`Document creation lacks a confirmed base in commit ${marker.seq}`);
+						}
+						writes.push({ type: "document.create", record: operation.record, content: record.payload.content });
+						break;
+					}
+					case "document.change": {
+						const record = JsonlStorage.confirmRecord(
+							marker,
+							operation.ordinal,
+							sidecarFileName("doc", operation.id),
+							recordByKey,
+							confirmed,
+						);
+						if (record.payload.type !== "document" || record.payload.id !== operation.id) {
+							throw new JsonlCorruptionError(
+								`Confirmed document sidecar data does not match commit ${marker.seq}`,
+							);
+						}
+						writes.push({ type: "document.change", id: operation.id, content: record.payload.content });
+						break;
+					}
+				}
+			}
+			try {
+				memory.prepareCommit(writes, marker.seq).apply();
+			} catch (error) {
+				throw new JsonlCorruptionError(
+					`Invalid committed state at sequence ${marker.seq}`,
+					error instanceof Error ? error : undefined,
+				);
+			}
+		}
+
+		for (const [file, parsed] of parsedFiles) {
+			let unconfirmedAt: number | undefined;
+			for (const line of parsed.lines) {
+				const key = sidecarKey(file, line.value.seq, line.value.ordinal);
+				if (confirmed.has(key)) {
+					if (unconfirmedAt !== undefined) {
+						throw new JsonlCorruptionError(`Confirmed record follows an unconfirmed tail in ${file}`);
+					}
+				} else if (unconfirmedAt === undefined) {
+					unconfirmedAt = line.start;
+				}
+			}
+			if (unconfirmedAt !== undefined) {
+				const truncated = await fs.truncateFile(parsed.path, unconfirmedAt, context);
+				if (!truncated.ok) throw errorFromFile(`tail truncation of ${file}`, truncated.error);
+			}
+		}
+	}
+
+	private static confirmRecord(
+		marker: MainMarker,
+		ordinal: number,
+		file: string,
+		recordByKey: ReadonlyMap<string, ParsedLine<SidecarRecord>>,
+		confirmed: Set<string>,
+	): SidecarRecord {
+		const key = sidecarKey(file, marker.seq, ordinal);
+		if (confirmed.has(key)) throw new JsonlCorruptionError(`Sidecar record is confirmed more than once`);
+		const line = recordByKey.get(key);
+		if (line === undefined) {
+			throw new JsonlCorruptionError(`Missing confirmed sidecar record ${file} at sequence ${marker.seq}`);
+		}
+		confirmed.add(key);
+		return line.value;
+	}
+
+	private static async readLines<T>(
+		fs: FileSystem,
+		path: string,
+		name: string,
+		context: Context,
+		parse: (text: string, line: number) => T,
+	): Promise<ParsedFile<T>> {
+		const read = await fs.readBinaryFile(path, context);
+		if (!read.ok) {
+			if (read.error.code === "not_found") return { path, lines: [] };
+			throw errorFromFile(`read of ${name}`, read.error);
+		}
+		const bytes = read.value;
+		let completeSize = bytes.length;
+		if (completeSize > 0 && bytes[completeSize - 1] !== 0x0a) {
+			completeSize = bytes.lastIndexOf(0x0a) + 1;
+			const truncated = await fs.truncateFile(path, completeSize, context);
+			if (!truncated.ok) throw errorFromFile(`torn-line truncation of ${name}`, truncated.error);
+		}
+		const lines: ParsedLine<T>[] = [];
+		let start = 0;
+		let lineNumber = 1;
+		for (let end = 0; end < completeSize; end++) {
+			if (bytes[end] !== 0x0a) continue;
+			let text: string;
+			try {
+				text = textDecoder.decode(bytes.subarray(start, end));
+			} catch (error) {
+				throw new JsonlCorruptionError(
+					`Invalid UTF-8 in complete ${name} line ${lineNumber}`,
+					error instanceof Error ? error : undefined,
+				);
+			}
+			lines.push({ value: parse(text, lineNumber), start });
+			start = end + 1;
+			lineNumber++;
+		}
+		return { path, lines };
+	}
+
+	private async resolveFile(file: string, context: Context): Promise<string> {
+		const path = await this.fs.joinPath([this.directory, file], context);
+		if (!path.ok) throw errorFromFile("path join", path.error);
+		return path.value;
+	}
+
+	private get store(): MemoryStorage {
+		this.assertUsable();
+		return this.memory;
+	}
+
+	private poison(cause: Error): JsonlStoragePoisonedError {
+		this.poisonError ??= new JsonlStoragePoisonedError(cause);
+		return this.poisonError;
+	}
+
+	private assertUsable(): void {
+		if (this.closed) throw new Error("JsonlStorage is closed");
+		if (this.poisonError !== undefined) throw this.poisonError;
+	}
+}
--- a/packages/durable/src/storage/memory.ts
+++ b/packages/durable/src/storage/memory.ts
@ -81,6 +81,12 @@ const clone = <T>(value: T): T => {
 	return result as T;
 };

+const freeze = <T>(value: T): T => {
+	if (value === null || typeof value !== "object" || Object.isFrozen(value)) return value;
+	for (const child of Object.values(value)) freeze(child);
+	return Object.freeze(value);
+};
+
 const cursorId = (cursor: Readonly<Record<string, JsonValue>> | undefined): Id | undefined =>
 	cursor?.after as Id | undefined;

@ -160,6 +166,14 @@ const page = <T extends { readonly id: Id }>(values: readonly T[], limit: number
 	return { items: clone(items), next: { after: items.at(-1)!.id } };
 };

+/** A fully validated, detached state mutation whose application performs no fallible preparation. */
+export interface PreparedMemoryCommit {
+	readonly seq: Seq;
+	/** Deeply frozen detached writes for persistence. */
+	readonly writes: readonly StorageWrite[];
+	apply(): Seq;
+}
+
 /**
 * Detached in-memory reference implementation of `Storage`.
 *
@ -188,13 +202,38 @@ export class MemoryStorage implements Storage {
 	private closed = false;

 	async commit(writes: readonly StorageWrite[], _context: Context): Promise<Seq> {
-		this.assertOpen();
-		const prepared = writes.map((write) => clone(write));
-		const seq = this.nextSeq;
-		this.checkGlobalIds(prepared);
-		const documentActions = this.prepareDocumentActions(prepared);
-		this.checkDocumentActions(documentActions);
+		return this.prepareCommit(writes).apply();
+	}

+	/** Validate and detach one commit without changing observable state. */
+	prepareCommit(writes: readonly StorageWrite[], seq: Seq = this.nextSeq): PreparedMemoryCommit {
+		this.assertOpen();
+		if (!Number.isSafeInteger(seq) || seq < this.nextSeq) {
+			throw new Error(`Commit sequence ${seq} does not strictly increase`);
+		}
+		const detachedWrites = freeze(writes.map((write) => clone(write)));
+		this.checkGlobalIds(detachedWrites);
+		const documentActions = this.prepareDocumentActions(detachedWrites);
+		this.checkDocumentActions(documentActions);
+		let applied = false;
+		return {
+			seq,
+			writes: detachedWrites,
+			apply: () => {
+				if (!applied) {
+					applied = true;
+					this.applyPreparedCommit(detachedWrites, documentActions, seq);
+				}
+				return seq;
+			},
+		};
+	}
+
+	private applyPreparedCommit(
+		prepared: readonly StorageWrite[],
+		documentActions: ReadonlyMap<Id, DocumentAction>,
+		seq: Seq,
+	): Seq {
 		for (const write of prepared) {
 			switch (write.type) {
 				case "conversation":
@ -264,7 +303,7 @@ export class MemoryStorage implements Storage {
 		}

 		this.applyDocumentActions(documentActions, seq);
-		this.nextSeq++;
+		this.nextSeq = seq + 1;
 		return seq;
 	}

--- a/packages/durable/test/jsonl-storage.test.ts
+++ b/packages/durable/test/jsonl-storage.test.ts
@ -0,0 +1,651 @@
+import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { basename, join } from "node:path";
+import type { Context, JsonValue } from "@earendil-works/chord";
+import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
+import { afterEach, describe, expect, it, onTestFinished } from "vitest";
+import { err, FileError, type FileSystem, type Result } from "../src/env/index.ts";
+import { NodeExecutionEnv } from "../src/env/node.ts";
+import { JsonlStorage } from "../src/storage/jsonl/index.ts";
+import { openNodeJsonlStorage } from "../src/storage/jsonl/node.ts";
+import type { DocumentCreate, Seq, Storage, StorageWrite, TaskRecord } from "../src/types.ts";
+import { ROOT_CONVERSATION_ID } from "../src/types.ts";
+import { registerStorageConformance } from "./storage-conformance.ts";
+
+const context = BACKGROUND_CONTEXT;
+type StoredTask = TaskRecord<JsonValue, JsonValue, JsonValue>;
+const tempDirectories = new Set<string>();
+const openStorages = new Set<JsonlStorage>();
+
+afterEach(async () => {
+	for (const storage of openStorages) await storage.close(context);
+	openStorages.clear();
+	for (const directory of tempDirectories) await rm(directory, { recursive: true, force: true });
+	tempDirectories.clear();
+});
+
+async function tempDirectory(prefix = "pi-durable-jsonl-"): Promise<string> {
+	const directory = await mkdtemp(join(tmpdir(), prefix));
+	tempDirectories.add(directory);
+	return directory;
+}
+
+async function openStorage(
+	directory: string,
+	fs: FileSystem = new NodeExecutionEnv({ cwd: directory }),
+	options: { readonly fsync?: boolean } = {},
+): Promise<JsonlStorage> {
+	const storage = await JsonlStorage.open(directory, fs, context, options);
+	openStorages.add(storage);
+	return storage;
+}
+
+async function createStorage(): Promise<JsonlStorage> {
+	return openStorage(await tempDirectory());
+}
+
+class ReopeningStorage implements Storage {
+	private current: JsonlStorage;
+	private readonly directory: string;
+	private closed = false;
+
+	constructor(current: JsonlStorage, directory: string) {
+		this.current = current;
+		this.directory = directory;
+	}
+
+	async commit(writes: readonly StorageWrite[], commitContext: Context): Promise<Seq> {
+		if (this.closed) return this.current.commit(writes, commitContext);
+		try {
+			return await this.current.commit(writes, commitContext);
+		} finally {
+			await this.current.close(context);
+			this.current = await JsonlStorage.open(this.directory, new NodeExecutionEnv({ cwd: this.directory }), context);
+		}
+	}
+
+	mintId: Storage["mintId"] = () => this.current.mintId();
+	conversation: Storage["conversation"] = (id, readContext) => this.current.conversation(id, readContext);
+	scanConversations: Storage["scanConversations"] = (cursor, limit, readContext) =>
+		this.current.scanConversations(cursor, limit, readContext);
+	entry: Storage["entry"] = (id, readContext) => this.current.entry(id, readContext);
+	findLatestHeadMarker: Storage["findLatestHeadMarker"] = (conversationId, at, readContext) =>
+		this.current.findLatestHeadMarker(conversationId, at, readContext);
+	scanEntries: Storage["scanEntries"] = (query, cursor, limit, readContext) =>
+		this.current.scanEntries(query, cursor, limit, readContext);
+	task: Storage["task"] = (id, readContext) => this.current.task(id, readContext);
+	scanTasks: Storage["scanTasks"] = (query, cursor, limit, readContext) =>
+		this.current.scanTasks(query, cursor, limit, readContext);
+	submission: Storage["submission"] = (id, readContext) => this.current.submission(id, readContext);
+	submissionByRequest: Storage["submissionByRequest"] = (conversationId, requestId, readContext) =>
+		this.current.submissionByRequest(conversationId, requestId, readContext);
+	findDocument: Storage["findDocument"] = (address, at, readContext) =>
+		this.current.findDocument(address, at, readContext);
+	document: Storage["document"] = (id, at, readContext) => this.current.document(id, at, readContext);
+	scanDocuments: Storage["scanDocuments"] = (query, cursor, limit, readContext) =>
+		this.current.scanDocuments(query, cursor, limit, readContext);
+
+	async close(closeContext: Context): Promise<void> {
+		if (this.closed) return;
+		this.closed = true;
+		await this.current.close(closeContext);
+	}
+}
+
+registerStorageConformance("Pico JsonlStorage conformance", createStorage);
+
+registerStorageConformance("Pico JsonlStorage conformance across reopen", async () => {
+	const directory = await tempDirectory("pi-durable-jsonl-conformance-");
+	const current = await JsonlStorage.open(directory, new NodeExecutionEnv({ cwd: directory }), context);
+	const storage = new ReopeningStorage(current, directory);
+	onTestFinished(() => storage.close(context));
+	return storage;
+});
+
+function pendingTask(id: number, phase = "ready"): StoredTask {
+	return {
+		id,
+		conversationId: ROOT_CONVERSATION_ID,
+		kind: "test.task",
+		version: 1,
+		input: null,
+		state: { status: "pending", checkpoint: { phase } },
+		after: [],
+		background: false,
+		abortRequested: false,
+	};
+}
+
+function terminalTask(id: number): StoredTask {
+	return {
+		id,
+		conversationId: ROOT_CONVERSATION_ID,
+		kind: "test.task",
+		version: 1,
+		input: null,
+		state: { status: "terminal", outcome: { status: "completed", result: null } },
+		after: [],
+		background: false,
+		abortRequested: false,
+	};
+}
+
+async function createRoot(storage: Storage): Promise<void> {
+	await storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context);
+}
+
+function sessionDocument(id: number, kind = "test.document"): DocumentCreate {
+	return { id, kind, scope: { kind: "session" } };
+}
+
+type Failure = {
+	readonly operation: "append" | "flush";
+	readonly call: number;
+	readonly mode: "before" | "after" | "short";
+};
+
+class InstrumentedEnv extends NodeExecutionEnv {
+	readonly operations: string[] = [];
+	private failure: Failure | undefined;
+	private appendCalls = 0;
+	private flushCalls = 0;
+
+	fail(failure: Failure): void {
+		this.failure = failure;
+		this.appendCalls = 0;
+		this.flushCalls = 0;
+		this.operations.length = 0;
+	}
+
+	clear(): void {
+		this.failure = undefined;
+		this.appendCalls = 0;
+		this.flushCalls = 0;
+		this.operations.length = 0;
+	}
+
+	override async appendFile(
+		path: string,
+		content: string | Uint8Array,
+		appendContext: Context,
+	): Promise<Result<void, FileError>> {
+		this.appendCalls++;
+		this.operations.push(`append:${basename(path)}`);
+		const failure = this.failure;
+		if (failure?.operation !== "append" || failure.call !== this.appendCalls) {
+			return super.appendFile(path, content, appendContext);
+		}
+		if (failure.mode === "before") return err(new FileError("unknown", "injected append failure", path));
+		if (failure.mode === "short") {
+			const bytes = typeof content === "string" ? new TextEncoder().encode(content) : content;
+			const partial = bytes.subarray(0, Math.max(1, Math.floor(bytes.length / 2)));
+			const written = await super.appendFile(path, partial, appendContext);
+			if (!written.ok) return written;
+			return err(new FileError("unknown", "injected short append", path));
+		}
+		const written = await super.appendFile(path, content, appendContext);
+		if (!written.ok) return written;
+		return err(new FileError("unknown", "injected post-append failure", path));
+	}
+
+	override async flushFile(path: string, flushContext: Context): Promise<Result<void, FileError>> {
+		this.flushCalls++;
+		this.operations.push(`flush:${basename(path)}`);
+		const failure = this.failure;
+		if (failure?.operation !== "flush" || failure.call !== this.flushCalls) {
+			return super.flushFile(path, flushContext);
+		}
+		if (failure.mode === "after") {
+			const flushed = await super.flushFile(path, flushContext);
+			if (!flushed.ok) return flushed;
+		}
+		return err(new FileError("unknown", "injected flush failure", path));
+	}
+}
+
+async function readLines(path: string): Promise<string[]> {
+	const text = await readFile(path, "utf8");
+	return text === "" ? [] : text.trimEnd().split("\n");
+}
+
+describe("Pico JsonlStorage publication and recovery", () => {
+	it("opens through the Node adapter", async () => {
+		const storage = await openNodeJsonlStorage(await tempDirectory(), context);
+		openStorages.add(storage);
+		await createRoot(storage);
+		expect(await storage.conversation(ROOT_CONVERSATION_ID, context)).toEqual({ id: ROOT_CONVERSATION_ID });
+	});
+
+	it("persists live tasks and document revisions in sidecars with one marker for every commit", async () => {
+		const directory = await tempDirectory();
+		const storage = await openStorage(directory);
+		await createRoot(storage);
+		const taskId = await storage.mintId();
+		const documentId = await storage.mintId();
+		await storage.commit([{ type: "task", value: pendingTask(taskId) }], context);
+		await storage.commit(
+			[
+				{
+					type: "document.create",
+					record: sessionDocument(documentId),
+					content: { kind: "base", version: 1, value: { count: 0 } },
+				},
+			],
+			context,
+		);
+		await storage.commit(
+			[{ type: "document.change", id: documentId, content: { kind: "delta", version: 1, ops: [] } }],
+			context,
+		);
+		await storage.commit(
+			[
+				{
+					type: "document.change",
+					id: documentId,
+					content: { kind: "base", version: 1, value: { count: 1 } },
+				},
+			],
+			context,
+		);
+		await storage.commit([{ type: "document.retire", id: documentId }], context);
+		await storage.commit([{ type: "task", value: terminalTask(taskId) }], context);
+
+		expect(await readLines(join(directory, "main.jsonl"))).toHaveLength(7);
+		expect(await readLines(join(directory, `task-${taskId}.jsonl`))).toHaveLength(1);
+		expect(await readLines(join(directory, `doc-${documentId}.jsonl`))).toHaveLength(3);
+		const markerTypes = (await readLines(join(directory, "main.jsonl"))).map(
+			(line) => (JSON.parse(line) as { readonly type: string }).type,
+		);
+		expect(markerTypes).toEqual(["commit", "commit", "commit", "commit", "commit", "commit", "commit"]);
+	});
+
+	it("orders multiple live-task replacements in one sidecar by commit ordinal", async () => {
+		const directory = await tempDirectory();
+		const storage = await openStorage(directory);
+		await createRoot(storage);
+		const taskId = await storage.mintId();
+		await storage.commit(
+			[
+				{ type: "task", value: pendingTask(taskId, "first") },
+				{ type: "task", value: pendingTask(taskId, "second") },
+			],
+			context,
+		);
+		expect(await readLines(join(directory, `task-${taskId}.jsonl`))).toHaveLength(2);
+
+		const reopened = await openStorage(directory);
+		expect(await reopened.task(taskId, context)).toEqual(pendingTask(taskId, "second"));
+	});
+
+	it("removes a complete-plus-torn unconfirmed multi-record sidecar append", async () => {
+		const directory = await tempDirectory();
+		const env = new InstrumentedEnv({ cwd: directory });
+		const storage = await openStorage(directory, env);
+		await createRoot(storage);
+		const taskId = await storage.mintId();
+		env.fail({ operation: "append", call: 1, mode: "short" });
+		await expect(
+			storage.commit(
+				[
+					{ type: "task", value: pendingTask(taskId, "first") },
+					{ type: "task", value: pendingTask(taskId, `second-${"x".repeat(512)}`) },
+				],
+				context,
+			),
+		).rejects.toThrow("poisoned");
+		const partial = await readFile(join(directory, `task-${taskId}.jsonl`));
+		expect(partial.at(-1)).not.toBe(0x0a);
+		expect([...partial].filter((byte) => byte === 0x0a)).toHaveLength(1);
+
+		const reopened = await openStorage(directory);
+		expect(await reopened.task(taskId, context)).toBeUndefined();
+		expect((await stat(join(directory, `task-${taskId}.jsonl`))).size).toBe(0);
+	});
+
+	it("serializes the complete candidate before I/O and leaves preparation failures usable", async () => {
+		const directory = await tempDirectory();
+		const env = new InstrumentedEnv({ cwd: directory });
+		const storage = await openStorage(directory, env);
+		await createRoot(storage);
+		env.clear();
+		const id = await storage.mintId();
+		await expect(
+			storage.commit(
+				[
+					{
+						type: "entry",
+						value: {
+							id,
+							conversationId: ROOT_CONVERSATION_ID,
+							kind: "bad",
+							data: 1n as unknown as JsonValue,
+						},
+					},
+				],
+				context,
+			),
+		).rejects.toThrow();
+		expect(env.operations).toEqual([]);
+		expect(await storage.entry(id, context)).toBeUndefined();
+		expect(
+			await storage.commit(
+				[{ type: "entry", value: { id, conversationId: ROOT_CONVERSATION_ID, kind: "good" } }],
+				context,
+			),
+		).toBe(2);
+	});
+
+	for (const failure of [
+		{ operation: "append", call: 1, mode: "before" },
+		{ operation: "append", call: 1, mode: "after" },
+		{ operation: "append", call: 1, mode: "short" },
+		{ operation: "append", call: 2, mode: "before" },
+		{ operation: "append", call: 2, mode: "after" },
+		{ operation: "append", call: 3, mode: "before" },
+		{ operation: "append", call: 3, mode: "short" },
+		{ operation: "append", call: 3, mode: "after" },
+	] as const satisfies readonly Failure[]) {
+		it(`poisons after ${failure.mode} failure at append ${failure.call} and recovers only confirmed state`, async () => {
+			const directory = await tempDirectory();
+			const env = new InstrumentedEnv({ cwd: directory });
+			const storage = await openStorage(directory, env);
+			await createRoot(storage);
+			const firstId = await storage.mintId();
+			const secondId = await storage.mintId();
+			env.fail(failure);
+			await expect(
+				storage.commit(
+					[
+						{
+							type: "document.create",
+							record: sessionDocument(firstId, "first"),
+							content: { kind: "base", version: 1, value: { text: "α" } },
+						},
+						{
+							type: "document.create",
+							record: sessionDocument(secondId, "second"),
+							content: { kind: "base", version: 1, value: { text: "β" } },
+						},
+					],
+					context,
+				),
+			).rejects.toThrow("poisoned");
+			await expect(storage.document(firstId, "current", context)).rejects.toThrow("poisoned");
+
+			const reopened = await openStorage(directory);
+			const markerSurvived = failure.call === 3 && failure.mode === "after";
+			expect(await reopened.document(firstId, "current", context)).toEqual(
+				markerSurvived ? expect.objectContaining({ value: { text: "α" } }) : undefined,
+			);
+			expect(await reopened.document(secondId, "current", context)).toEqual(
+				markerSurvived ? expect.objectContaining({ value: { text: "β" } }) : undefined,
+			);
+		});
+	}
+
+	for (const failure of [
+		{ operation: "flush", call: 1, mode: "before" },
+		{ operation: "flush", call: 1, mode: "after" },
+		{ operation: "flush", call: 2, mode: "before" },
+		{ operation: "flush", call: 2, mode: "after" },
+	] as const satisfies readonly Failure[]) {
+		it(`poisons after ${failure.mode} failure at flush ${failure.call} and never writes a marker`, async () => {
+			const directory = await tempDirectory();
+			const env = new InstrumentedEnv({ cwd: directory });
+			const storage = await openStorage(directory, env, { fsync: true });
+			await createRoot(storage);
+			const firstId = await storage.mintId();
+			const secondId = await storage.mintId();
+			env.fail(failure);
+			await expect(
+				storage.commit(
+					[
+						{
+							type: "document.create",
+							record: sessionDocument(firstId, "flush.first"),
+							content: { kind: "base", version: 1, value: {} },
+						},
+						{
+							type: "document.create",
+							record: sessionDocument(secondId, "flush.second"),
+							content: { kind: "base", version: 1, value: {} },
+						},
+					],
+					context,
+				),
+			).rejects.toThrow("poisoned");
+			const reopened = await openStorage(directory);
+			expect(await reopened.document(firstId, "current", context)).toBeUndefined();
+			expect(await reopened.document(secondId, "current", context)).toBeUndefined();
+		});
+	}
+
+	it("orders appends and optional flushes exactly and never flushes main.jsonl", async () => {
+		for (const fsync of [false, true]) {
+			const directory = await tempDirectory();
+			const env = new InstrumentedEnv({ cwd: directory });
+			const storage = await openStorage(directory, env, { fsync });
+			await createRoot(storage);
+			const firstId = await storage.mintId();
+			const secondId = await storage.mintId();
+			env.clear();
+			await storage.commit(
+				[
+					{
+						type: "document.create",
+						record: sessionDocument(secondId, "second"),
+						content: { kind: "base", version: 1, value: {} },
+					},
+					{
+						type: "document.create",
+						record: sessionDocument(firstId, "first"),
+						content: { kind: "base", version: 1, value: {} },
+					},
+				],
+				context,
+			);
+			const expected = [
+				`append:doc-${secondId}.jsonl`,
+				`append:doc-${firstId}.jsonl`,
+				...(fsync ? [`flush:doc-${secondId}.jsonl`, `flush:doc-${firstId}.jsonl`] : []),
+				"append:main.jsonl",
+			];
+			expect(env.operations).toEqual(expected);
+
+			env.clear();
+			await storage.commit(
+				[
+					{
+						type: "entry",
+						value: {
+							id: await storage.mintId(),
+							conversationId: ROOT_CONVERSATION_ID,
+							kind: "main-only",
+						},
+					},
+				],
+				context,
+			);
+			expect(env.operations).toEqual(["append:main.jsonl"]);
+		}
+	});
+
+	it("truncates torn UTF-8 tails at exact byte offsets and reuses the unconfirmed sequence", async () => {
+		const directory = await tempDirectory();
+		const storage = await openStorage(directory);
+		await createRoot(storage);
+		const documentId = await storage.mintId();
+		await storage.commit(
+			[
+				{
+					type: "document.create",
+					record: sessionDocument(documentId),
+					content: { kind: "base", version: 1, value: { text: "kept" } },
+				},
+			],
+			context,
+		);
+		const sidecarPath = join(directory, `doc-${documentId}.jsonl`);
+		const mainPath = join(directory, "main.jsonl");
+		const sidecarSize = (await stat(sidecarPath)).size;
+		const mainSize = (await stat(mainPath)).size;
+		const torn = new TextEncoder().encode('{"text":"€');
+		await writeFile(sidecarPath, torn.subarray(0, torn.length - 1), { flag: "a" });
+		await writeFile(mainPath, torn.subarray(0, torn.length - 1), { flag: "a" });
+
+		const reopened = await openStorage(directory);
+		expect((await stat(sidecarPath)).size).toBe(sidecarSize);
+		expect((await stat(mainPath)).size).toBe(mainSize);
+		expect(
+			await reopened.commit(
+				[{ type: "document.change", id: documentId, content: { kind: "delta", version: 1, ops: [] } }],
+				context,
+			),
+		).toBe(3);
+	});
+
+	it("removes complete unconfirmed sidecar tails without resurrecting terminal tasks", async () => {
+		const directory = await tempDirectory();
+		const storage = await openStorage(directory);
+		await createRoot(storage);
+		const taskId = await storage.mintId();
+		await storage.commit([{ type: "task", value: pendingTask(taskId) }], context);
+		await storage.commit([{ type: "task", value: terminalTask(taskId) }], context);
+		const sidecarPath = join(directory, `task-${taskId}.jsonl`);
+		const confirmedSize = (await stat(sidecarPath)).size;
+		await writeFile(
+			sidecarPath,
+			`${JSON.stringify({
+				format: 1,
+				type: "record",
+				seq: 4,
+				ordinal: 0,
+				payload: { type: "task", value: pendingTask(taskId, "stale") },
+			})}\n`,
+			{ flag: "a" },
+		);
+
+		const reopened = await openStorage(directory);
+		expect((await stat(sidecarPath)).size).toBe(confirmedSize);
+		expect(await reopened.task(taskId, context)).toEqual(terminalTask(taskId));
+		expect(await reopened.commit([], context)).toBe(4);
+	});
+
+	it("fails open when confirmed sidecar data is missing", async () => {
+		const directory = await tempDirectory();
+		const storage = await openStorage(directory);
+		await createRoot(storage);
+		const documentId = await storage.mintId();
+		await storage.commit(
+			[
+				{
+					type: "document.create",
+					record: sessionDocument(documentId),
+					content: { kind: "base", version: 1, value: {} },
+				},
+			],
+			context,
+		);
+		await writeFile(join(directory, `doc-${documentId}.jsonl`), "");
+		await expect(JsonlStorage.open(directory, new NodeExecutionEnv({ cwd: directory }), context)).rejects.toThrow(
+			"Missing confirmed sidecar record",
+		);
+	});
+
+	it("rejects a confirmed record after an unconfirmed sidecar record", async () => {
+		const directory = await tempDirectory();
+		const storage = await openStorage(directory);
+		await createRoot(storage);
+		const documentId = await storage.mintId();
+		await storage.commit(
+			[
+				{
+					type: "document.create",
+					record: sessionDocument(documentId),
+					content: { kind: "base", version: 1, value: { count: 0 } },
+				},
+			],
+			context,
+		);
+		await storage.commit(
+			[
+				{
+					type: "document.change",
+					id: documentId,
+					content: { kind: "delta", version: 1, ops: [["s", ["count"], 1]] },
+				},
+			],
+			context,
+		);
+		const path = join(directory, `doc-${documentId}.jsonl`);
+		const [first, second] = await readLines(path);
+		const unconfirmed = JSON.stringify({
+			format: 1,
+			type: "record",
+			seq: 2,
+			ordinal: 999,
+			payload: { type: "document", id: documentId, content: { kind: "delta", version: 1, ops: [] } },
+		});
+		await writeFile(path, `${first}\n${unconfirmed}\n${second}\n`);
+		await expect(JsonlStorage.open(directory, new NodeExecutionEnv({ cwd: directory }), context)).rejects.toThrow(
+			"Confirmed record follows an unconfirmed tail",
+		);
+	});
+
+	it("rejects non-increasing main commit sequences", async () => {
+		const directory = await tempDirectory();
+		const storage = await openStorage(directory);
+		await createRoot(storage);
+		const path = join(directory, "main.jsonl");
+		await writeFile(path, await readFile(path), { flag: "a" });
+
+		await expect(JsonlStorage.open(directory, new NodeExecutionEnv({ cwd: directory }), context)).rejects.toThrow(
+			"Commit sequence does not strictly increase",
+		);
+	});
+
+	it("rejects structurally invalid confirmed document content", async () => {
+		const directory = await tempDirectory();
+		const storage = await openStorage(directory);
+		await createRoot(storage);
+		const documentId = await storage.mintId();
+		await storage.commit(
+			[
+				{
+					type: "document.create",
+					record: sessionDocument(documentId),
+					content: { kind: "base", version: 1, value: {} },
+				},
+			],
+			context,
+		);
+		const path = join(directory, `doc-${documentId}.jsonl`);
+		const record = JSON.parse((await readFile(path, "utf8")).trim()) as {
+			payload: { content: Record<string, unknown> };
+		};
+		delete record.payload.content.value;
+		await writeFile(path, `${JSON.stringify(record)}\n`);
+
+		await expect(JsonlStorage.open(directory, new NodeExecutionEnv({ cwd: directory }), context)).rejects.toThrow(
+			"Invalid document content",
+		);
+	});
+
+	it("rejects malformed complete main and sidecar lines", async () => {
+		const mainDirectory = await tempDirectory();
+		const mainStorage = await openStorage(mainDirectory);
+		await createRoot(mainStorage);
+		await writeFile(join(mainDirectory, "main.jsonl"), "{bad}\n", { flag: "a" });
+		await expect(
+			JsonlStorage.open(mainDirectory, new NodeExecutionEnv({ cwd: mainDirectory }), context),
+		).rejects.toThrow("Malformed complete main.jsonl");
+
+		const sidecarDirectory = await tempDirectory();
+		const sidecarStorage = await openStorage(sidecarDirectory);
+		await createRoot(sidecarStorage);
+		await writeFile(join(sidecarDirectory, "doc-99.jsonl"), "{bad}\n");
+		await expect(
+			JsonlStorage.open(sidecarDirectory, new NodeExecutionEnv({ cwd: sidecarDirectory }), context),
+		).rejects.toThrow("Malformed complete doc-99.jsonl");
+	});
+});
--- a/packages/durable/test/memory-storage.test.ts
+++ b/packages/durable/test/memory-storage.test.ts
@ -1,4 +1,30 @@
+import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
+import { expect, it } from "vitest";
 import { MemoryStorage } from "../src/storage/memory.ts";
+import { ROOT_CONVERSATION_ID } from "../src/types.ts";
 import { registerStorageConformance } from "./storage-conformance.ts";

 registerStorageConformance("Pico MemoryStorage", () => new MemoryStorage());
+
+it("does not expose retained state through a prepared commit", async () => {
+	const storage = new MemoryStorage();
+	await storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], BACKGROUND_CONTEXT);
+	const prepared = storage.prepareCommit([
+		{
+			type: "entry",
+			value: {
+				id: 2,
+				conversationId: ROOT_CONVERSATION_ID,
+				kind: "test",
+				data: { nested: [1] },
+			},
+		},
+	]);
+	const exposed = prepared.writes[0];
+	if (exposed.type !== "entry") throw new Error("Expected an entry write");
+	expect(() => (exposed.value.data as { nested: number[] }).nested.push(2)).toThrow();
+
+	expect(prepared.apply()).toBe(2);
+	expect(prepared.apply()).toBe(2);
+	expect((await storage.entry(2, BACKGROUND_CONTEXT))?.entry.data).toEqual({ nested: [1] });
+});
--- a/packages/durable/test/storage-benchmark.ts
+++ b/packages/durable/test/storage-benchmark.ts
@ -3,7 +3,7 @@ import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
 import type { DocumentCreate, Id, Seq, Storage, StorageWrite, TaskRecord } from "../src/types.ts";
 import { ROOT_CONVERSATION_ID } from "../src/types.ts";

-export const STORAGE_BENCHMARK_BACKENDS = ["memory", "sqlite"] as const;
+export const STORAGE_BENCHMARK_BACKENDS = ["memory", "sqlite", "jsonl"] as const;
 export type StorageBenchmarkBackend = (typeof STORAGE_BENCHMARK_BACKENDS)[number];

 export type StorageBenchmarkScale = {
--- a/packages/durable/test/storage-memory.ts
+++ b/packages/durable/test/storage-memory.ts
@ -1,11 +1,12 @@
 import { execFile } from "node:child_process";
-import { mkdtemp, rm, stat } from "node:fs/promises";
+import { mkdtemp, readdir, rm, stat } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { DatabaseSync } from "node:sqlite";
 import { fileURLToPath } from "node:url";
 import { promisify } from "node:util";
 import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
+import { openNodeJsonlStorage } from "../src/storage/jsonl/node.ts";
 import { MemoryStorage } from "../src/storage/memory.ts";
 import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts";
 import type { Storage } from "../src/types.ts";
@ -25,12 +26,23 @@ type MemorySnapshot = {
 	readonly external: number;
 };

-type SqliteFootprint = {
-	readonly mainBytes: number;
-	readonly walBytes: number;
-	readonly pageCount: number;
-	readonly freelistCount: number;
-};
+type DiskFootprint =
+	| {
+			readonly kind: "sqlite";
+			readonly mainBytes: number;
+			readonly auxiliaryBytes: number;
+			readonly fileCount: number;
+			readonly pageCount: number;
+			readonly freelistCount: number;
+	  }
+	| {
+			readonly kind: "jsonl";
+			readonly mainBytes: number;
+			readonly auxiliaryBytes: number;
+			readonly fileCount: number;
+			readonly documentFiles: number;
+			readonly taskFiles: number;
+	  };

 type StorageMemoryResult = {
 	readonly backend: StorageBenchmarkBackend;
@ -39,7 +51,7 @@ type StorageMemoryResult = {
 	readonly baseline: MemorySnapshot;
 	readonly postSeed: MemorySnapshot;
 	readonly postRead: MemorySnapshot;
-	readonly sqlite?: SqliteFootprint;
+	readonly disk?: DiskFootprint;
 };

 const execFileAsync = promisify(execFile);
@ -63,7 +75,7 @@ async function fileSize(path: string): Promise<number> {
 	}
 }

-function sqliteMetrics(path: string): SqliteFootprint {
+function sqliteMetrics(path: string): Extract<DiskFootprint, { readonly kind: "sqlite" }> {
 	const database = new DatabaseSync(path, { readOnly: true });
 	try {
 		const pageCount = database.prepare("SELECT page_count AS value FROM pragma_page_count()").get() as {
@ -72,12 +84,38 @@ function sqliteMetrics(path: string): SqliteFootprint {
 		const freelistCount = database.prepare("SELECT freelist_count AS value FROM pragma_freelist_count()").get() as {
 			readonly value: number;
 		};
-		return { mainBytes: 0, walBytes: 0, pageCount: pageCount.value, freelistCount: freelistCount.value };
+		return {
+			kind: "sqlite",
+			mainBytes: 0,
+			auxiliaryBytes: 0,
+			fileCount: 0,
+			pageCount: pageCount.value,
+			freelistCount: freelistCount.value,
+		};
 	} finally {
 		database.close();
 	}
 }

+async function jsonlMetrics(directory: string): Promise<Extract<DiskFootprint, { readonly kind: "jsonl" }>> {
+	const entries = await readdir(directory, { withFileTypes: true });
+	let mainBytes = 0;
+	let auxiliaryBytes = 0;
+	let fileCount = 0;
+	let documentFiles = 0;
+	let taskFiles = 0;
+	for (const entry of entries) {
+		if (!entry.isFile()) continue;
+		const bytes = await fileSize(join(directory, entry.name));
+		fileCount++;
+		if (entry.name === "main.jsonl") mainBytes += bytes;
+		else auxiliaryBytes += bytes;
+		if (entry.name.startsWith("doc-") && entry.name.endsWith(".jsonl")) documentFiles++;
+		if (entry.name.startsWith("task-") && entry.name.endsWith(".jsonl")) taskFiles++;
+	}
+	return { kind: "jsonl", mainBytes, auxiliaryBytes, fileCount, documentFiles, taskFiles };
+}
+
 async function runWorker(backend: StorageBenchmarkBackend, scale: StorageBenchmarkScale): Promise<void> {
 	let storage: Storage;
 	let directory: string | undefined;
@ -85,9 +123,12 @@ async function runWorker(backend: StorageBenchmarkBackend, scale: StorageBenchma
 	if (backend === "memory") {
 		storage = new MemoryStorage();
 	} else {
-		directory = await mkdtemp(join(tmpdir(), "pi-durable-memory-"));
-		path = join(directory, "storage.sqlite");
-		storage = await openNodeSqliteStorage(path);
+		directory = await mkdtemp(join(tmpdir(), `pi-durable-${backend}-memory-`));
+		path = join(directory, backend === "sqlite" ? "storage.sqlite" : "storage");
+		storage =
+			backend === "sqlite"
+				? await openNodeSqliteStorage(path)
+				: await openNodeJsonlStorage(path, BACKGROUND_CONTEXT);
 	}

 	try {
@ -108,17 +149,19 @@ async function runWorker(backend: StorageBenchmarkBackend, scale: StorageBenchma
 		if (!Number.isFinite(checksum)) throw new Error("Storage memory read checksum is invalid");
 		collectGarbage();
 		const postRead = snapshot();
-		let sqlite: SqliteFootprint | undefined;
-		if (path !== undefined) {
-			sqlite = sqliteMetrics(path);
-			sqlite = {
-				...sqlite,
+		let disk: DiskFootprint | undefined;
+		if (path !== undefined && backend === "sqlite") {
+			disk = {
+				...sqliteMetrics(path),
 				mainBytes: await fileSize(path),
-				walBytes: await fileSize(`${path}-wal`),
+				auxiliaryBytes: (await fileSize(`${path}-wal`)) + (await fileSize(`${path}-shm`)),
+				fileCount: 3,
 			};
+		} else if (path !== undefined) {
+			disk = await jsonlMetrics(path);
 		}
 		const recordCount = storageBenchmarkPrimaryRecordCount(scale);
-		console.log(JSON.stringify({ backend, scale: scale.name, recordCount, baseline, postSeed, postRead, sqlite }));
+		console.log(JSON.stringify({ backend, scale: scale.name, recordCount, baseline, postSeed, postRead, disk }));
 	} finally {
 		await storage.close(BACKGROUND_CONTEXT);
 		if (directory !== undefined) await rm(directory, { recursive: true, force: true });
@ -167,10 +210,17 @@ async function runDriver(): Promise<void> {
 			"JS heap bytes/primary record": Math.round(
 				delta(result.postSeed, result.baseline, "heapUsed") / result.recordCount,
 			),
-			"live SQLite main MiB": result.sqlite === undefined ? "-" : mebibytes(result.sqlite.mainBytes),
-			"live SQLite WAL MiB": result.sqlite === undefined ? "-" : mebibytes(result.sqlite.walBytes),
-			"SQLite pages/free":
-				result.sqlite === undefined ? "-" : `${result.sqlite.pageCount}/${result.sqlite.freelistCount}`,
+			"disk main MiB": result.disk === undefined ? "-" : mebibytes(result.disk.mainBytes),
+			"disk auxiliary MiB": result.disk === undefined ? "-" : mebibytes(result.disk.auxiliaryBytes),
+			"disk total MiB":
+				result.disk === undefined ? "-" : mebibytes(result.disk.mainBytes + result.disk.auxiliaryBytes),
+			"disk files": result.disk?.fileCount ?? "-",
+			"disk detail":
+				result.disk === undefined
+					? "-"
+					: result.disk.kind === "sqlite"
+						? `pages/free ${result.disk.pageCount}/${result.disk.freelistCount}`
+						: `documents/tasks ${result.disk.documentFiles}/${result.disk.taskFiles}`,
 		})),
 	);
 }
--- a/packages/durable/test/storage-runtime-boundary.test.ts
+++ b/packages/durable/test/storage-runtime-boundary.test.ts
@ -22,8 +22,10 @@ async function sourceGraph(entry: string): Promise<Set<string>> {
 }

 describe("durable storage runtime boundaries", () => {
-	it("keeps the package root free of SQLite and Node imports", async () => {
+	it("keeps the package root limited to portable core storage", async () => {
 		const graph = await sourceGraph("index.ts");
+		expect([...graph].some((path) => path.includes("/env/"))).toBe(false);
+		expect([...graph].some((path) => path.includes("/storage/jsonl/"))).toBe(false);
 		expect([...graph].some((path) => path.includes("/storage/sqlite/"))).toBe(false);
 	});

@ -31,4 +33,15 @@ describe("durable storage runtime boundaries", () => {
 		const graph = await sourceGraph("storage/sqlite/index.ts");
 		expect([...graph].some((path) => path.endsWith("/storage/sqlite/node.ts"))).toBe(false);
 	});
+
+	it("keeps the portable environment subpath free of Node imports", async () => {
+		const graph = await sourceGraph("env/index.ts");
+		expect([...graph].some((path) => path.endsWith("/env/node.ts"))).toBe(false);
+	});
+
+	it("keeps the portable JSONL subpath free of Node imports", async () => {
+		const graph = await sourceGraph("storage/jsonl/index.ts");
+		expect([...graph].some((path) => path.endsWith("/storage/jsonl/node.ts"))).toBe(false);
+		expect([...graph].some((path) => path.endsWith("/env/node.ts"))).toBe(false);
+	});
 });
--- a/packages/durable/test/storage.bench.ts
+++ b/packages/durable/test/storage.bench.ts
@ -1,9 +1,10 @@
 import { strictEqual } from "node:assert/strict";
-import { copyFile, mkdtemp, rm } from "node:fs/promises";
+import { copyFile, cp, mkdtemp, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
 import { afterAll, bench, describe } from "vitest";
+import { openNodeJsonlStorage } from "../src/storage/jsonl/node.ts";
 import { MemoryStorage } from "../src/storage/memory.ts";
 import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts";
 import type { Storage } from "../src/types.ts";
@ -22,7 +23,6 @@ const REOPEN_OPTIONS = { time: 0, iterations: 20, warmupTime: 0, warmupIteration
 type Fixture = {
 	readonly backend: StorageBenchmarkBackend;
 	readonly storage: Storage;
-	readonly path?: string;
 };

 const fixtures: Fixture[] = [];
@ -36,8 +36,17 @@ async function createFixture(backend: StorageBenchmarkBackend): Promise<Fixture>
 	}
 	const directory = await mkdtemp(join(tmpdir(), "pi-durable-benchmark-"));
 	directories.push(directory);
-	const path = join(directory, "storage.sqlite");
-	const fixture = { backend, storage: await openNodeSqliteStorage(path), path } satisfies Fixture;
+	if (backend === "sqlite") {
+		const path = join(directory, "storage.sqlite");
+		const fixture = { backend, storage: await openNodeSqliteStorage(path) } satisfies Fixture;
+		fixtures.push(fixture);
+		return fixture;
+	}
+	const path = join(directory, "storage");
+	const fixture = {
+		backend,
+		storage: await openNodeJsonlStorage(path, BACKGROUND_CONTEXT),
+	} satisfies Fixture;
 	fixtures.push(fixture);
 	return fixture;
 }
@ -117,41 +126,74 @@ for (const scenario of STORAGE_WRITE_BENCHMARKS) {
 	});
 }

-const reopenDirectory = await mkdtemp(join(tmpdir(), "pi-durable-reopen-benchmark-"));
-directories.push(reopenDirectory);
-const reopenPath = join(reopenDirectory, "storage.sqlite");
-const reopenSeed = await openNodeSqliteStorage(reopenPath);
-const reopenDataset = await seedStorageBenchmark(reopenSeed);
-await reopenSeed.close(BACKGROUND_CONTEXT);
-const reopenPaths = await Promise.all(
-	Array.from({ length: REOPEN_OPTIONS.iterations + REOPEN_OPTIONS.warmupIterations }, async () => {
-		const directory = await mkdtemp(join(tmpdir(), "pi-durable-reopen-sample-"));
-		directories.push(directory);
-		const path = join(directory, "storage.sqlite");
-		await copyFile(reopenPath, path);
-		return path;
-	}),
-);
+type PersistentBackend = Exclude<StorageBenchmarkBackend, "memory">;
+type ReopenFixture = {
+	readonly backend: PersistentBackend;
+	readonly path: string;
+	readonly firstEntryId: number;
+	readonly samples: string[];
+};
+
+async function openPersistentStorage(backend: PersistentBackend, path: string): Promise<Storage> {
+	return backend === "sqlite" ? openNodeSqliteStorage(path) : openNodeJsonlStorage(path, BACKGROUND_CONTEXT);
+}
+
+async function copyPersistentStorage(backend: PersistentBackend, source: string, destination: string): Promise<void> {
+	if (backend === "sqlite") await copyFile(source, destination);
+	else await cp(source, destination, { recursive: true });
+}
+
+const reopenFixtures: ReopenFixture[] = [];
+for (const backend of STORAGE_BENCHMARK_BACKENDS) {
+	if (backend === "memory") continue;
+	const seedDirectory = await mkdtemp(join(tmpdir(), `pi-durable-${backend}-reopen-benchmark-`));
+	directories.push(seedDirectory);
+	const seedPath = join(seedDirectory, backend === "sqlite" ? "storage.sqlite" : "storage");
+	const seed = await openPersistentStorage(backend, seedPath);
+	const dataset = await seedStorageBenchmark(seed);
+	await seed.close(BACKGROUND_CONTEXT);
+	const samples = await Promise.all(
+		Array.from({ length: REOPEN_OPTIONS.iterations + REOPEN_OPTIONS.warmupIterations }, async () => {
+			const directory = await mkdtemp(join(tmpdir(), `pi-durable-${backend}-reopen-sample-`));
+			directories.push(directory);
+			const path = join(directory, backend === "sqlite" ? "storage.sqlite" : "storage");
+			await copyPersistentStorage(backend, seedPath, path);
+			return path;
+		}),
+	);
+	reopenFixtures.push({ backend, path: seedPath, firstEntryId: dataset.firstEntryId, samples });
+}
+
 const reopenedStorages: Storage[] = [];
-async function reopenAndRead(path: string): Promise<{ readonly id: number; readonly storage: Storage }> {
-	const storage = await openNodeSqliteStorage(path);
-	const id = (await storage.entry(reopenDataset.firstEntryId, BACKGROUND_CONTEXT))?.entry.id ?? -1;
+async function reopenAndRead(
+	backend: PersistentBackend,
+	path: string,
+	firstEntryId: number,
+): Promise<{ readonly id: number; readonly storage: Storage }> {
+	const storage = await openPersistentStorage(backend, path);
+	const id = (await storage.entry(firstEntryId, BACKGROUND_CONTEXT))?.entry.id ?? -1;
 	return { id, storage };
 }
-const reopenValidation = await reopenAndRead(reopenPath);
-strictEqual(reopenValidation.id, reopenDataset.firstEntryId);
-await reopenValidation.storage.close(BACKGROUND_CONTEXT);
-describe("SQLite reopen and first exact read", () => {
-	bench(
-		"sqlite",
-		async () => {
-			const path = reopenPaths.shift();
-			if (path === undefined) throw new Error("Reopen benchmark fixture pool was exhausted");
-			const result = await reopenAndRead(path);
-			reopenedStorages.push(result.storage);
-		},
-		REOPEN_OPTIONS,
-	);
+
+for (const fixture of reopenFixtures) {
+	const reopenValidation = await reopenAndRead(fixture.backend, fixture.path, fixture.firstEntryId);
+	strictEqual(reopenValidation.id, fixture.firstEntryId);
+	await reopenValidation.storage.close(BACKGROUND_CONTEXT);
+}
+
+describe("reopen and first exact read", () => {
+	for (const fixture of reopenFixtures) {
+		bench(
+			fixture.backend,
+			async () => {
+				const path = fixture.samples.shift();
+				if (path === undefined) throw new Error("Reopen benchmark fixture pool was exhausted");
+				const result = await reopenAndRead(fixture.backend, path, fixture.firstEntryId);
+				reopenedStorages.push(result.storage);
+			},
+			REOPEN_OPTIONS,
+		);
+	}
 });

 afterAll(async () => {
--- a/scripts/check-browser-smoke.mjs
+++ b/scripts/check-browser-smoke.mjs
@ -65,6 +65,9 @@ try {
 	const durableInputs = durableBuild.metafile.inputs;
 	for (const expectedInput of [
 		"packages/durable/src/index.ts",
+		"packages/durable/src/env/index.ts",
+		"packages/durable/src/storage/jsonl/index.ts",
+		"packages/durable/src/storage/jsonl/storage.ts",
 		"packages/durable/src/storage/memory.ts",
 		"packages/durable/src/storage/sqlite/index.ts",
 		"packages/durable/src/storage/sqlite/storage.ts",
@ -73,8 +76,14 @@ try {
 			throw new Error(`Durable browser bundle does not include ${expectedInput}`);
 		}
 	}
-	const nodeAdapter = findInput(durableInputs, "packages/durable/src/storage/sqlite/node.ts");
-	if (nodeAdapter) throw new Error(`Durable browser bundle unexpectedly includes ${nodeAdapter}`);
+	for (const forbiddenInput of [
+		"packages/durable/src/env/node.ts",
+		"packages/durable/src/storage/jsonl/node.ts",
+		"packages/durable/src/storage/sqlite/node.ts",
+	]) {
+		const nodeAdapter = findInput(durableInputs, forbiddenInput);
+		if (nodeAdapter) throw new Error(`Durable browser bundle unexpectedly includes ${nodeAdapter}`);
+	}

 	const agentTreeshakeBuild = await build({
 		entryPoints: ["scripts/agent-treeshake-smoke-entry.ts"],
--- a/scripts/durable-browser-smoke-entry.ts
+++ b/scripts/durable-browser-smoke-entry.ts
@ -1,6 +1,8 @@
 import * as durable from "@earendil-works/pi-durable";
+import * as environment from "@earendil-works/pi-durable/env";
+import * as jsonl from "@earendil-works/pi-durable/storage/jsonl";
 import * as sqlite from "@earendil-works/pi-durable/storage/sqlite";

-// Keep both runtime-neutral public entry points live so the browser smoke build
+// Keep runtime-neutral public entry points live so the browser smoke build
 // catches accidental imports of Node-only adapters or built-ins.
-console.log(Object.keys(durable), Object.keys(sqlite));
+console.log(Object.keys(durable), Object.keys(environment), Object.keys(jsonl), Object.keys(sqlite));