diff --git a/packages/chord/src/sync/batch.ts b/packages/chord/src/sync/batch.ts
index a486bfd66..bc6196ea0 100644
--- a/packages/chord/src/sync/batch.ts
+++ b/packages/chord/src/sync/batch.ts
@@ -1,53 +1,139 @@
+// Bodies on the remote, encoded and decoded as streams, so neither side holds a whole batch or snapshot in memory:
+// gzip-compressed JSON lines, one outbox commit (a batch) or one archive part (a snapshot) per line. gzip, because
+// `CompressionStream` offers it in Node, Workers, and browsers alike, and a browser viewer reads batches too.
+
 import type { ArchivePart } from "../store/archive.ts";
+import { decodeJsonl, encodeJsonl } from "../store/jsonl.ts";
 import type { OutboxCommit } from "../store/outbox.ts";
-import { decodeLogLines, encodeLogLine, LogFormatError } from "./lines.ts";
+import { decodeLogLine, encodeLogLine, LogFormatError } from "./lines.ts";
+
+/** A body on the remote: its bytes in chunks, read once. */
+export type SyncBody = AsyncIterable<Uint8Array>;
+
+/** The bytes a chunk of encoded lines collects before it is compressed. */
+const CHUNK = 64 * 1024;
+
+/** A batch body: `commits` as gzip-compressed JSON lines, read from `commits` only as the body is read. */
+export function encodeBatch(commits: Iterable<OutboxCommit> | AsyncIterable<OutboxCommit>): SyncBody {
+	return encodeLines(
+		(async function* () {
+			for await (const commit of commits) yield encodeLogLine(commit);
+		})(),
+	);
+}
+
+/** The commits of a batch body, in order, as they arrive. */
+export async function* decodeBatch(body: SyncBody): AsyncGenerator<OutboxCommit> {
+	for await (const { commit } of decodeBatchLines(body)) yield commit;
+}
+
+/** The commits of a batch body with the size of each line, for callers that bound what they hold. */
+export async function* decodeBatchLines(
+	body: SyncBody,
+): AsyncGenerator<{ readonly commit: OutboxCommit; readonly bytes: number }> {
+	let previous = 0;
+	for await (const { text, number } of lines(gunzip(body))) {
+		const commit = decodeLogLine(text, number);
+		if (commit.seq <= previous) throw new LogFormatError(`Log line ${number} does not increase the sequence`);
+		previous = commit.seq;
+		yield { commit, bytes: text.length + 1 };
+	}
+}
+
+/** A snapshot body: a store archive (`store.export()`), one part per JSON line, gzip-compressed. */
+export function encodeSnapshot(parts: AsyncIterable<ArchivePart>): SyncBody {
+	return gzip(encodeJsonl(parts));
+}
+
+/** The parts of a snapshot body as they arrive, unchecked: `importStore()` checks them (a trust boundary). */
+export function decodeSnapshot(body: SyncBody): AsyncIterable<unknown> {
+	return decodeJsonl(gunzip(body));
+}
+
+/** Text lines, already newline-terminated, gzip-compressed in chunks of whole lines. */
+function encodeLines(lines: AsyncIterable<string>): SyncBody {
+	return gzip(
+		(async function* () {
+			const encoder = new TextEncoder();
+			let pending = "";
+			for await (const line of lines) {
+				pending += line;
+				if (pending.length >= CHUNK) {
+					yield encoder.encode(pending);
+					pending = "";
+				}
+			}
+			if (pending.length > 0) yield encoder.encode(pending);
+		})(),
+	);
+}
+
+/** The lines of decompressed `chunks` without their newlines; a body that ends inside a line is an error. */
+async function* lines(chunks: AsyncIterable<Uint8Array>): AsyncGenerator<{ readonly text: string; number: number }> {
+	const decoder = new TextDecoder("utf-8", { fatal: true });
+	let rest = "";
+	let number = 0;
+	const decode = (chunk?: Uint8Array): string => {
+		try {
+			return chunk === undefined ? decoder.decode() : decoder.decode(chunk, { stream: true });
+		} catch (error) {
+			throw new LogFormatError(`Log line ${number + 1} is not UTF-8`, { cause: error });
+		}
+	};
+	for await (const chunk of chunks) {
+		rest += decode(chunk);
+		let start = 0;
+		for (let end = rest.indexOf("\n"); end >= 0; end = rest.indexOf("\n", start)) {
+			yield { text: rest.slice(start, end), number: ++number };
+			start = end + 1;
+		}
+		rest = rest.slice(start);
+	}
+	rest += decode();
+	if (rest.length > 0) throw new LogFormatError(`The body ends inside log line ${number + 1}`);
+}
+
+function gzip(chunks: AsyncIterable<Uint8Array>): SyncBody {
+	return pipe(chunks, new CompressionStream("gzip"));
+}
+
+function gunzip(chunks: SyncBody): AsyncIterable<Uint8Array> {
+	return pipe(chunks, new DecompressionStream("gzip"));
+}
 
 /**
- * A batch as uploaded: outbox commits after one sequence as gzip-compressed JSON lines. gzip, because
- * `CompressionStream` offers it in Node, Workers, and browsers alike, and a browser viewer reads batches too.
+ * `chunks` through `transform`, as they come: the source is written while the output is read, so each holds no more
+ * than the stream's buffers. Stopping early, or a failure on either side, ends the source too (its `return()`, which
+ * for an export gives the store's commit queue back).
  */
