feat(agent): fork closed legacy JSONL through format 4

Normalize closed v3 sources into temporary format-4 files and reuse the two-pass fork path without modifying source files.

Remove obsolete JSONL snapshot creation and capture plumbing.
This commit is contained in:
David Brailovsky 2026-09-04 17:40:30 +02:00
parent e3c966b6ff
commit 9908b2176a
9 changed files with 136 additions and 252 deletions

View file

@ -258,3 +258,11 @@ Original requirement: capture the exact source file length at the commit-queue b
Current implementation: capture the source's current `nextSeq` at the commit-queue boundary and have both independently opened scans stop before that sequence. This is simpler and avoids adding byte-range semantics to `FileSystem`, while still excluding later append-only commits transaction by transaction.
The current implementation assumes a format-4 source remains append-only and is not replaced while the fork runs. Concurrent compaction or another whole-file replacement could otherwise make the two scans observe different file incarnations. Decide whether to restore fixed-length capture, retain the sequence boundary with an explicit no-replacement invariant, or add replacement coordination when J1 compaction is implemented.
### 9.4 Legacy-v3 normalization memory
Original requirement: normalize a closed legacy-v3 source through a bounded disk-backed parser and index.
Current implementation: read and normalize the complete v3 source in memory, write the normalized records to a temporary format-4 file, then fork that stable file through the ordinary two-pass format-4 path. This reuses the current fork semantics, keeps reminted IDs stable across both passes, and never modifies the v3 source, but retains source-sized lines, entries, maps, and normalized writes in memory.
Decide whether closed legacy-v3 files are sufficiently rare and bounded to keep this exception, or replace only the normalization internals with a disk-backed index before WP08 completion.

View file