-export async function encodeBatch(commits: readonly OutboxCommit[]): Promise<Uint8Array> {
-	return gzip(new TextEncoder().encode(commits.map(encodeLogLine).join("")));
-}
-
-export async function decodeBatch(bytes: Uint8Array): Promise<OutboxCommit[]> {
-	const { commits, torn } = decodeLogLines(await gunzipText(bytes));
-	if (torn > 0) throw new LogFormatError("Batch ends inside a log line");
-	return commits;
-}
-
-/** A snapshot as uploaded: a store archive (`store.export()`), one part per JSON line, gzip-compressed. */
-export async function encodeSnapshot(parts: AsyncIterable<ArchivePart>): Promise<Uint8Array> {
-	let text = "";
-	for await (const part of parts) text += `${JSON.stringify(part)}\n`;
-	return gzip(new TextEncoder().encode(text));
-}
-
-/** The parts of a snapshot, unchecked: `importStore()` checks them (a trust boundary). */
-export async function decodeSnapshot(bytes: Uint8Array): Promise<unknown[]> {
-	const text = await gunzipText(bytes);
-	if (text.length > 0 && !text.endsWith("\n")) throw new LogFormatError("Snapshot ends inside a line");
-	return text
-		.split("\n")
-		.filter((line) => line.length > 0)
-		.map((line, index) => {
-			try {
-				return JSON.parse(line) as unknown;
-			} catch (error) {
-				throw new LogFormatError(`Snapshot line ${index + 1} is not JSON`, { cause: error });
-			}
-		});
-}
-
-function gzip(bytes: Uint8Array): Promise<Uint8Array> {
-	return transform(bytes, new CompressionStream("gzip"));
-}
-
-async function gunzipText(bytes: Uint8Array): Promise<string> {
-	return new TextDecoder("utf-8", { fatal: true }).decode(await transform(bytes, new DecompressionStream("gzip")));
-}
-
-async function transform(bytes: Uint8Array, stream: CompressionStream | DecompressionStream): Promise<Uint8Array> {
-	const output = new Blob([bytes as Uint8Array<ArrayBuffer>]).stream().pipeThrough(stream);
-	return new Uint8Array(await new Response(output).arrayBuffer());
+async function* pipe(
+	chunks: AsyncIterable<Uint8Array>,
+	transform: CompressionStream | DecompressionStream,
+): AsyncGenerator<Uint8Array> {
+	const writer = transform.writable.getWriter();
+	const reader = transform.readable.getReader();
+	const writing = (async () => {
+		try {
+			for await (const chunk of chunks) await writer.write(chunk as Uint8Array<ArrayBuffer>);
+			await writer.close();
+		} catch (error) {
+			await writer.abort(error).catch(() => {});
+			throw error;
+		}
+	})();
+	// Its failure reaches the reader too (the abort errors the readable side); this only keeps it from going unhandled.
+	writing.catch(() => {});
+	let done = false;
+	try {
+		while (true) {
+			const result = await reader.read();
+			if (result.done) break;
+			yield result.value;
+		}
+		done = true;
+		await writing;
+	} finally {
+		if (!done) await reader.cancel().catch(() => {});
+		reader.releaseLock();
+	}
 }
diff --git a/packages/chord/src/sync/client.ts b/packages/chord/src/sync/client.ts
index 18f506146..532c44b40 100644
--- a/packages/chord/src/sync/client.ts
+++ b/packages/chord/src/sync/client.ts
@@ -1,7 +1,10 @@
 // Session sync over a Chord store's outbox: upload the commits after the last sequence a copy shares with a remote
 // into a slot the remote creates once, pull a copy forward with `applyBatch`, restore from an archive, fold history
-// into snapshots, and decide when to upload. The remote and the host's sync state are interfaces (`remote.ts`); a
-// directory-backed pair for Node is in `./node`.
+// into snapshots, and decide when to upload. The remote and the host's sync state are interfaces (`remote.ts`).
+//
+// Bounded memory: bodies are streams (`batch.ts`). An upload reads the outbox a page at a time and stops a batch at
+// `BATCH_BYTES`; a pull applies a batch in chunks of at most `APPLY_BYTES`; a snapshot is exported and imported as it
+// streams. Nothing holds a whole batch, snapshot or store, so a Durable Object's isolate (128 MB) can run it.
 
 import type { Context } from "../context/index.ts";
 import { applyBatch } from "../store/apply-batch.ts";
@@ -9,11 +12,12 @@ import type { Backend } from "../store/backend.ts";
 import { importStore } from "../store/import.ts";
 import type { OutboxCommit } from "../store/outbox.ts";
 import { type ChordStore, openStore, type Store } from "../store/store.ts";
-import { decodeBatch, decodeSnapshot, encodeBatch, encodeSnapshot } from "./batch.ts";
+import { decodeBatchLines, decodeSnapshot, encodeBatch, encodeSnapshot, type SyncBody } from "./batch.ts";
 import {
 	type BatchInfo,
 	HistoryUnavailableError,
 	LOG_START,
+	type PutBatchResult,
 	remoteHead,
 	remoteHistory,
 	type Seq,
@@ -21,8 +25,17 @@ import {
 	type SessionSyncStore,
 } from "./remote.ts";
 
-/** Commits read from the outbox per uploaded batch. */
-const BATCH_COMMITS = 1000;
+/** Commits read from the outbox per page at most. */
+const OUTBOX_PAGE = 100;
+/** About how much one outbox page holds: the count of the next page follows the size of the commits read. */
+const PAGE_BYTES = 1024 * 1024;
+/**
+ * Where an upload ends a batch: once its lines (before compression) reach this size. One commit larger than this is
+ * a batch of its own.
+ */
+export const BATCH_BYTES = 8 * 1024 * 1024;
+/** How much of a batch a pull holds before it applies it: commits are applied in chunks of about this size. */
+export const APPLY_BYTES = 4 * 1024 * 1024;
 
 /**
  * What a host keeps for each copy of a session it syncs: the last sequence the copy shares with the remote. Commits
@@ -123,22 +136,30 @@ export async function uploadSession(options: {
 		const head = storeHead(source);
 		if (head < synced) return { type: "behind", head, syncedHead: synced };
 		let to: Seq;
-		let body: Uint8Array;
+		let body: SyncBody;
 		let archive = false;
+		let close: (() => Promise<void>) | undefined;
 		if (synced === LOG_START && outbox.floor > 0) {
 			const snapshot = await exportSnapshot(source, context);
 			to = snapshot.to;
 			body = snapshot.body;
+			close = snapshot.close;
 			archive = true;
 		} else {
-			const commits = await readOutbox(source, synced, BATCH_COMMITS, context);
-			if (commits.length === 0)
+			const range = await outboxRange(source, synced, context);
+			if (range === undefined)
 				return uploaded === undefined ? { type: "idle" } : { type: "uploaded", head: uploaded };
-			to = commits.at(-1)!.seq;
-			body = await encodeBatch(commits);
+			to = range.to;
+			body = encodeBatch(outboxCommits(source, synced, to, context));
 		}
 		const after = synced;
-		const result = await store.putBatch(sessionId, after, to, device, body, context, archive ? { archive } : {});
+		let result: PutBatchResult;
+		try {
+			result = await store.putBatch(sessionId, after, to, device, body, context, archive ? { archive } : {});
+		} finally {
+			// A remote that answered without reading the archive leaves the export holding the commit queue.
+			await close?.();
+		}
 		// A history that does not end at `after` is another one too: the remote never held this copy's past `after`.
 		if (result.type === "not_head") return { type: "diverged", at: after };
 		if (result.type === "exists" && result.device !== device) return { type: "diverged", at: after };
@@ -150,23 +171,123 @@ export async function uploadSession(options: {
 	}
 }
 
-/** The store's state at its head as a snapshot body. `export()` holds the commit queue, so the head cannot move. */
-async function exportSnapshot(
-	store: Store,
-	context: Context,
-): Promise<{ readonly to: Seq; readonly body: Uint8Array }> {
+/**
+ * Where the next batch after `after` ends: the outbox read a page at a time, holding one page, until the lines reach
+ * `BATCH_BYTES` (one commit at least). Undefined when the outbox has nothing after `after`.
+ */
+async function outboxRange(store: Store, after: Seq, context: Context): Promise<{ readonly to: Seq } | undefined> {
 	let to: Seq | undefined;
+	let bytes = 0;
+	const pages = new OutboxPages(store, after, context);
+	for (let page = await pages.next(); page.length > 0; page = await pages.next()) {
+		for (const commit of page) {
+			if (to !== undefined && bytes + encodedSize(commit) > BATCH_BYTES) return { to };
+			bytes += encodedSize(commit);
+			to = commit.seq;
+		}
+		if (bytes >= BATCH_BYTES) break;
+	}
+	return to === undefined ? undefined : { to };
+}
+
+/** The outbox commits after `after` through `to`, a page at a time, as the body is read. */
+async function* outboxCommits(store: Store, after: Seq, to: Seq, context: Context): AsyncGenerator<OutboxCommit> {
+	const pages = new OutboxPages(store, after, context);
+	let at = after;
+	while (at < to) {
+		const page = await pages.next();
+		if (page.length === 0) throw new OutboxUnavailableError(`The outbox ends at ${at}, before ${to}`);
+		for (const commit of page) {
+			if (commit.seq > to) return;
+			yield commit;
+			at = commit.seq;
+		}
+	}
+}
+
+/**
+ * Pages of the outbox after a sequence, each about `PAGE_BYTES`: the outbox reads by count, so the next page's count
+ * follows the commits read so far (`OUTBOX_PAGE` to start, one commit at least). A run of large commits reads one at a
+ * time instead of `OUTBOX_PAGE` of them at once.
+ */
+class OutboxPages {
+	readonly #store: Store;
+	readonly #context: Context;
+	#after: Seq;
+	#limit = OUTBOX_PAGE;
+	#done = false;
+
+	constructor(store: Store, after: Seq, context: Context) {
+		this.#store = store;
+		this.#after = after;
+		this.#context = context;
+	}
+
+	async next(): Promise<OutboxCommit[]> {
+		if (this.#done) return [];
+		const page = await readOutbox(this.#store, this.#after, this.#limit, this.#context);
+		if (page.length < this.#limit) this.#done = true;
+		if (page.length === 0) return page;
+		this.#after = page.at(-1)!.seq;
+		const average = page.reduce((size, commit) => size + encodedSize(commit), 0) / page.length;
+		this.#limit = Math.max(1, Math.min(OUTBOX_PAGE, Math.floor(PAGE_BYTES / average)));
+		return page;
+	}
+}
+
+/** A commit's line size before compression, as `BATCH_BYTES` counts it. */
+function encodedSize(commit: OutboxCommit): number {
+	return JSON.stringify(commit.writes).length + 64;
+}
+
+/** A snapshot of a store being exported: where it ends, its body, and `close()` for when the body is not read to the end. */
+type ExportedSnapshot = { readonly to: Seq; readonly body: SyncBody; close(): Promise<void> };
+
+/**
+ * The store's state as a streamed snapshot body. The export holds the store's commit queue from its first part until
+ * the body is read to the end or `close()` is called, so `to` (its manifest's `lastSeq`) is exactly what it holds.
+ */
+async function exportSnapshot(store: Store, context: Context): Promise<ExportedSnapshot> {
 	const parts = store.export(context);
-	const body = await encodeSnapshot(
-		(async function* () {
-			for await (const part of parts) {
-				if (part.kind === "manifest") to = part.lastSeq;
-				yield part;
-			}
-		})(),
-	);
-	if (to === undefined) throw new Error("A store export has no manifest");
-	return { to, body };
+	const first = await parts.next();
+	if (first.done || first.value.kind !== "manifest") {
+		await parts.return?.(undefined);
+		throw new Error("A store export does not start with its manifest");
+	}
+	const manifest = first.value;
+	return {
+		to: manifest.lastSeq,
+		body: encodeSnapshot(
+			(async function* () {
+				yield manifest;
+				yield* parts;
+			})(),
+		),
+		close: async () => {
+			await parts.return?.(undefined);
+		},
+	};
+}
+
+/**
+ * Apply a batch body to `store` as it streams, in chunks of about `APPLY_BYTES`, skipping the commits the store has
+ * (a pull resuming inside a batch). Each chunk is one `applyBatch()` transaction; a pull stopped between chunks
+ * resumes after the last applied one.
+ */
+async function applyBody(store: ChordStore, body: SyncBody, context: Context): Promise<void> {
+	let chunk: OutboxCommit[] = [];
+	let bytes = 0;
+	for await (const line of decodeBatchLines(body)) {
+		if (line.commit.seq <= store.lastSeq) continue;
+		chunk.push(line.commit);
+		bytes += line.bytes;
+		if (bytes >= APPLY_BYTES) {
+			await applyBatch(store, chunk, context);
+			chunk = [];
+			bytes = 0;
+		}
+	}
+	if (chunk.length > 0) await applyBatch(store, chunk, context);
 }
 
 /** This copy has commits it has not uploaded: upload them first, which may find that the copy diverged. */
@@ -208,7 +329,7 @@ export async function pullSession(options: {
 	if (at() > synced) throw new UnsyncedCommitsError(sessionId, at(), synced);
 	const history = remoteHistory(listing, at());
 	try {
-		const bodies: { readonly to: Seq; readonly archive: boolean; read(): Promise<Uint8Array | undefined> }[] = [];
+		const bodies: { readonly to: Seq; readonly archive: boolean; read(): Promise<SyncBody | undefined> }[] = [];
 		if (history.snapshot !== undefined) {
 			const snapshot = history.snapshot;
 			bodies.push({ to: snapshot, archive: true, read: () => remote.getSnapshot(sessionId, snapshot, context) });
@@ -228,15 +349,13 @@ export async function pullSession(options: {
 			if (body.archive) {
 				// An archive restores a whole store: only into an empty backend.
 				if (target !== undefined || local.kind !== "empty") throw new HistoryUnavailableError(at());
-				target = await importStore(local.backend, await decodeSnapshot(bytes), context, { outbox: true });
+				target = await importStore(local.backend, decodeSnapshot(bytes), context, { outbox: true });
 			} else {
 				if (target === undefined) {
 					if (local.kind !== "empty") throw new Error("unreachable: a store copy always has a target");
 					target = await openStore(local.backend, context, { outbox: true });
 				}
-				const from = target.lastSeq;
-				const commits = (await decodeBatch(bytes)).filter((commit) => commit.seq > from);
-				if (commits.length > 0) await applyBatch(target as ChordStore, commits, context);
+				await applyBody(target as ChordStore, bytes, context);
 			}
 			await state.set(sessionId, { syncedHead: at() }, context);
 		}
@@ -259,8 +378,8 @@ export async function pullSession(options: {
 }
 
 /**
- * Fold a session's history into one snapshot: restore it into `scratch` (an empty backend, for example an in-memory
- * SQLite database), upload its `export()` as a snapshot, drop the commits of the batches it covers, keeping their
+ * Fold a session's history into one snapshot: restore it into `scratch` (an empty backend; a file or a Durable Object's
+ * SQLite storage, not `:memory:`, so the restored store is not held in memory), upload its `export()` as a snapshot, drop the commits of the batches it covers, keeping their
  * slots taken, and delete older snapshots. The remote runs this, not clients.
  */
 export async function foldSession(options: {
@@ -282,7 +401,11 @@ export async function foldSession(options: {
 	});
 	try {
 		const snapshot = await exportSnapshot(restored.store, context);
-		await store.putSnapshot(sessionId, snapshot.to, snapshot.body, context);
+		try {
+			await store.putSnapshot(sessionId, snapshot.to, snapshot.body, context);
+		} finally {
+			await snapshot.close();
+		}
 		await store.foldBatches(
 			sessionId,
 			history.batches.filter((batch: BatchInfo) => batch.to <= snapshot.to).map((batch) => batch.after),
diff --git a/packages/chord/src/sync/files.ts b/packages/chord/src/sync/files.ts
index 9481d4253..0daa8528c 100644
--- a/packages/chord/src/sync/files.ts
+++ b/packages/chord/src/sync/files.ts
@@ -10,12 +10,27 @@ export type FileResult<T> = { readonly ok: true; readonly value: T } | { readonl
 /** A directory entry. */
 export type SyncFileInfo = { readonly name: string; readonly kind: "file" | "directory" | "symlink" };
 
+/** A file opened for reads at offsets: what streaming a body back needs. */
+export interface SyncFileReader {
+	/** Up to `length` bytes at `offset`; fewer only at end of file. */
+	read(offset: number, length: number, context: Context): Promise<FileResult<Uint8Array>>;
+	close(context: Context): Promise<void>;
+}
+
 /** File operations for the directory-backed sync pieces. A missing file or directory fails with `code: "not_found"`. */
 export interface SyncFileSystem {
 	joinPath(parts: string[], context: Context): Promise<FileResult<string>>;
 	readTextFile(path: string, context: Context): Promise<FileResult<string>>;
 	readBinaryFile(path: string, context: Context): Promise<FileResult<Uint8Array>>;
+	/** Open a file for bounded reads, so a body streams back without being read whole. */
+	openBinaryReader(
+		path: string,
+		options: { noFollow?: boolean } | undefined,
+		context: Context,
+	): Promise<FileResult<SyncFileReader>>;
 	writeFile(path: string, content: string | Uint8Array, context: Context): Promise<FileResult<void>>;
+	/** Append to a file, creating it: a body is written chunk by chunk as it streams. */
+	appendFile(path: string, content: string | Uint8Array, context: Context): Promise<FileResult<void>>;
 	renameFile(sourcePath: string, destinationPath: string, context: Context): Promise<FileResult<void>>;
 	listDir(path: string, context: Context): Promise<FileResult<SyncFileInfo[]>>;
 	createDir(path: string, options: { recursive?: boolean } | undefined, context: Context): Promise<FileResult<void>>;
diff --git a/packages/chord/src/sync/filesystem.ts b/packages/chord/src/sync/filesystem.ts
index 14b1a48b9..2d9e3593d 100644
--- a/packages/chord/src/sync/filesystem.ts
+++ b/packages/chord/src/sync/filesystem.ts
@@ -1,4 +1,5 @@
 import type { Context } from "../context/index.ts";
+import type { SyncBody } from "./batch.ts";
 import { isNotFound, type SyncFileSystem, unwrap } from "./files.ts";
 import {
 	type BatchInfo,
@@ -12,6 +13,13 @@ import {
 } from "./remote.ts";
 
 const pad = (value: number): string => String(value).padStart(16, "0");
+/** How much of a body is read from its file at a time. */
+const READ_CHUNK = 256 * 1024;
+
+/** One buffer as a body. */
+async function* once(bytes: Uint8Array): SyncBody {
+	yield bytes;
+}
 const SLOT_NAME = /^(\d{16})\.json$/;
 const SNAPSHOT_NAME = /^(\d{16})\.batch$/;
 const SESSION_ID = /^[A-Za-z0-9_-][A-Za-z0-9._-]*$/;
@@ -49,7 +57,7 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
 		after: Seq,
 		to: Seq,
 		device: string,
-		body: Uint8Array,
+		body: SyncBody,
 		context: Context,
 		options: { readonly archive?: boolean } = {},
 	): Promise<PutBatchResult> {
@@ -64,23 +72,23 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
 			sessionId,
 			["slots"],
 			`${pad(after)}.json`,
-			new TextEncoder().encode(JSON.stringify(slot)),
+			once(new TextEncoder().encode(JSON.stringify(slot))),
 			context,
 		);
 		return { type: "created" };
 	}
 
-	async getBatch(sessionId: string, after: Seq, context: Context): Promise<Uint8Array | undefined> {
+	async getBatch(sessionId: string, after: Seq, context: Context): Promise<SyncBody | undefined> {
 		const slot = await this.#slot(sessionId, after, context);
 		if (slot === undefined || slot.folded) return undefined;
-		return this.#read(context, sessionId, "batches", `${pad(after)}.batch`);
+		return this.#stream(context, sessionId, "batches", `${pad(after)}.batch`);
 	}
 
-	getSnapshot(sessionId: string, to: Seq, context: Context): Promise<Uint8Array | undefined> {
-		return this.#read(context, sessionId, "snapshots", `${pad(to)}.batch`);
+	getSnapshot(sessionId: string, to: Seq, context: Context): Promise<SyncBody | undefined> {
+		return this.#stream(context, sessionId, "snapshots", `${pad(to)}.batch`);
 	}
 
-	async putSnapshot(sessionId: string, to: Seq, body: Uint8Array, context: Context): Promise<void> {
+	async putSnapshot(sessionId: string, to: Seq, body: SyncBody, context: Context): Promise<void> {
 		await this.#write(sessionId, ["snapshots"], `${pad(to)}.batch`, body, context);
 	}
 
@@ -93,7 +101,7 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
 				sessionId,
 				["slots"],
 				`${pad(after)}.json`,
-				new TextEncoder().encode(JSON.stringify(folded)),
+				once(new TextEncoder().encode(JSON.stringify(folded))),
 				context,
 			);
 			const body = await this.#path(context, sessionId, "batches", `${pad(after)}.batch`);
@@ -145,7 +153,7 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
 	}
 
 	async putMeta(sessionId: string, meta: JsonObject, context: Context): Promise<void> {
-		await this.#write(sessionId, [], "meta.json", new TextEncoder().encode(JSON.stringify(meta)), context);
+		await this.#write(sessionId, [], "meta.json", once(new TextEncoder().encode(JSON.stringify(meta))), context);
 	}
 
 	async getMeta(sessionId: string, context: Context): Promise<JsonObject | undefined> {
@@ -170,12 +178,38 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
 		return unwrap(result);
 	}
 
-	/** Write through a temporary file and a rename, so a reader never sees half a file. */
+	/** A body streamed back from its file, a chunk at a time; undefined when the file is missing. */
+	async #stream(context: Context, sessionId: string, ...parts: string[]): Promise<SyncBody | undefined> {
+		const opened = await this.#fs.openBinaryReader(
+			await this.#path(context, sessionId, ...parts),
+			undefined,
+			context,
+		);
+		if (isNotFound(opened)) return undefined;
+		const reader = unwrap(opened);
+		return (async function* () {
+			try {
+				for (let offset = 0; ; ) {
+					const chunk = unwrap(await reader.read(offset, READ_CHUNK, context));
+					if (chunk.length === 0) return;
+					offset += chunk.length;
+					yield chunk;
+				}
+			} finally {
+				await reader.close(context);
+			}
+		})();
+	}
+
+	/**
+	 * Write a body through a temporary file and a rename, so a reader never sees half a file; the body is appended chunk
+	 * by chunk as it streams.
+	 */
 	async #write(
 		sessionId: string,
 		directory: readonly string[],
 		name: string,
-		body: Uint8Array,
+		body: SyncBody,
 		context: Context,
 	): Promise<void> {
 		unwrap(
@@ -183,7 +217,13 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance {
 		);
 		const path = await this.#path(context, sessionId, ...directory, name);
 		const temporary = await this.#path(context, sessionId, ...directory, `${name}.tmp`);
-		unwrap(await this.#fs.writeFile(temporary, body, context));
+		unwrap(await this.#fs.writeFile(temporary, new Uint8Array(), context));
+		try {
+			for await (const chunk of body) unwrap(await this.#fs.appendFile(temporary, chunk, context));
+		} catch (error) {
+			await this.#fs.remove(temporary, { force: true }, context);
+			throw error;
+		}
 		unwrap(await this.#fs.renameFile(temporary, path, context));
 	}
 
diff --git a/packages/chord/src/sync/index.ts b/packages/chord/src/sync/index.ts
index 106cdeb64..64de26764 100644
--- a/packages/chord/src/sync/index.ts
+++ b/packages/chord/src/sync/index.ts
@@ -1,8 +1,17 @@
 // Session sync over a store's outbox (`@earendil-works/chord/sync`, experimental). Portable: the remote and the host's
 // sync state are interfaces, and the directory-backed ones take their file operations (`SyncFileSystem`).
 
-export { decodeBatch, decodeSnapshot, encodeBatch, encodeSnapshot } from "./batch.ts";
 export {
+	decodeBatch,
+	decodeBatchLines,
+	decodeSnapshot,
+	encodeBatch,
+	encodeSnapshot,
+	type SyncBody,
+} from "./batch.ts";
+export {
+	APPLY_BYTES,
+	BATCH_BYTES,
 	FlushScheduler,
 	type FlushSchedulerOptions,
 	foldSession,
@@ -17,9 +26,9 @@ export {
 	type UploadResult,
 	uploadSession,
 } from "./client.ts";
-export type { FileResult, SyncFileInfo, SyncFileSystem } from "./files.ts";
+export type { FileResult, SyncFileInfo, SyncFileReader, SyncFileSystem } from "./files.ts";
 export { checkSessionId, FileSystemSyncRemote } from "./filesystem.ts";
-export { decodeLogLines, encodeLogLine, LOG_VERSION, LogFormatError } from "./lines.ts";
+export { decodeLogLine, encodeLogLine, LOG_VERSION, LogFormatError } from "./lines.ts";
 export {
 	type BatchInfo,
 	HistoryUnavailableError,
diff --git a/packages/chord/src/sync/lines.ts b/packages/chord/src/sync/lines.ts
index a267a91ea..3860c58dd 100644
--- a/packages/chord/src/sync/lines.ts
+++ b/packages/chord/src/sync/lines.ts
@@ -27,29 +27,11 @@ export function encodeLogLine(commit: OutboxCommit): string {
 }
 
 /**
- * Decode JSON lines. A last line without its newline is a torn append and is returned as `torn` instead of a commit;
- * anything else that does not decode is an error. A line of an unknown version is an error too: an older pi must not
- * apply writes it does not understand. The writes themselves are checked by `applyBatch()`, a trust boundary.
+ * One line (without its newline) as an outbox commit; `number` names the line in errors. A line of an unknown version
+ * is an error: an older pi must not apply writes it does not understand. The writes themselves are checked by
+ * `applyBatch()`, a trust boundary.
  */
-export function decodeLogLines(text: string): { readonly commits: OutboxCommit[]; readonly torn: number } {
-	const commits: OutboxCommit[] = [];
-	let start = 0;
-	let previous = 0;
-	while (start < text.length) {
-		const end = text.indexOf("\n", start);
-		if (end === -1) return { commits, torn: text.length - start };
-		const commit = decodeLine(text.slice(start, end), commits.length + 1);
-		if (commit.seq <= previous) {
-			throw new LogFormatError(`Log line ${commits.length + 1} does not increase the sequence`);
-		}
-		previous = commit.seq;
-		commits.push(commit);
-		start = end + 1;
-	}
-	return { commits, torn: 0 };
-}
-
-function decodeLine(text: string, number: number): OutboxCommit {
+export function decodeLogLine(text: string, number: number): OutboxCommit {
 	let value: unknown;
 	try {
 		value = JSON.parse(text);
diff --git a/packages/chord/src/sync/remote.ts b/packages/chord/src/sync/remote.ts
index b32ccdddd..773f48dc8 100644
--- a/packages/chord/src/sync/remote.ts
+++ b/packages/chord/src/sync/remote.ts
@@ -1,4 +1,5 @@
 import type { Context } from "../context/index.ts";
+import type { SyncBody } from "./batch.ts";
 
 /** A store's commit sequence (`store.lastSeq`, an outbox commit's `seq`); 0 is before every commit. */
 export type Seq = number;
@@ -62,19 +63,22 @@ export type SessionChanges = {
 
 /** The client's view of a remote store. Clients never delete or fold. */
 export interface SessionSyncStore {
-	/** Create the slot after `after` unless it exists or the history does not end at `after`. */
+	/**
+	 * Create the slot after `after` unless it exists or the history does not end at `after`. `body` is a stream read once;
+	 * a remote that answers without reading it (`exists`, `not_head`) leaves it unread.
+	 */
 	putBatch(
 		sessionId: string,
 		after: Seq,
 		to: Seq,
 		device: string,
-		body: Uint8Array,
+		body: SyncBody,
 		context: Context,
 		options?: { readonly archive?: boolean },
 	): Promise<PutBatchResult>;
 	/** A batch's body; `undefined` when its slot is folded or absent. */
-	getBatch(sessionId: string, after: Seq, context: Context): Promise<Uint8Array | undefined>;
-	getSnapshot(sessionId: string, to: Seq, context: Context): Promise<Uint8Array | undefined>;
+	getBatch(sessionId: string, after: Seq, context: Context): Promise<SyncBody | undefined>;
+	getSnapshot(sessionId: string, to: Seq, context: Context): Promise<SyncBody | undefined>;
 	list(sessionId: string, context: Context): Promise<RemoteListing>;
 	/**
 	 * The sessions changed after `cursor` (all of them without one). A remote may also list sessions that did not change,
@@ -91,7 +95,7 @@ export interface SessionSyncStore {
 
 /** The remote's own side: folding history into snapshots, which clients never do. */
 export interface SessionSyncMaintenance extends SessionSyncStore {
-	putSnapshot(sessionId: string, to: Seq, body: Uint8Array, context: Context): Promise<void>;
+	putSnapshot(sessionId: string, to: Seq, body: SyncBody, context: Context): Promise<void>;
 	/** Drop the bodies of these slots, keeping the slots taken so nobody can create them again. */
 	foldBatches(sessionId: string, afters: readonly Seq[], context: Context): Promise<void>;
 	deleteSnapshots(sessionId: string, snapshots: readonly Seq[], context: Context): Promise<void>;
diff --git a/packages/chord/test/bench/sync-memory.ts b/packages/chord/test/bench/sync-memory.ts
new file mode 100644
index 000000000..fd4daed52
--- /dev/null
+++ b/packages/chord/test/bench/sync-memory.ts
@@ -0,0 +1,169 @@
+// Memory benchmark of the sync layer inside Durable Objects (RFC 0074's budget: 96 to 128 MB per isolate): workerd
+// through Miniflare, the objects of `sync-worker.ts` on their own SQLite storage, and the isolate's V8 heap sampled
+// through the inspector while each step runs. Steps per session size: fill the origin's store (commits with its
+// outbox), upload (outbox to remote, bodies streamed into the remote's SQLite), pull (restore a replica through the
+// origin's remote over RPC), fold (restore into scratch, export a snapshot back). Each step starts after a full GC; its
+// baseline is the used heap then, and its peak the highest `usedSize` sampled until it answers.
+//
+// workerd runs with `--expose-gc` and `--max-old-space-size=128` (Cloudflare's per-isolate limit), so a step that
+// needed more than 128 MB would fail, not just report a big number.
+//
+// Run by hand from packages/chord (not in CI; about three minutes):
+//   node --conditions=source --experimental-strip-types test/bench/sync-memory.ts
+
+import { build } from "esbuild";
+import { convertV4MiniflareOptions, Miniflare } from "miniflare";
+import WebSocket from "ws";
+
+/** Session histories: total MB of row text, and the size of each commit's row. */
+const RUNS = [
+	{ sizeMb: 8, rowBytes: 64 * 1024 },
+	{ sizeMb: 32, rowBytes: 64 * 1024 },
+	{ sizeMb: 128, rowBytes: 64 * 1024 },
+	{ sizeMb: 256, rowBytes: 64 * 1024 },
+	// Few large commits: each one close to a whole batch, the case that holds the most at once.
+	{ sizeMb: 64, rowBytes: 4 * 1024 * 1024 },
+];
+const INSPECTOR_PORT = 9341;
+const SAMPLE_MS = 10;
+const MB = 1024 * 1024;
+
+process.env.MINIFLARE_WORKERD_V8_FLAGS = "--expose-gc --max-old-space-size=128";
+
+const bundled = await build({
+	entryPoints: [new URL("./sync-worker.ts", import.meta.url).pathname],
+	bundle: true,
+	format: "esm",
+	platform: "neutral",
+	write: false,
+	external: ["cloudflare:*"],
+	conditions: ["source", "workerd", "worker"],
+	mainFields: ["module", "main"],
+	target: "es2023",
+});
+
+const mf = new Miniflare(
+	convertV4MiniflareOptions({
+		inspectorPort: INSPECTOR_PORT,
+		workers: [
+			{
+				name: "sync-bench",
+				compatibilityDate: "2026-10-01",
+				compatibilityFlags: ["nodejs_compat"],
+				modules: [{ type: "ESModule", path: "worker.mjs", contents: bundled.outputFiles[0]!.text }],
+				durableObjects: {
+					ORIGIN: { className: "Origin", useSQLite: true },
+					REPLICA: { className: "Replica", useSQLite: true },
+					FOLDER: { className: "Folder", useSQLite: true },
+				},
+			},
+		],
+	}),
+);
+
+try {
+	await mf.ready;
+	await mf.dispatchFetch("http://bench/unknown");
+	const [target] = (await (await fetch(`http://127.0.0.1:${INSPECTOR_PORT}/json/list`)).json()) as {
+		webSocketDebuggerUrl: string;
+	}[];
+	const inspector = await connect(target!.webSocketDebuggerUrl);
+
+	const rows: string[] = [];
+	for (const { sizeMb, rowBytes } of RUNS) {
+		const run = `run${sizeMb}x${rowBytes}`;
+		const count = (sizeMb * MB) / rowBytes;
+		const label = `${sizeMb} MB in ${mb(rowBytes)} MB rows`;
+		for (const step of ["fill", "upload", "pull", "fold"] as const) {
+			const query = step === "fill" ? `&count=${count}&size=${rowBytes}` : "";
+			const measured = await measure(inspector, () => mf.dispatchFetch(`http://bench/${step}?run=${run}${query}`));
+			const body = (await measured.response.json()) as { ms: number; result: unknown };
+			if (!measured.response.ok) throw new Error(`${step} ${label} failed: ${JSON.stringify(body)}`);
+			rows.push(
+				`| ${label} | ${step} | ${mb(measured.baseline)} | ${mb(measured.peak)} | ${mb(measured.peak - measured.baseline)} | ${body.ms} | ${summary(step, body.result)} |`,
+			);
+			console.log(rows.at(-1));
+		}
+	}
+	inspector.close();
+	console.log(
+		[
+			"",
+			"| history | step | heap before (MB) | peak heap (MB) | peak above before (MB) | ms | result |",
+			"|---|---|---|---|---|---|---|",
+			...rows,
+		].join("\n"),
+	);
+} finally {
+	await mf.dispose();
+}
+
+/** What a step's answer must offer (Miniflare's own `Response` type, not the DOM one). */
+type StepResponse = { readonly ok: boolean; json(): Promise<unknown>; text(): Promise<string> };
+
+type Inspector = { call(method: string): Promise<{ result?: { usedSize?: number } }>; close(): void };
+
+async function connect(url: string): Promise<Inspector> {
+	const ws = new WebSocket(url);
+	await new Promise((resolve, reject) => {
+		ws.once("open", resolve);
+		ws.once("error", reject);
+	});
+	let id = 0;
+	const pending = new Map<number, (value: never) => void>();
+	ws.on("message", (raw) => {
+		const message = JSON.parse(String(raw)) as { id?: number };
+		if (message.id !== undefined) pending.get(message.id)?.(message as never);
+		pending.delete(message.id ?? -1);
+	});
+	const inspector: Inspector = {
+		call: (method) =>
+			new Promise((resolve) => {
+				const n = ++id;
+				pending.set(n, resolve as (value: never) => void);
+				ws.send(JSON.stringify({ id: n, method }));
+			}),
+		close: () => ws.close(),
+	};
+	await inspector.call("Runtime.enable");
+	return inspector;
+}
+
+/**
+ * Runs `step` from a settled heap: a full GC in the isolate first, its used heap then as the baseline, then samples
+ * every `SAMPLE_MS` until the step answers; the peak is the highest sample.
+ */
+async function measure(
+	inspector: Inspector,
+	step: () => Promise<StepResponse>,
+): Promise<{ response: StepResponse; baseline: number; peak: number }> {
+	const used = async () => (await inspector.call("Runtime.getHeapUsage")).result?.usedSize ?? 0;
+	await (await mf.dispatchFetch("http://bench/gc")).text();
+	const baseline = await used();
+	let peak = baseline;
+	let running = true;
+	const sampling = (async () => {
+		while (running) {
+			peak = Math.max(peak, await used());
+			await new Promise((resolve) => setTimeout(resolve, SAMPLE_MS));
+		}
+	})();
+	const response = await step();
+	running = false;
+	await sampling;
+	peak = Math.max(peak, await used());
+	return { response, baseline, peak };
+}
+
+function mb(bytes: number): string {
+	return (bytes / MB).toFixed(1);
+}
+
+function summary(step: string, result: unknown): string {
+	const value = result as Record<string, unknown>;
+	if (step === "upload") {
+		const stored = value.stored as number;
+		return `${value.slots} slots, ${mb(stored)} MB stored`;
+	}
+	return JSON.stringify(result);
+}
diff --git a/packages/chord/test/bench/sync-worker.ts b/packages/chord/test/bench/sync-worker.ts
new file mode 100644
index 000000000..3c9eb13dd
--- /dev/null
+++ b/packages/chord/test/bench/sync-worker.ts
@@ -0,0 +1,449 @@
+// The Durable Objects of the sync memory benchmark (`sync-memory.ts`), one Chord store each on the object's own SQLite
+// storage (`chord/sqlite/cloudflare`), as a deployment has them:
+//   - `Origin`: the copy that commits and uploads, and the remote, its slots and bodies in its SQLite (`SqlRemote`);
+//   - `Replica`: a copy restored by a pull, through the origin's remote over RPC;
+//   - `Folder`: folds the origin's history into a snapshot, its scratch store on its own SQLite.
+// Bodies cross objects as byte streams and are stored in parts, so what an object holds is what the sync layer holds.
+// The benchmark drives the objects over `fetch` and reads the isolate's heap through the inspector.
+
+import { DurableObject } from "cloudflare:workers";
+import { BACKGROUND_CONTEXT as context } from "../../src/context/index.ts";
+import { int, str } from "../../src/contracts/index.ts";
+import { DurableObjectSqliteDatabase } from "../../src/sqlite/cloudflare.ts";
+import { sqliteBackend } from "../../src/store/backend.ts";
+import type { StagedWrite } from "../../src/store/step.ts";
+import { type ChordStore, openStore } from "../../src/store/store.ts";
+import { defineTable } from "../../src/store/table.ts";
+import { TableChanges } from "../../src/store/table-changes.ts";
+import { TableHandle } from "../../src/store/table-handle.ts";
+import { prepareTables } from "../../src/store/table-migration.ts";
+import {
+	type BatchInfo,
+	foldSession,
+	type JsonObject,
+	type PutBatchResult,
+	pullSession,
+	type RemoteListing,
+	remoteHead,
+	type Seq,
+	type SessionChanges,
+	type SessionSyncMaintenance,
+	type SyncBody,
+	type SyncState,
+	type SyncStateStore,
+	uploadSession,
+} from "../../src/sync/index.ts";
+
+// The parts of the Workers runtime this bench uses, typed here: full Workers types would clash with the repository's
+// Node types, and `DurableObject` (packages/codemode/src/cloudflare-modules.d.ts) leaves `ctx` untyped.
+type SqlValue = string | number | null | ArrayBuffer | Uint8Array;
+type SqlCursor<T> = { toArray(): T[]; one(): T };
+type SqlStorage = { exec<T = Record<string, SqlValue>>(query: string, ...bindings: SqlValue[]): SqlCursor<T> };
+type ObjectStorage = {
+	readonly sql: SqlStorage;
+	transaction<T>(closure: () => Promise<T>): Promise<T>;
+	transactionSync<T>(closure: () => T): T;
+};
+type ObjectState = { readonly storage: ObjectStorage };
+type ObjectId = unknown;
+type Namespace<T> = { idFromName(name: string): ObjectId; get(id: ObjectId): T };
+
+/** A Durable Object of this bench: the runtime's base class with its `ctx` typed. */
+const BenchObject = DurableObject as unknown as abstract new (
+	ctx: ObjectState,
+	env: Env,
+) => { readonly ctx: ObjectState; readonly env: Env };
+
+type Env = {
+	readonly ORIGIN: Namespace<Origin>;
+	readonly REPLICA: Namespace<Replica>;
+	readonly FOLDER: Namespace<Folder>;
+};
+
+const Rows = defineTable("rows", { version: 1, columns: { id: str(), n: int(), text: str() }, key: ["id"] });
+const TABLE = "bench.sync:rows";
+const OPEN = { admit: () => true, adopt: () => {} };
+const owner = { label: "bench", scope: {}, assertLive: () => {}, fail: () => {} };
+/** Where a stored body is split into rows: below the object's 2 MB value limit. */
+const PART = 512 * 1024;
+
+/** The isolate's own GC (the benchmark starts workerd with `--expose-gc`), so a step starts from a settled heap. */
+const collect = (): void => (globalThis as { gc?: () => void }).gc?.();
+
+/** A body as a byte stream, for RPC. */
+function toStream(body: SyncBody): ReadableStream<Uint8Array> {
+	const iterator = body[Symbol.asyncIterator]();
+	return new ReadableStream<Uint8Array>({
+		async pull(controller) {
+			const next = await iterator.next();
+			if (next.done) controller.close();
+			else controller.enqueue(next.value);
+		},
+		async cancel() {
+			await iterator.return?.();
+		},
+	});
+}
+
+/** A byte stream from RPC as a body. */
+async function* fromStream(stream: ReadableStream<Uint8Array>): SyncBody {
+	const reader = stream.getReader();
+	try {
+		while (true) {
+			const next = await reader.read();
+			if (next.done) return;
+			yield next.value;
+		}
+	} finally {
+		reader.releaseLock();
+	}
+}
+
+type SlotRow = { after: number; upto: number; device: string; archive: number; folded: number };
+
+/**
+ * A remote in an object's SQLite storage, in tables of its own beside the store's: slots, and bodies written and read
+ * back a part at a time as they stream.
+ */
+class SqlRemote implements SessionSyncMaintenance {
+	readonly #sql: SqlStorage;
+
+	constructor(sql: SqlStorage) {
+		this.#sql = sql;
+		sql.exec(`CREATE TABLE IF NOT EXISTS r_slots (after INTEGER PRIMARY KEY, upto INTEGER, device TEXT, archive INTEGER, folded INTEGER);
+			CREATE TABLE IF NOT EXISTS r_parts (kind TEXT, at INTEGER, part INTEGER, data BLOB, PRIMARY KEY (kind, at, part));
+			CREATE TABLE IF NOT EXISTS r_snapshots (upto INTEGER PRIMARY KEY);`);
+	}
+
+	async putBatch(
+		_session: string,
+		after: Seq,
+		to: Seq,
+		device: string,
+		body: SyncBody,
+		_context: unknown,
+		options: { readonly archive?: boolean } = {},
+	): Promise<PutBatchResult> {
+		const existing = this.#sql
+			.exec<{ upto: number; device: string }>("SELECT upto, device FROM r_slots WHERE after = ?", after)
+			.toArray()[0];
+		if (existing) return { type: "exists", to: existing.upto, device: existing.device };
+		const head = remoteHead(await this.list());
+		if (after !== head) return { type: "not_head", head };
+		await this.#write("batch", after, body);
+		this.#sql.exec("INSERT INTO r_slots VALUES (?, ?, ?, ?, 0)", after, to, device, options.archive ? 1 : 0);
+		return { type: "created" };
+	}
+
+	async getBatch(_session: string, after: Seq): Promise<SyncBody | undefined> {
+		const slot = this.#sql.exec<{ folded: number }>("SELECT folded FROM r_slots WHERE after = ?", after).toArray()[0];
+		return slot && !slot.folded ? this.#read("batch", after) : undefined;
+	}
+
+	async getSnapshot(_session: string, to: Seq): Promise<SyncBody | undefined> {
+		return this.#sql.exec("SELECT 1 FROM r_snapshots WHERE upto = ?", to).toArray().length > 0
+			? this.#read("snapshot", to)
+			: undefined;
+	}
+
+	async putSnapshot(_session: string, to: Seq, body: SyncBody): Promise<void> {
+		await this.#write("snapshot", to, body);
+		this.#sql.exec("INSERT OR REPLACE INTO r_snapshots VALUES (?)", to);
+	}
+
+	async foldBatches(_session: string, afters: readonly Seq[]): Promise<void> {
+		for (const after of afters) {
+			this.#sql.exec("UPDATE r_slots SET folded = 1 WHERE after = ?", after);
+			this.#sql.exec("DELETE FROM r_parts WHERE kind = 'batch' AND at = ?", after);
+		}
+	}
+
+	async deleteSnapshots(_session: string, snapshots: readonly Seq[]): Promise<void> {
+		for (const to of snapshots) {
+			this.#sql.exec("DELETE FROM r_snapshots WHERE upto = ?", to);
+			this.#sql.exec("DELETE FROM r_parts WHERE kind = 'snapshot' AND at = ?", to);
+		}
+	}
+
+	async list(): Promise<RemoteListing> {
+		const batches = this.#sql
+			.exec<SlotRow>("SELECT after, upto, device, archive, folded FROM r_slots ORDER BY after")
+			.toArray()
+			.map(
+				(slot): BatchInfo => ({
+					after: slot.after,
+					to: slot.upto,
+					device: slot.device,
+					...(slot.archive ? { archive: true as const } : {}),
+					...(slot.folded ? { folded: true as const } : {}),
+				}),
+			);
+		const snapshots = this.#sql
+			.exec<{ upto: number }>("SELECT upto FROM r_snapshots")
+			.toArray()
+			.map((row) => row.upto);
+		return { batches, snapshots };
+	}
+
+	async sessions(): Promise<SessionChanges> {
+		return { sessions: [], cursor: "", more: false };
+	}
+	async putMeta(): Promise<void> {}
+	async getMeta(): Promise<JsonObject | undefined> {
+		return undefined;
+	}
+
+	/** Bytes the remote stores, all bodies together. */
+	stored(): number {
+		return this.#sql.exec<{ n: number }>("SELECT coalesce(sum(length(data)), 0) AS n FROM r_parts").one().n;
+	}
+
+	/** The body's chunks gathered into parts of `PART` bytes, one row each, as it streams. */
+	async #write(kind: string, at: number, body: SyncBody): Promise<void> {
+		this.#sql.exec("DELETE FROM r_parts WHERE kind = ? AND at = ?", kind, at);
+		let part = 0;
+		let pending: Uint8Array[] = [];
+		let size = 0;
+		const flush = () => {
+			const data = new Uint8Array(size);
+			let offset = 0;
+			for (const chunk of pending) {
+				data.set(chunk, offset);
+				offset += chunk.length;
+			}
+			this.#sql.exec("INSERT INTO r_parts VALUES (?, ?, ?, ?)", kind, at, part++, data);
+			pending = [];
+			size = 0;
+		};
+		for await (const chunk of body) {
+			pending.push(chunk);
+			size += chunk.length;
+			if (size >= PART) flush();
+		}
+		if (size > 0) flush();
+	}
+
+	/** The body's parts, read back one row at a time. */
+	#read(kind: string, at: number): SyncBody {
+		const sql = this.#sql;
+		return (async function* () {
+			for (let part = 0; ; part++) {
+				const row = sql
+					.exec<{ data: ArrayBuffer }>(
+						"SELECT data FROM r_parts WHERE kind = ? AND at = ? AND part = ?",
+						kind,
+						at,
+						part,
+					)
+					.toArray()[0];
+				if (!row) return;
+				yield new Uint8Array(row.data);
+			}
+		})();
+	}
+}
+
+/** The origin's remote seen from another object: its calls over RPC, bodies as byte streams. */
+class RpcRemote implements SessionSyncMaintenance {
+	readonly #origin: Origin;
+
+	constructor(origin: Origin) {
+		this.#origin = origin;
+	}
+
+	async putBatch(): Promise<PutBatchResult> {
+		throw new Error("The benchmark uploads only at the origin");
+	}
+	async getBatch(_session: string, after: Seq): Promise<SyncBody | undefined> {
+		const stream = await this.#origin.getBatch(after);
+		return stream === null ? undefined : fromStream(stream);
+	}
+	async getSnapshot(_session: string, to: Seq): Promise<SyncBody | undefined> {
+		const stream = await this.#origin.getSnapshot(to);
+		return stream === null ? undefined : fromStream(stream);
+	}
+	async putSnapshot(_session: string, to: Seq, body: SyncBody): Promise<void> {
+		await this.#origin.putSnapshot(to, toStream(body));
+	}
+	async foldBatches(_session: string, afters: readonly Seq[]): Promise<void> {
+		await this.#origin.foldBatches([...afters]);
+	}
+	async deleteSnapshots(_session: string, snapshots: readonly Seq[]): Promise<void> {
+		await this.#origin.deleteSnapshots([...snapshots]);
+	}
+	async list(): Promise<RemoteListing> {
+		return (await this.#origin.list()) as RemoteListing;
+	}
+	async sessions(): Promise<SessionChanges> {
+		return { sessions: [], cursor: "", more: false };
+	}
+	async putMeta(): Promise<void> {}
+	async getMeta(): Promise<JsonObject | undefined> {
+		return undefined;
+	}
+}
+
+class MemoryState implements SyncStateStore {
+	#state: SyncState | undefined;
+	async get(): Promise<SyncState | undefined> {
+		return this.#state;
+	}
+	async set(_session: string, state: SyncState): Promise<void> {
+		this.#state = state;
+	}
+}
+
+/**
+ * This object's SQLite storage as a store backend. Durable Object SQLite refuses `PRAGMA encoding` and
+ * `pragma_encoding` (`SQLITE_AUTH`), which `openStore` reads for its UTF-8 check; its databases are always UTF-8, so
+ * the benchmark answers that one query itself. (A store bug, to fix in `openStore` or the Cloudflare adapter.)
+ */
+function backendOf(storage: ObjectStorage) {
+	const rewrite = (query: string) =>
+		query
+			.replace(/^PRAGMA encoding$/, "SELECT 'UTF-8' AS encoding")
+			.replace(/pragma_encoding e\b/, "(SELECT 'UTF-8' AS encoding) e");
+	const sql = storage.sql;
+	const wrapped = {
+		sql: { exec: (query: string, ...bindings: SqlValue[]) => sql.exec(rewrite(query), ...bindings) },
+		transaction: <T>(closure: () => Promise<T>) => storage.transaction(closure),
+		transactionSync: <T>(closure: () => T) => storage.transactionSync(closure),
+	};
+	return sqliteBackend(new DurableObjectSqliteDatabase(wrapped as never));
+}
+
+/** Deterministic text that compresses about as transcript text does (about 2:1), not to nothing. */
+function text(size: number, seed: number): string {
+	let state = seed * 2654435761 + 1;
+	const words = ["the", "agent", "reads", "a", "file", "and", "writes", "tests", "for", "it", "then", "runs"];
+	let out = "";
+	while (out.length < size) {
+		state ^= state << 13;
+		state ^= state >>> 17;
+		state ^= state << 5;
+		out += `${words[(state >>> 0) % words.length]} ${(state >>> 8) & 0xffff} `;
+	}
+	return out.slice(0, size);
+}
+
+export class Origin extends BenchObject {
+	#remote = new SqlRemote(this.ctx.storage.sql);
+
+	/** `count` commits of one row of `size` characters each. */
+	async fill(count: number, size: number): Promise<{ lastSeq: number }> {
+		collect();
+		const store = await openStore(backendOf(this.ctx.storage), context, { outbox: true });
+		try {
+			const chord = store as ChordStore;
+			await prepareTables([{ definition: Rows, table: TABLE }], chord, OPEN, "bench.sync", context);
+			const handle = new TableHandle(Rows, TABLE, chord, owner);
+			for (let index = 0; index < count; index++) {
+				const staged: StagedWrite[] = [];
+				new TableChanges(handle, staged).put({ id: `r${index}`, n: index, text: text(size, index) });
+				await chord.commit(staged, OPEN, context);
+			}
+			return { lastSeq: store.lastSeq };
+		} finally {
+			await store.close(context);
+		}
+	}
+
+	async upload(): Promise<unknown> {
+		collect();
+		const store = await openStore(backendOf(this.ctx.storage), context, { outbox: true });
+		try {
+			const result = await uploadSession({
+				sessionId: "s",
+				device: "origin",
+				source: store,
+				store: this.#remote,
+				state: new MemoryState(),
+				context,
+			});
+			return { result, slots: (await this.#remote.list()).batches.length, stored: this.#remote.stored() };
+		} finally {
+			await store.close(context);
+		}
+	}
+
+	async getBatch(after: number): Promise<ReadableStream<Uint8Array> | null> {
+		const body = await this.#remote.getBatch("s", after);
+		return body === undefined ? null : toStream(body);
+	}
+	async getSnapshot(to: number): Promise<ReadableStream<Uint8Array> | null> {
+		const body = await this.#remote.getSnapshot("s", to);
+		return body === undefined ? null : toStream(body);
+	}
+	async putSnapshot(to: number, stream: ReadableStream<Uint8Array>): Promise<void> {
+		await this.#remote.putSnapshot("s", to, fromStream(stream));
+	}
+	async foldBatches(afters: number[]): Promise<void> {
+		await this.#remote.foldBatches("s", afters);
+	}
+	async deleteSnapshots(snapshots: number[]): Promise<void> {
+		await this.#remote.deleteSnapshots("s", snapshots);
+	}
+	async list(): Promise<RemoteListing> {
+		return this.#remote.list();
+	}
+}
+
+export class Replica extends BenchObject {
+	/** Restores the session from the origin's remote into this object's empty storage. */
+	async pull(origin: string): Promise<unknown> {
+		collect();
+		const remote = new RpcRemote(this.env.ORIGIN.get(this.env.ORIGIN.idFromName(origin)));
+		const pulled = await pullSession({
+			sessionId: "s",
+			store: remote,
+			local: { kind: "empty", backend: backendOf(this.ctx.storage) },
+			state: new MemoryState(),
+			context,
+		});
+		await pulled.store.close(context);
+		return { head: pulled.head };
+	}
+}
+
+export class Folder extends BenchObject {
+	/** Folds the origin's history into a snapshot, restoring it into this object's empty storage as scratch. */
+	async fold(origin: string): Promise<unknown> {
+		collect();
+		const remote = new RpcRemote(this.env.ORIGIN.get(this.env.ORIGIN.idFromName(origin)));
+		return {
+			folded: await foldSession({ sessionId: "s", store: remote, scratch: backendOf(this.ctx.storage), context }),
+		};
+	}
+}
+
+export default {
+	async fetch(request: Request, env: Env): Promise<Response> {
+		const url = new URL(request.url);
+		const run = url.searchParams.get("run") ?? "bench";
+		const count = Number(url.searchParams.get("count") ?? 0);
+		const size = Number(url.searchParams.get("size") ?? 0);
+		const started = Date.now();
+		let result: unknown;
+		switch (url.pathname) {
+			case "/gc":
+				// Every object of the benchmark shares this isolate: one full GC here settles the heap for the next step.
+				collect();
+				return new Response("ok");
+			case "/fill":
+				result = await env.ORIGIN.get(env.ORIGIN.idFromName(run)).fill(count, size);
+				break;
+			case "/upload":
+				result = await env.ORIGIN.get(env.ORIGIN.idFromName(run)).upload();
+				break;
+			case "/pull":
+				result = await env.REPLICA.get(env.REPLICA.idFromName(run)).pull(run);
+				break;
+			case "/fold":
+				result = await env.FOLDER.get(env.FOLDER.idFromName(run)).fold(run);
+				break;
+			default:
+				return new Response("unknown step", { status: 404 });
+		}
+		return Response.json({ ms: Date.now() - started, result });
+	},
+};
diff --git a/packages/chord/test/sync/lines.test.ts b/packages/chord/test/sync/lines.test.ts
index 8e15d07b2..b1690758f 100644
--- a/packages/chord/test/sync/lines.test.ts
+++ b/packages/chord/test/sync/lines.test.ts
@@ -1,16 +1,20 @@
-// The sync wire format: outbox commits as versioned JSON lines, and snapshots as store archives.
+// The sync wire format, as streams: outbox commits as versioned, gzip-compressed JSON lines, snapshots as store
+// archives, and decoding that holds one line at a time.
 import { describe, expect, it } from "vitest";
 import { BACKGROUND_CONTEXT as context } from "../../src/context/index.ts";
 import { createDocument } from "../../src/store/engine.ts";
 import type { ChordStore } from "../../src/store/store.ts";
 import {
-	decodeLogLines,
+	decodeBatch,
+	decodeLogLine,
 	decodeSnapshot,
+	encodeBatch,
 	encodeLogLine,
 	encodeSnapshot,
 	LOG_START,
 	LogFormatError,
 	readOutbox,
+	type SyncBody,
 } from "../../src/sync/index.ts";
 import { memoryStore } from "../support.ts";
 
@@ -36,44 +40,107 @@ async function seeded(): Promise<ChordStore> {
 	return store;
 }
 
-describe("log lines", () => {
-	it("round-trips outbox commits", async () => {
+async function collect<T>(items: AsyncIterable<T>): Promise<T[]> {
+	const out: T[] = [];
+	for await (const item of items) out.push(item);
+	return out;
+}
+
+/** A body's bytes, joined. */
+async function bytesOf(body: SyncBody): Promise<Uint8Array> {
+	const chunks = await collect(body);
+	const out = new Uint8Array(chunks.reduce((size, chunk) => size + chunk.length, 0));
+	let offset = 0;
+	for (const chunk of chunks) {
+		out.set(chunk, offset);
+		offset += chunk.length;
+	}
+	return out;
+}
+
+/** Plain bytes gzip-compressed, as a body: for bodies the encoders would never write. */
+function gzipped(text: string): SyncBody {
+	return (async function* () {
+		const stream = new Blob([text]).stream().pipeThrough(new CompressionStream("gzip"));
+		yield new Uint8Array(await new Response(stream).arrayBuffer());
+	})();
+}
+
+/** `bytes` as a body of chunks of `size`: decoding must not depend on where chunks split. */
+async function* chunked(bytes: Uint8Array, size: number): SyncBody {
+	for (let offset = 0; offset < bytes.length; offset += size) yield bytes.subarray(offset, offset + size);
+}
+
+describe("batch bodies", () => {
+	it("round-trip outbox commits, whatever the chunks", async () => {
 		const store = await seeded();
 		try {
 			const commits = await readOutbox(store, LOG_START, 100, context);
 			expect(commits).toHaveLength(2);
-			expect(decodeLogLines(commits.map(encodeLogLine).join(""))).toEqual({ commits, torn: 0 });
+			const bytes = await bytesOf(encodeBatch(commits));
+			expect(bytes[0]).toBe(0x1f);
+			for (const size of [1, 7, bytes.length]) {
+				expect(await collect(decodeBatch(chunked(bytes, size)))).toEqual(commits);
+			}
 		} finally {
 			await store.close(context);
 		}
 	});
 
-	it("reports a torn last line instead of failing", () => {
+	it("refuse a body that ends inside a line", async () => {
 		const line = encodeLogLine({ seq: 1, nextId: 2, writes: [] });
-		expect(decodeLogLines(line + line.slice(0, 5))).toEqual({
-			commits: [{ seq: 1, nextId: 2, writes: [] }],
-			torn: 5,
-		});
+		await expect(collect(decodeBatch(gzipped(line + line.slice(0, 5))))).rejects.toThrow("ends inside log line 2");
 	});
 
-	it("refuses unknown versions, a missing next id, and sequences that do not increase", () => {
-		expect(() => decodeLogLines('{"v":1,"seq":1,"nextId":2,"writes":[]}\n')).toThrow(LogFormatError);
-		expect(() => decodeLogLines('{"v":2,"seq":1,"writes":[]}\n')).toThrow(LogFormatError);
+	it("refuse sequences that do not increase", async () => {
 		const line = encodeLogLine({ seq: 3, nextId: 4, writes: [] });
-		expect(() => decodeLogLines(line + line)).toThrow("does not increase");
+		await expect(collect(decodeBatch(gzipped(line + line)))).rejects.toThrow("does not increase");
+	});
+
+	it("decode a commit only when it is read", async () => {
+		const lines = [1, 2, 3].map((seq) => encodeLogLine({ seq, nextId: seq + 1, writes: [] })).join("");
+		const commits = decodeBatch(gzipped(`${lines}not json\n`));
+		const iterator = commits[Symbol.asyncIterator]();
+		expect((await iterator.next()).value).toMatchObject({ seq: 1 });
+		await iterator.return?.(undefined);
 	});
 });
 
-describe("snapshots", () => {
+describe("log lines", () => {
+	it("refuse unknown versions and a missing next id", () => {
+		expect(() => decodeLogLine('{"v":1,"seq":1,"nextId":2,"writes":[]}', 1)).toThrow(LogFormatError);
+		expect(() => decodeLogLine('{"v":2,"seq":1,"writes":[]}', 1)).toThrow(LogFormatError);
+		expect(decodeLogLine('{"v":2,"seq":1,"nextId":2,"writes":[]}', 1)).toEqual({ seq: 1, nextId: 2, writes: [] });
+	});
+});
+
+describe("snapshot bodies", () => {
 	it("round-trip a store archive", async () => {
 		const store = await seeded();
 		try {
-			const parts: unknown[] = [];
-			for await (const part of store.export(context)) parts.push(part);
-			expect(await decodeSnapshot(await encodeSnapshot(store.export(context)))).toEqual(parts);
+			const parts = await collect(store.export(context));
+			expect(await collect(decodeSnapshot(encodeSnapshot(store.export(context))))).toEqual(parts);
 			expect(parts[0]).toMatchObject({ kind: "manifest", lastSeq: store.lastSeq });
 		} finally {
 			await store.close(context);
 		}
 	});
+
+	it("give the store's commit queue back when the body is not read to the end", async () => {
+		const store = await seeded();
+		try {
+			const body = encodeSnapshot(store.export(context))[Symbol.asyncIterator]();
+			await body.next();
+			await body.return?.();
+			const base = { version: 1, fingerprint: "f1", value: { key: "c" } };
+			const committed = store.commit(
+				[createDocument({ namespace: "chord", owner: "app.sync", kind: "note", key: "c" }, base, "{}")],
+				OPEN,
+				context,
+			);
+			await expect(committed).resolves.toBeGreaterThan(0);
+		} finally {
+			await store.close(context);
+		}
+	});
 });
diff --git a/packages/chord/test/sync/node-files.ts b/packages/chord/test/sync/node-files.ts
index 3c5ae0ed1..bd1faa4ec 100644
--- a/packages/chord/test/sync/node-files.ts
+++ b/packages/chord/test/sync/node-files.ts
@@ -1,9 +1,9 @@
 // `SyncFileSystem` over Node's file system, for the sync tests (a host passes its own environment, such as Pi Durable's
 // `NodeExecutionEnv`).
 
-import { mkdir, readdir, readFile, rename, rm, writeFile } from "node:fs/promises";
+import { appendFile, mkdir, open, readdir, readFile, rename, rm, writeFile } from "node:fs/promises";
 import { join } from "node:path";
-import type { FileResult, SyncFileInfo, SyncFileSystem } from "../../src/sync/index.ts";
+import type { FileResult, SyncFileInfo, SyncFileReader, SyncFileSystem } from "../../src/sync/index.ts";
 
 /** An error with a `code`: `not_found` for a missing file or directory, as `SyncFileSystem` reports it. */
 class FileError extends Error {
@@ -27,7 +27,23 @@ export const nodeFiles: SyncFileSystem = {
 	joinPath: async (parts) => ({ ok: true, value: join(...parts) }),
 	readTextFile: (path) => attempt(() => readFile(path, "utf8")),
 	readBinaryFile: (path) => attempt(async () => new Uint8Array(await readFile(path))),
+	openBinaryReader: (path) =>
+		attempt(async (): Promise<SyncFileReader> => {
+			const file = await open(path, "r");
+			return {
+				read: (offset, length) =>
+					attempt(async () => {
+						const buffer = new Uint8Array(length);
+						const { bytesRead } = await file.read(buffer, 0, length, offset);
+						return buffer.subarray(0, bytesRead);
+					}),
+				close: async () => {
+					await file.close();
+				},
+			};
+		}),
 	writeFile: (path, content) => attempt(() => writeFile(path, content)),
+	appendFile: (path, content) => attempt(() => appendFile(path, content)),
 	renameFile: (source, destination) => attempt(() => rename(source, destination)),
 	listDir: (path) =>
 		attempt(async () =>
diff --git a/packages/chord/test/sync/sync.test.ts b/packages/chord/test/sync/sync.test.ts
index ae995b5ec..c8c416049 100644
--- a/packages/chord/test/sync/sync.test.ts
+++ b/packages/chord/test/sync/sync.test.ts
@@ -6,7 +6,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { afterEach, describe, expect, it } from "vitest";
 import { BACKGROUND_CONTEXT, type Context } from "../../src/context/index.ts";
-import { int, str } from "../../src/contracts/index.ts";
+import { int, opt, str } from "../../src/contracts/index.ts";
 import { openNodeSqliteDatabase } from "../../src/sqlite/node.ts";
 import { type Backend, sqliteBackend } from "../../src/store/backend.ts";
 import type { StagedWrite } from "../../src/store/step.ts";
@@ -67,12 +67,16 @@ async function sessionStore(path = ":memory:"): Promise<Store> {
 }
 
 /** The table every copy writes: a counter per id. */
-const Counts = defineTable("counts", { version: 1, columns: { id: str(), n: int() }, key: ["id"] });
+const Counts = defineTable("counts", {
+	version: 1,
+	columns: { id: str(), n: int(), pad: opt(str()) },
+	key: ["id"],
+});
 const TABLE = "app.sync:counts";
 const OPEN = { admit: () => true, adopt: () => {} };
 
 /** One batch of the test's history: rows to put. */
-type Rows = readonly { readonly id: string; readonly n: number }[];
+type Rows = readonly { readonly id: string; readonly n: number; readonly pad?: string }[];
 
 /** Commits each batch of rows to `store`, preparing the table first (its creation is a commit of its own). */
 async function commitAll(store: Store, batches: readonly Rows[]): Promise<void> {
@@ -227,9 +231,9 @@ describe("batches", () => {
 		await commitAll(store, seedBatches());
 		const commits = await readOutbox(store, LOG_START, 1000, context);
 		expect(commits.length).toBeGreaterThan(0);
-		const bytes = await encodeBatch(commits);
-		expect(bytes[0]).toBe(0x1f);
-		expect(await decodeBatch(bytes)).toEqual(commits);
+		const decoded = [];
+		for await (const commit of decodeBatch(encodeBatch(commits))) decoded.push(commit);
+		expect(decoded).toEqual(commits);
 	});
 });
 
@@ -582,6 +586,61 @@ describe("syncing a session between devices", () => {
 	});
 });
 
+describe("bounded memory", () => {
+	/** Rows of `size` characters each, one per commit: about `count * size` bytes of history. */
+	function bigBatches(count: number, size: number): Rows[] {
+		return Array.from({ length: count }, (_, index) => [{ id: `big${index}`, n: index, pad: "x".repeat(size) }]);
+	}
+
+	it("ends a batch at BATCH_BYTES, so a large history uploads in several slots", async () => {
+		const store = await remote();
+		const laptop = device("laptop", store);
+		// 24 commits of ~1 MB: about three batches of BATCH_BYTES (8 MB).
+		await laptop.commit(bigBatches(24, 1024 * 1024));
+		expect(await laptop.upload()).toEqual({ type: "uploaded", head: (await laptop.store()).lastSeq });
+		const slots = (await store.list(SESSION, context)).batches;
+		expect(slots.length).toBeGreaterThanOrEqual(3);
+		for (const [index, slot] of slots.entries()) if (index > 0) expect(slot.after).toBe(slots[index - 1]!.to);
+		const replica = device("replica", store);
+		await replica.pull();
+		expect(await reads(await replica.store())).toEqual(await reads(await laptop.store()));
+	});
+
+	it("applies a large batch in chunks of APPLY_BYTES, and a pull stopped between them resumes", async () => {
+		const store = await remote();
+		const laptop = device("laptop", store);
+		// 6 commits of ~1 MB in one batch: about two apply chunks of APPLY_BYTES (4 MB).
+		await laptop.commit(bigBatches(6, 1024 * 1024));
+		await laptop.upload();
+		expect((await store.list(SESSION, context)).batches).toHaveLength(1);
+		const viewer = device("viewer", store);
+		const local = tracked(await sessionStore());
+		const chord = local as ChordStore;
+		let applied = 0;
+		const apply = chord.replay.bind(chord);
+		chord.replay = ((...args: Parameters<ChordStore["replay"]>) => {
+			if (++applied === 2) return Promise.reject(new Error("power cut"));
+			return apply(...args);
+		}) as ChordStore["replay"];
+		const pull = () =>
+			pullSession({
+				sessionId: SESSION,
+				store,
+				local: { kind: "store", store: local },
+				state: viewer.state,
+				context,
+			});
+		await expect(pull()).rejects.toThrow("power cut");
+		const midway = local.lastSeq;
+		expect(midway).toBeGreaterThan(0);
+		expect(midway).toBeLessThan((await laptop.store()).lastSeq);
+		await pull();
+		// The first chunk was applied, the second refused, then the resumed pull applied the rest in more chunks.
+		expect(applied).toBeGreaterThanOrEqual(3);
+		expect(await reads(local)).toEqual(await reads(await laptop.store()));
+	});
+});
+
 describe("host pieces", () => {
 	it("lists sessions and keeps the last metadata written", async () => {
 		const store = await remote();