@ -1,4 +1,3 @@
import type { CommittedWrite } from "./commit.ts";
import { type ForkCurrentStatePlan, projectForkCurrentStateWrite, selectBranchFork } from "./fork-policy.ts";
import type { Entry, ForkOptions } from "./types.ts";
import { branchTip, laneConfig, laneState, type StoredValue, type Value, value } from "./values.ts";
@ -63,22 +62,6 @@ export function createForkSnapshot(source: ForkSourceSnapshot, options: ForkOpti
return { entries, scalarValues, nextSeq };
}
export function forkSnapshotWrites(snapshot: ForkDestinationSnapshot): CommittedWrite[] {
const writes: CommittedWrite[] = [];
for (const entry of snapshot.entries.values()) writes.push({ kind: "entry", ...entry });
for (const stored of snapshot.scalarValues) {
writes.push({
kind: "value",
op: "set",
seq: stored.seq,
namespace: stored.address.namespace,
key: stored.address.key,
value: stored.value,
});
}
return writes.sort((left, right) => left.seq - right.seq);
}
function selectForkContents(
sourceEntries: Map<string, Entry>,
sourceTips: StoredValue<string | null>[],

View file

@ -382,11 +382,4 @@ export class InMemoryStorageState {
getNextSeq(): number {
return this.nextSeq;
}
snapshotEntriesAndValues(): { entries: Entry[]; scalarValues: StoredValue<unknown>[] } {
return {
entries: [...this.entries.values()].sort((left, right) => left.seq - right.seq),
scalarValues: [...this.scalarValues.values()],
};
}
}

View file

@ -1,5 +1,5 @@
import type { Context } from "../../context.ts";
import type { FileError, FileSystem, Result, TextLineReader } from "../../types.ts";
import type { FileSystem, TextLineReader } from "../../types.ts";
import type {
CommittedEntryWrite,
CommittedListAppendWrite,
@ -9,12 +9,13 @@ import type {
import { type ForkCurrentStatePlan, projectForkCurrentStateWrite, selectBranchFork } from "../fork-policy.ts";
import type { ForkOptions } from "../types.ts";
import { parseJsonlSessionHeader } from "./codec.ts";
import { parseJsonlTransaction, serializeJsonlTransaction } from "./io.ts";
import { JSONL_STORAGE_VERSION, type JsonlSessionMetadata, type JsonlStorageHeader } from "./types.ts";
import { fileValue, parseJsonlTransaction, serializeJsonlTransaction } from "./io.ts";
import { JSONL_STORAGE_VERSION, type JsonlStorageHeader } from "./types.ts";
function fileValue<T>(result: Result<T, FileError>, action: string): T {
if (!result.ok) throw new Error(`${action}: ${result.error.message}`, { cause: result.error });
return result.value;
interface JsonlForkSourceMetadata {
id: string;
cwd: string;
path: string;
}
function physicalKey(namespace: string, key: string): string {
@ -23,7 +24,7 @@ function physicalKey(namespace: string, key: string): string {
async function readJsonlForkHeader(
reader: TextLineReader,
source: JsonlSessionMetadata,
source: JsonlForkSourceMetadata,
context: Context,
): Promise<JsonlStorageHeader> {
const line = fileValue(await reader.readLine(context), `Failed to read JSONL fork source ${source.path}`);
@ -59,7 +60,7 @@ function reachesForkBoundary(writes: readonly CommittedWrite[], stopBeforeSeq: n
async function scanJsonlForkTransactions(
fileSystem: FileSystem,
source: JsonlSessionMetadata,
source: JsonlForkSourceMetadata,
stopBeforeSeq: number | undefined,
onTransaction: (writes: readonly CommittedWrite[]) => void | Promise<void>,
context: Context,
@ -275,8 +276,8 @@ function projectJsonlForkWrite(
}
export type JsonlForkSource =
| { kind: "open"; metadata: JsonlSessionMetadata; nextSeq: number }
| { kind: "closed"; metadata: JsonlSessionMetadata };
| { kind: "open"; metadata: JsonlForkSourceMetadata; nextSeq: number }
| { kind: "closed"; metadata: JsonlForkSourceMetadata };
export async function runJsonlFork(
options: {

View file

@ -10,7 +10,7 @@ import type {
CommittedWrite,
} from "../commit.ts";
function fileValue<T>(result: Result<T, FileError>, action: string): T {
export function fileValue<T>(result: Result<T, FileError>, action: string): T {
if (!result.ok) throw new Error(`${action}: ${result.error.message}`, { cause: result.error });
return result.value;
}

View file

@ -9,6 +9,7 @@ import type { CommittedEntryWrite, CommittedValueSetWrite, CommittedWrite } from
import type { JsonValue, LaneConfiguration } from "../types.ts";
import { branchTip, entryLabel, laneConfig, laneState, sessionName } from "../values.ts";
import { type LegacyV3SessionHeader, parseJsonlSessionHeader } from "./codec.ts";
import { fileValue } from "./io.ts";
import {
JSONL_FORMAT_VERSION,
JSONL_STORAGE_VERSION,
@ -119,6 +120,12 @@ export interface NormalizedLegacyV3Records {
nextSeq: number;
}
export interface NormalizedLegacyV3Source {
header: JsonlStorageHeader;
writes: CommittedWrite[];
nextSeq: number;
}
type JsonlSessionMetadataBase = Omit<JsonlSessionMetadata, "path" | "modifiedAt">;
async function resolveLegacyV3ParentSessionId(
@ -537,3 +544,58 @@ export function normalizeLegacyV3Records(recordLines: readonly string[]): Normal
nextSeq: writes.length + 1,
};
}
// TODO: Share read-only complete-line and header parsing with JsonlStorage.open().
async function readLegacyV3SourceLines(
fileSystem: FileSystem,
source: JsonlSessionMetadata,
context: Context,
): Promise<{ header: LegacyV3SessionHeader; recordLines: string[] }> {
const reader = fileValue(
await fileSystem.openTextLineReader(source.path, context),
`Failed to open legacy v3 fork source ${source.path}`,
);
try {
const headerLine = fileValue(
await reader.readLine(context),
`Failed to read legacy v3 fork source ${source.path}`,
);
if (headerLine === undefined || !headerLine.terminated || headerLine.text === "") {
throw new Error(`Invalid legacy v3 JSONL storage ${source.path}: missing header`);
}
const parsed = parseJsonlSessionHeader(headerLine.text);
if (!parsed.ok || parsed.value.format !== "v3-legacy") {
throw new Error(`Invalid legacy v3 JSONL storage ${source.path}: expected format 3 header`, {
cause: parsed.ok ? undefined : parsed.error,
});
}
if (parsed.value.header.id !== source.id || parsed.value.header.cwd !== source.cwd) {
throw new Error(`Session identity does not match header: ${source.id}`);
}
const recordLines: string[] = [];
while (true) {
const line = fileValue(await reader.readLine(context), `Failed to read legacy v3 fork source ${source.path}`);
if (line === undefined || !line.terminated) break;
recordLines.push(line.text);
}
return { header: parsed.value.header, recordLines };
} finally {
await reader.close(context);
}
}
/** Read and normalize one closed legacy-v3 source without modifying it. */
export async function readNormalizedLegacyV3Source(
fileSystem: FileSystem,
source: JsonlSessionMetadata,
context: Context,
): Promise<NormalizedLegacyV3Source> {
const { header, recordLines } = await readLegacyV3SourceLines(fileSystem, source, context);
const { writes, nextSeq } = normalizeLegacyV3Records(recordLines);
return {
header: await normalizeLegacyV3Header(fileSystem, header, context),
writes,
nextSeq,
};
}

View file

@ -1,12 +1,12 @@
import { uuidv7 } from "@earendil-works/pi-ai/utils/uuid";
import type { Context } from "../../context.ts";
import type { FileError, FileInfo, FileSystem, Result } from "../../types.ts";
import { createForkSnapshot, type ForkSourceSnapshot } from "../fork.ts";
import type { FileInfo, FileSystem } from "../../types.ts";
import { StorageBackedSession } from "../session.ts";
import type { ForkOptions, Session, SessionRepo } from "../types.ts";
import { parseJsonlSessionHeader } from "./codec.ts";
import { type JsonlForkSource, runJsonlFork } from "./fork.ts";
import { metadataFromLegacyV3Header } from "./legacy-v3.ts";
import { fileValue, serializeJsonlTransaction } from "./io.ts";
import { metadataFromLegacyV3Header, readNormalizedLegacyV3Source } from "./legacy-v3.ts";
import { JsonlStorage } from "./storage.ts";
import {
JSONL_FORMAT_VERSION,
@ -18,11 +18,6 @@ import {
type JsonlStorageHeader,
} from "./types.ts";
function fileValue<T>(result: Result<T, FileError>, action: string): T {
if (!result.ok) throw new Error(`${action}: ${result.error.message}`, { cause: result.error });
return result.value;
}
function metadataFromHeader(header: JsonlStorageHeader, path: string, modifiedAt: number): JsonlSessionMetadata {
return {
id: header.id,
@ -47,9 +42,7 @@ function sessionFileName(createdAt: number, id: string): string {
return `${timestamp}_${encodeURIComponent(id)}.jsonl`;
}
type ResolvedJsonlForkSource =
| { kind: "snapshot"; snapshot: ForkSourceSnapshot }
| { kind: "stream"; source: JsonlForkSource };
type ResolvedJsonlForkSource = JsonlForkSource | { kind: "legacy-v3"; metadata: JsonlSessionMetadata };
/** File-backed format-4 session repository lifecycle. */
export class JsonlSessionRepo
@ -307,12 +300,10 @@ export class JsonlSessionRepo
);
}
const nextSeq = await storage.captureForkNextSeq(context);
return { kind: "stream", source: { kind: "open", metadata: source, nextSeq } };
return { kind: "open", metadata: source, nextSeq };
}
if (await this.isLegacyV3ForkSource(source, context)) {
return { kind: "snapshot", snapshot: await this.loadClosedForkSourceSnapshot(source, context) };
}
return { kind: "stream", source: { kind: "closed", metadata: source } };
if (await this.isLegacyV3ForkSource(source, context)) return { kind: "legacy-v3", metadata: source };
return { kind: "closed", metadata: source };
}
private async createForkStorage(
@ -322,18 +313,12 @@ export class JsonlSessionRepo
fork: ForkOptions,
context: Context,
): Promise<JsonlStorage> {
const storageOptions = { fileSystem: this.fileSystem, path, now: this.now };
if (source.kind === "snapshot") {
return JsonlStorage.createFromForkSnapshot(
storageOptions,
header,
createForkSnapshot(source.snapshot, fork),
context,
);
if (source.kind === "legacy-v3") {
return this.createLegacyV3ForkStorage(path, header, source.metadata, fork, context);
}
await runJsonlFork(
{
source: source.source,
source,
fileSystem: this.fileSystem,
destinationPath: path,
destinationHeader: header,
@ -341,7 +326,48 @@ export class JsonlSessionRepo
},
context,
);
return JsonlStorage.open(storageOptions, context);
return JsonlStorage.open({ fileSystem: this.fileSystem, path, now: this.now }, context);
}
/** Normalize a closed v3 source into a temporary v4 file, then fork it through the format-4 path. */
private async createLegacyV3ForkStorage(
path: string,
header: Omit<JsonlStorageHeader, "nextSeq">,
source: JsonlSessionMetadata,
fork: ForkOptions,
context: Context,
): Promise<JsonlStorage> {
const normalizedPath = `${path}.legacy-v3-source.tmp`;
try {
const normalized = await readNormalizedLegacyV3Source(this.fileSystem, source, context);
const normalizedHeader = { ...normalized.header, nextSeq: normalized.nextSeq };
fileValue(
await this.fileSystem.writeFile(normalizedPath, `${JSON.stringify(normalizedHeader)}\n`, context),
`Failed to stage normalized legacy v3 source ${source.path}`,
);
for (const write of normalized.writes) {
fileValue(
await this.fileSystem.appendFile(normalizedPath, `${serializeJsonlTransaction([write])}\n`, context),
`Failed to stage normalized legacy v3 source ${source.path}`,
);
}
await runJsonlFork(
{
source: {
kind: "closed",
metadata: { id: normalizedHeader.id, cwd: normalizedHeader.cwd, path: normalizedPath },
},
fileSystem: this.fileSystem,
destinationPath: path,
destinationHeader: header,
fork,
},
context,
);
return JsonlStorage.open({ fileSystem: this.fileSystem, path, now: this.now }, context);
} finally {
await this.fileSystem.remove(normalizedPath, { force: true }, context);
}
}
private async isLegacyV3ForkSource(source: JsonlSessionMetadata, context: Context): Promise<boolean> {
@ -354,18 +380,6 @@ export class JsonlSessionRepo
return parsed.ok && parsed.value.format === "v3-legacy";
}
private async loadClosedForkSourceSnapshot(
source: JsonlSessionMetadata,
context: Context,
): Promise<ForkSourceSnapshot> {
const storage = await this.loadStorage(source, context);
try {
return await storage.captureForkSource(context);
} finally {
await storage.close(context);
}
}
private publishOpenSession(
metadata: JsonlSessionMetadata,
storage: JsonlStorage,

View file

@ -1,9 +1,8 @@
import type { Usage } from "@earendil-works/pi-ai";
import { uuidv7 } from "@earendil-works/pi-ai/utils/uuid";
import type { Context } from "../../context.ts";
import type { FileError, FileSystem, Result } from "../../types.ts";
import type { FileSystem } from "../../types.ts";
import { insertUsage } from "../commit.ts";
import { type ForkDestinationSnapshot, type ForkSourceSnapshot, forkSnapshotWrites } from "../fork.ts";
import { type CommittedWrite, InMemoryStorageState } from "../in-memory-storage-state.ts";
import type {
CommitResult,
@ -19,15 +18,10 @@ import type {
} from "../types.ts";
import type { ListElement, ListReadOptions, StoredValue, Value, ValueList } from "../values.ts";
import { type LegacyV3SessionHeader, parseJsonlSessionHeader } from "./codec.ts";
import { parseJsonlTransaction, publishFileAtomically, serializeJsonlTransaction } from "./io.ts";
import { fileValue, parseJsonlTransaction, publishFileAtomically, serializeJsonlTransaction } from "./io.ts";
import { normalizeLegacyV3Header, normalizeLegacyV3Records } from "./legacy-v3.ts";
import { JSONL_STORAGE_VERSION, type JsonlStorageHeader, type JsonlStorageOptions } from "./types.ts";
function fileValue<T>(result: Result<T, FileError>, action: string): T {
if (!result.ok) throw new Error(`${action}: ${result.error.message}`, { cause: result.error });
return result.value;
}
function serializeStorage(header: JsonlStorageHeader, transactions: CommittedWrite[][]): string {
return `${[JSON.stringify(header), ...transactions.map(serializeJsonlTransaction)].join("\n")}\n`;
}
@ -81,27 +75,6 @@ export class JsonlStorage implements Storage {
return storage;
}
/** Atomically create storage from a complete prepared snapshot. */
static async createFromForkSnapshot(
options: JsonlStorageOptions,
header: JsonlStorageHeader,
snapshot: ForkDestinationSnapshot,
context: Context,
): Promise<JsonlStorage> {
const writes = forkSnapshotWrites(snapshot);
const snapshotHeader = { ...header, nextSeq: snapshot.nextSeq };
await publishFileAtomically(
options.fileSystem,
options.path,
serializeStorage(
snapshotHeader,
writes.map((write) => [write]),
),
context,
);
return JsonlStorage.open(options, context);
}
static async open(options: JsonlStorageOptions, context: Context): Promise<JsonlStorage> {
const content = fileValue(
await options.fileSystem.readTextFile(options.path, context),
@ -295,17 +268,6 @@ export class JsonlStorage implements Storage {
return result;
}
/** Capture the state needed to fork at one serialized boundary between commits. */
captureForkSource(_context: Context): Promise<ForkSourceSnapshot> {
if (this.state !== "open") return Promise.reject(new Error("JsonlStorage is closed"));
const result = this.commitQueue.then(() => this.storageState.snapshotEntriesAndValues());
this.commitQueue = result.then(
() => undefined,
() => undefined,
);
return result;
}
close(_context: Context): Promise<void> {
if (this.closePromise !== undefined) return this.closePromise;
this.state = "closing";

View file

@ -2,7 +2,6 @@ import { describe, expect, it } from "vitest";
import { BACKGROUND_CONTEXT } from "../../src/harness/context.ts";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import * as sessionWrites from "../../src/harness/session/commit.ts";
import type { ForkDestinationSnapshot } from "../../src/harness/session/fork.ts";
import { JSONL_FORMAT_VERSION, JsonlStorage, type JsonlStorageHeader } from "../../src/harness/session/jsonl/index.ts";
import * as storedValues from "../../src/harness/session/values.ts";
import { getOrThrow } from "../../src/harness/types.ts";
@ -21,106 +20,6 @@ function header(id: string): JsonlStorageHeader {
};
}
function stored<T>(address: storedValues.Value<T>, value: T, seq: number): storedValues.StoredValue<unknown> {
return {
address: storedValues.value<unknown>(address.namespace, address.key),
value,
seq,
};
}
function preparedSnapshot(): ForkDestinationSnapshot {
return {
entries: new Map([
[
"root",
{
id: "root",
parentId: null,
type: "message",
message: { role: "user", content: "hello", timestamp: 1 },
seq: 1,
timestamp: NOW,
},
],
]),
scalarValues: [
stored(storedValues.branchTip("main"), "root", 2),
stored(storedValues.laneState("main"), { currentOperationId: null, lastOperationId: null, inbox: [] }, 3),
stored(storedValues.sessionName, "forked", 4),
],
nextSeq: 9,
};
}
describe("JsonlStorage snapshot creation", () => {
it("atomically publishes, advances, and reopens a prepared snapshot", async () => {
const fileSystem = new NodeExecutionEnv({ cwd: createTempDir() });
const options = { fileSystem, path: "fork.jsonl", now: () => NOW };
const storage = await JsonlStorage.createFromForkSnapshot(
options,
header("fork"),
preparedSnapshot(),
BACKGROUND_CONTEXT,
);
const lines = getOrThrow(await fileSystem.readTextFile("fork.jsonl", BACKGROUND_CONTEXT))
.trimEnd()
.split("\n");
expect(lines.slice(1).every((line) => !Array.isArray(JSON.parse(line)))).toBe(true);
expect((await storage.getEntries(["root"], BACKGROUND_CONTEXT)).get("root")?.timestamp).toBe(NOW);
expect((await storage.getValue(storedValues.branchTip("main"), BACKGROUND_CONTEXT))?.value).toBe("root");
expect((await storage.getValue(storedValues.laneState("main"), BACKGROUND_CONTEXT))?.value).toEqual({
currentOperationId: null,
lastOperationId: null,
inbox: [],
});
expect((await storage.getValue(storedValues.sessionName, BACKGROUND_CONTEXT))?.value).toBe("forked");
expect(await storage.getStats(BACKGROUND_CONTEXT)).toEqual({
messageCount: 1,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
});
expect(await storage.scanUsage({ order: "asc" }, BACKGROUND_CONTEXT)).toEqual([]);
expect(
(await storage.commit([storedValues.setValue(storedValues.sessionName, "updated")], BACKGROUND_CONTEXT))
.firstSeq,
).toBe(9);
expect(getOrThrow(await fileSystem.exists("fork.jsonl.tmp", BACKGROUND_CONTEXT))).toBe(false);
await storage.close(BACKGROUND_CONTEXT);
const reopened = await JsonlStorage.open(options, BACKGROUND_CONTEXT);
expect((await reopened.getEntries(["root"], BACKGROUND_CONTEXT)).has("root")).toBe(true);
expect((await reopened.getValue(storedValues.sessionName, BACKGROUND_CONTEXT))?.value).toBe("updated");
expect((await reopened.commit([], BACKGROUND_CONTEXT)).firstSeq).toBe(10);
await reopened.close(BACKGROUND_CONTEXT);
});
it("removes the temporary file when publication fails", async () => {
const fileSystem = new NodeExecutionEnv({ cwd: createTempDir() });
getOrThrow(await fileSystem.createDir("blocked.jsonl", undefined, BACKGROUND_CONTEXT));
await expect(
JsonlStorage.createFromForkSnapshot(
{ fileSystem, path: "blocked.jsonl", now: () => NOW },
header("blocked"),
preparedSnapshot(),
BACKGROUND_CONTEXT,
),
).rejects.toThrow("Failed to publish JSONL storage");
expect(getOrThrow(await fileSystem.exists("blocked.jsonl.tmp", BACKGROUND_CONTEXT))).toBe(false);
expect(getOrThrow(await fileSystem.fileInfo("blocked.jsonl", BACKGROUND_CONTEXT))).toMatchObject({
kind: "directory",
});
});
});
describe("JsonlStorage persistence", () => {
it("replays whole-list deletion without resurrecting earlier appends", async () => {
const fileSystem = new NodeExecutionEnv({ cwd: createTempDir() });
@ -217,44 +116,6 @@ describe("JsonlStorage persistence", () => {
});
});
describe("JsonlStorage snapshots", () => {
it("captures one serialized boundary between commits", async () => {
const fileSystem = new NodeExecutionEnv({ cwd: createTempDir() });
const options = { fileSystem, path: "session.jsonl", now: () => NOW };
const storage = await JsonlStorage.create(options, header("snapshot"), [], BACKGROUND_CONTEXT);
const firstCommit = storage.commit(
[
sessionWrites.insertEntry({ id: "root", parentId: null, type: "custom", customType: "root" }),
storedValues.setValue(storedValues.branchTip("main"), "root"),
],
BACKGROUND_CONTEXT,
);
const snapshot = storage.captureForkSource(BACKGROUND_CONTEXT);
const secondCommit = storage.commit(
[
sessionWrites.insertEntry({ id: "child", parentId: "root", type: "custom", customType: "child" }),
storedValues.setValue(storedValues.branchTip("main"), "child"),
],
BACKGROUND_CONTEXT,
);
await firstCommit;
const captured = await snapshot;
await secondCommit;
expect(captured.entries.map(({ id }) => id)).toEqual(["root"]);
expect(
captured.scalarValues.find(
({ address }) => address.namespace === storedValues.branchTip("main").namespace && address.key === "main",
)?.value,
).toBe("root");
expect((await storage.getEntries(["child"], BACKGROUND_CONTEXT)).has("child")).toBe(true);
expect((await storage.getValue(storedValues.branchTip("main"), BACKGROUND_CONTEXT))?.value).toBe("child");
await storage.close(BACKGROUND_CONTEXT);
});
});
describe("JsonlStorage torn tail", () => {
function entryWrite(id: string) {
return sessionWrites.insertEntry({