sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(agent): stream legacy JSONL migration and forks

Browse source 
Stream legacy v3 normalization through repeatable file passes, reuse atomic JSONL publication, and preserve stable fork IDs without an intermediate v4 source file.
This commit is contained in:
David Brailovsky 2026-09-07 12:47:26 +02:00
parent 85186f823d
commit 6dfc66d32a
11 changed files with 1135 additions and 655 deletions
Download patch file Download diff file Expand all files Collapse all files

27
packages/agent/docs/work-packages/08-named-branch-streaming-forks.md
View file

@ -239,9 +239,9 @@ WP08 is complete when:
- normative docs and the roadmap reflect the new contract with historical documents untouched;
- final Fable review reports no blocker.
## 9. Open questions
## 9. Open questions / potential spec changes
These do not change the current contract until resolved.
These record implementation decisions and potential specification changes. The requirements above remain unchanged until explicitly revised.
### 9.1 Corruption checks
@ -263,6 +263,25 @@ The current implementation assumes a format-4 source remains append-only and is
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.
Current implementation does not do that. Instead, it:
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.
- scans the v3 file once for structure, stable IDs, labels/configuration, usage, and current values;
- does not keep all message payloads in memory during that scan;
- later reopens the v3 file to stream selected normalized writes directly into the fork output;
- no longer writes an intermediate normalized v4 temp file.
For compactions, the replay pass may cache needed context messages so it can build `retainedTail`. Later append-only records are excluded by the captured complete-record count.
The source must not be edited or replaced between passes. Header and physical-entry checks catch some changes, but this is not snapshot isolation.
Memory is improved versus storing all normalized writes, but can still grow large if selected compactions need many context messages. Decide whether this is acceptable, or whether WP08 still needs the originally planned disk-backed implementation.
### 9.5 JSONL structural-index memory
The current fork implementation uses in-memory structural indexes. For v4, this includes scalar/list state, entry parent links, selected ancestry, and lane inventory. For v3, this includes structural metadata, stable ID mappings, derived values, and the compaction context-message cache described in §9.4.
This does not meet the disk-backed indexing requirement in §1.3. Decide whether in-memory structural indexes are acceptable, or whether WP08 still requires disk-backed indexes before completion.
### 9.6 TODO: Cleanup after invocation cancellation
Cancellation-safe filesystem cleanup is deferred. Cleanup currently receives the caller's Context, so an aborted invocation can leave a staged file behind even though readers close and the destination id reservation is released. Define the cleanup Context policy before implementing this; do not import Chord's `withoutAbortSignal` directly into JSONL code as a temporary workaround.

24
packages/agent/src/harness/env/nodejs.ts vendored
View file

@ -30,6 +30,7 @@ import {
type Result,
type ShellExecOptions,
type ShellExecResult,
type TextLine,
type TextLineReader,
toError,
} from "../types.ts";
@ -375,6 +376,7 @@ class NodeTextLineReader implements TextLineReader {
private readonly path: string;
private readonly decoder = new TextDecoder();
private readonly chunk = new Uint8Array(64 * 1024);
private byteOffset = 0;
private buffered = "";
private ended = false;
private closed = false;
@ -384,8 +386,8 @@ class NodeTextLineReader implements TextLineReader {
this.path = path;
}
async readLine(context: Context): Promise<Result<{ text: string; terminated: boolean } | undefined, FileError>> {
const aborted = abortResult<{ text: string; terminated: boolean } | undefined>(context.abortSignal, this.path);
async readLine(context: Context): Promise<Result<TextLine | undefined, FileError>> {
const aborted = abortResult<TextLine | undefined>(context.abortSignal, this.path);
if (aborted) return aborted;
if (this.closed) return err(new FileError("invalid", "Text line reader is closed", this.path));
@ -393,7 +395,7 @@ class NodeTextLineReader implements TextLineReader {
while (true) {
const newline = this.buffered.indexOf("\n");
if (newline !== -1) {
const text = this.buffered.slice(0, newline).replace(/\r$/, "");
const text = this.buffered.slice(0, newline);
this.buffered = this.buffered.slice(newline + 1);
return ok({ text, terminated: true });
}
@ -404,12 +406,11 @@ class NodeTextLineReader implements TextLineReader {
return ok({ text, terminated: false });
}
const { bytesRead } = await this.file.read(this.chunk, 0, this.chunk.length, null);
const afterReadAbort = abortResult<{ text: string; terminated: boolean } | undefined>(
context.abortSignal,
this.path,
);
// Explicit positions allow an aborted read to be retried without skipping bytes.
const { bytesRead } = await this.file.read(this.chunk, 0, this.chunk.length, this.byteOffset);
const afterReadAbort = abortResult<TextLine | undefined>(context.abortSignal, this.path);
if (afterReadAbort) return afterReadAbort;
this.byteOffset += bytesRead;
if (bytesRead === 0) {
this.buffered += this.decoder.decode();
this.ended = true;
@ -425,7 +426,12 @@ class NodeTextLineReader implements TextLineReader {
async close(_context: Context): Promise<void> {
if (this.closed) return;
this.closed = true;
await this.file.close().catch(() => undefined);
this.buffered = "";
try {
await this.file.close();
} catch {
// Closing is best-effort, including after cancellation or an earlier I/O failure.
}
}
}

253
packages/agent/src/harness/session/jsonl/fork.ts
View file

@ -8,8 +8,8 @@ import type {
} from "../commit.ts";
import { type ForkCurrentStatePlan, projectForkCurrentStateWrite, selectBranchFork } from "../fork-policy.ts";
import type { ForkOptions } from "../types.ts";
import { parseJsonlSessionHeader } from "./codec.ts";
import { fileValue, parseJsonlTransaction, serializeJsonlTransaction } from "./io.ts";
import { fileValue, parseJsonlTransaction, publishJsonl, readJsonlHeader } from "./io.ts";
import type { LegacyV3Source } from "./legacy-v3.ts";
import { JSONL_STORAGE_VERSION, type JsonlStorageHeader } from "./types.ts";
interface JsonlForkSourceMetadata {
@ -27,17 +27,11 @@ async function readJsonlForkHeader(
source: JsonlForkSourceMetadata,
context: Context,
): Promise<JsonlStorageHeader> {
const line = fileValue(await reader.readLine(context), `Failed to read JSONL fork source ${source.path}`);
if (line === undefined || !line.terminated || line.text === "") {
throw new Error(`Invalid JSONL storage ${source.path}: missing header`);
const parsed = await readJsonlHeader(reader, source.path, context);
if (parsed.format !== "v4") {
throw new Error(`Invalid JSONL storage ${source.path}: expected format 4 header`);
}
const parsed = parseJsonlSessionHeader(line.text);
if (!parsed.ok || parsed.value.format !== "v4") {
throw new Error(`Invalid JSONL storage ${source.path}: expected format 4 header`, {
cause: parsed.ok ? undefined : parsed.error,
});
}
const header = parsed.value.header;
const header = parsed.header;
if (header.id !== source.id || header.cwd !== source.cwd) {
throw new Error(`Session identity does not match header: ${source.id}`);
}
@ -58,31 +52,19 @@ function reachesForkBoundary(writes: readonly CommittedWrite[], stopBeforeSeq: n
return false;
}
async function scanJsonlForkTransactions(
fileSystem: FileSystem,
source: JsonlForkSourceMetadata,
/** Read complete transactions after the header, never splitting a transaction at the sequence boundary. */
async function* readJsonlForkTransactions(
reader: TextLineReader,
path: string,
stopBeforeSeq: number | undefined,
onTransaction: (writes: readonly CommittedWrite[]) => void | Promise<void>,
context: Context,
): Promise<{ header: JsonlStorageHeader; highestCompleteSeq: number }> {
const reader = fileValue(
await fileSystem.openTextLineReader(source.path, context),
`Failed to open JSONL fork source ${source.path}`,
);
try {
const header = await readJsonlForkHeader(reader, source, context);
let highestCompleteSeq = 0;
while (true) {
const line = fileValue(await reader.readLine(context), `Failed to read JSONL fork source ${source.path}`);
if (line === undefined || !line.terminated) break;
const writes = parseJsonlTransaction(line.text);
if (reachesForkBoundary(writes, stopBeforeSeq)) break;
if (writes.length !== 0) highestCompleteSeq = writes.at(-1)!.seq;
await onTransaction(writes);
}
return { header, highestCompleteSeq };
} finally {
await reader.close(context);
): AsyncIterable<CommittedWrite[]> {
while (true) {
const line = fileValue(await reader.readLine(context), `Failed to read JSONL fork source ${path}`);
if (line === undefined || !line.terminated) break;
const writes = parseJsonlTransaction(line.text);
if (reachesForkBoundary(writes, stopBeforeSeq)) break;
yield writes;
}
}
@ -95,11 +77,15 @@ class JsonlForkIndex {
private readonly laneConfigs = new Set<string>();
private readonly laneStates = new Set<string>();
applyEntry(id: string, parentId: string | null): void {
this.entryParents.set(id, parentId);
}
applyWrites(writes: readonly CommittedWrite[]): void {
for (const write of writes) {
switch (write.kind) {
case "entry":
this.entryParents.set(write.id, write.parentId);
this.applyEntry(write.id, write.parentId);
break;
case "value": {
const key = physicalKey(write.namespace, write.key);
@ -192,6 +178,16 @@ class JsonlForkIndex {
}
}
/**
* Validate the source lanes and return the fork plan.
*
* Branch scope updates JsonlForkIndex's selected entries via index.selectEntry(): it selects the
* destination tip and its ancestors after validating entryId and applying "at"/"before". The plan
* records the branch name and destination tip; the copy pass uses the index to filter entries and labels.
*
* Tree scope leaves the index's selected entries unchanged because the copy pass keeps every entry.
* Both formats use the same indexed metadata; this function performs no file I/O.
*/
function selectJsonlFork(index: JsonlForkIndex, options: ForkOptions): ForkCurrentStatePlan {
index.validateLanes(options);
if (options.scope === "tree") return { scope: "tree" };
@ -202,55 +198,6 @@ function selectJsonlFork(index: JsonlForkIndex, options: ForkOptions): ForkCurre
});
}
class JsonlForkWriter {
private readonly fileSystem: FileSystem;
private readonly destinationPath: string;
private readonly tempPath: string;
private constructor(fileSystem: FileSystem, destinationPath: string) {
this.fileSystem = fileSystem;
this.destinationPath = destinationPath;
this.tempPath = `${destinationPath}.tmp`;
}
static async create(
fileSystem: FileSystem,
destinationPath: string,
header: JsonlStorageHeader,
context: Context,
): Promise<JsonlForkWriter> {
const writer = new JsonlForkWriter(fileSystem, destinationPath);
try {
fileValue(
await fileSystem.writeFile(writer.tempPath, `${JSON.stringify(header)}\n`, context),
`Failed to stage JSONL storage ${destinationPath}`,
);
return writer;
} catch (error) {
await writer.discard(context);
throw error;
}
}
async append(write: JsonlForkWrite, context: Context): Promise<void> {
fileValue(
await this.fileSystem.appendFile(this.tempPath, `${serializeJsonlTransaction([write])}\n`, context),
`Failed to append JSONL fork destination ${this.destinationPath}`,
);
}
async publish(context: Context): Promise<void> {
fileValue(
await this.fileSystem.renameFile(this.tempPath, this.destinationPath, context),
`Failed to publish JSONL storage ${this.destinationPath}`,
);
}
async discard(context: Context): Promise<void> {
await this.fileSystem.remove(this.tempPath, { force: true }, context);
}
}
type JsonlForkWrite = CommittedEntryWrite | CommittedValueSetWrite | CommittedListAppendWrite;
function projectJsonlForkWrite(
@ -275,13 +222,97 @@ function projectJsonlForkWrite(
}
}
export type JsonlForkSource =
/** Prepared fork input: format-4 file metadata or an already-normalized legacy source. */
export type JsonlForkInput =
| { kind: "open"; metadata: JsonlForkSourceMetadata; nextSeq: number }
| { kind: "closed"; metadata: JsonlForkSourceMetadata };
| { kind: "closed"; metadata: JsonlForkSourceMetadata }
| { kind: "legacy-v3"; normalized: LegacyV3Source };
/**
* Build the index used to select branch ancestry and identify current scalar/list writes.
* Retains entry parent links, current-row sequences, and lane inventory, not entry payloads.
* Returns the index and the source's nextSeq high-water mark; no destination writes occur here.
*
* V3: LegacyV3Source already scanned the file and assigned normalized IDs/sequences. Copy its
* in-memory entry structures and derived values without reopening the file or calling writes(),
* which would unnecessarily parse payloads and reconstruct compaction tails before fork selection.
*
* V4: Open the source file and fold complete transactions into the index. For an open source,
* stop at the captured commit-queue sequence boundary. For a closed source, scan to EOF or a torn
* final line and derive nextSeq from the header and highest complete write sequence.
* The later copy pass reopens the source file in both formats and emits only selected writes.
*/
async function indexForkInput(
input: JsonlForkInput,
fileSystem: FileSystem,
context: Context,
): Promise<{ index: JsonlForkIndex; nextSeq: number }> {
const index = new JsonlForkIndex();
if (input.kind === "legacy-v3") {
for (const entry of input.normalized.entryStructures()) index.applyEntry(entry.id, entry.parentId);
index.applyWrites(input.normalized.values);
return { index, nextSeq: input.normalized.nextSeq };
}
const reader = fileValue(
await fileSystem.openTextLineReader(input.metadata.path, context),
`Failed to open JSONL fork source ${input.metadata.path}`,
);
try {
const header = await readJsonlForkHeader(reader, input.metadata, context);
const stopBeforeSeq = input.kind === "open" ? input.nextSeq : undefined;
let highestCompleteSeq = 0;
for await (const writes of readJsonlForkTransactions(reader, input.metadata.path, stopBeforeSeq, context)) {
index.applyWrites(writes);
if (writes.length !== 0) highestCompleteSeq = writes.at(-1)!.seq;
}
return {
index,
nextSeq: input.kind === "open" ? input.nextSeq : Math.max(header.nextSeq ?? 1, highestCompleteSeq + 1),
};
} finally {
await reader.close(context);
}
}
/** Yield source writes; the caller owns final projection and filtering. */
async function* streamForkWrites(
input: JsonlForkInput,
fileSystem: FileSystem,
stopBeforeSeq: number,
isEntryCopied: (entryId: string) => boolean,
context: Context,
): AsyncIterable<CommittedWrite> {
if (input.kind === "legacy-v3") {
// V3 filters early to avoid rereading messages and reconstructing unselected compaction tails.
// V4 already stores complete payloads, so it streams all captured writes below.
// Both formats still pass through projectJsonlForkWrite() for final filtering and transformation.
yield* input.normalized.writes(context, isEntryCopied);
return;
}
const reader = fileValue(
await fileSystem.openTextLineReader(input.metadata.path, context),
`Failed to open JSONL fork source ${input.metadata.path}`,
);
try {
await readJsonlForkHeader(reader, input.metadata, context);
for await (const writes of readJsonlForkTransactions(reader, input.metadata.path, stopBeforeSeq, context)) {
yield* writes;
}
} finally {
await reader.close(context);
}
}
/**
* Index the source, validate the requested fork, and stream selected entries and current state
* into an atomically published format-4 destination without modifying the source.
* Preserve copied sequences and the source's nextSeq while excluding usage and open-operation state.
* Source files must not be replaced or edited between passes; later append-only writes are excluded
* by the captured sequence boundary or legacy record count. Does not open the destination Session.
*/
export async function runJsonlFork(
options: {
source: JsonlForkSource;
input: JsonlForkInput;
fileSystem: FileSystem;
destinationPath: string;
destinationHeader: Omit<JsonlStorageHeader, "nextSeq">;
@ -289,45 +320,27 @@ export async function runJsonlFork(
},
context: Context,
): Promise<void> {
const index = new JsonlForkIndex();
const firstPass = await scanJsonlForkTransactions(
options.fileSystem,
options.source.metadata,
options.source.kind === "open" ? options.source.nextSeq : undefined,
(writes) => index.applyWrites(writes),
context,
);
const nextSeq =
options.source.kind === "open"
? options.source.nextSeq
: Math.max(firstPass.header.nextSeq ?? 1, firstPass.highestCompleteSeq + 1);
const plan = selectJsonlFork(index, options.fork);
const { index, nextSeq } = await indexForkInput(options.input, options.fileSystem, context);
let fork = options.fork;
if (options.input.kind === "legacy-v3" && fork.scope === "branch" && fork.entryId !== undefined) {
fork = { ...fork, entryId: options.input.normalized.translateForkEntryId(fork.entryId) };
}
const plan = selectJsonlFork(index, fork);
const isEntryCopied = (entryId: string): boolean => {
if (plan.scope === "tree") return true;
return index.isEntrySelected(entryId);
};
const writer = await JsonlForkWriter.create(
await publishJsonl(
options.fileSystem,
options.destinationPath,
{ ...options.destinationHeader, nextSeq },
context,
async (append) => {
const sourceWrites = streamForkWrites(options.input, options.fileSystem, nextSeq, isEntryCopied, context);
for await (const write of sourceWrites) {
const projected = projectJsonlForkWrite(write, index, plan, isEntryCopied);
if (projected !== undefined) await append([projected]);
}
},
);
try {
await scanJsonlForkTransactions(
options.fileSystem,
options.source.metadata,
nextSeq,
async (writes) => {
for (const write of writes) {
const projected = projectJsonlForkWrite(write, index, plan, isEntryCopied);
if (projected !== undefined) await writer.append(projected, context);
}
},
context,
);
await writer.publish(context);
} catch (error) {
await writer.discard(context);
throw error;
}
}

48
packages/agent/src/harness/session/jsonl/io.ts
View file

@ -1,5 +1,5 @@
import type { Context } from "../../context.ts";
import type { FileError, FileSystem, Result } from "../../types.ts";
import type { FileError, FileSystem, Result, TextLineReader } from "../../types.ts";
import type {
CommittedEntryWrite,
CommittedListAppendWrite,
@ -9,12 +9,30 @@ import type {
CommittedValueSetWrite,
CommittedWrite,
} from "../commit.ts";
import { type JsonlParsedSessionHeader, parseJsonlSessionHeader } from "./codec.ts";
import type { JsonlStorageHeader } from "./types.ts";
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;
}
export async function readJsonlHeader(
reader: TextLineReader,
path: string,
context: Context,
): Promise<JsonlParsedSessionHeader> {
const line = fileValue(await reader.readLine(context), `Failed to read JSONL storage ${path}`);
if (line === undefined || !line.terminated || line.text === "") {
throw new Error(`Invalid JSONL storage ${path}: missing header`);
}
const parsed = parseJsonlSessionHeader(line.text);
if (!parsed.ok) {
throw new Error(`Invalid JSONL storage ${path}: invalid header`, { cause: parsed.error });
}
return parsed.value;
}
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
@ -59,18 +77,22 @@ export function serializeJsonlTransaction(writes: readonly CommittedWrite[]): st
return JSON.stringify(writes.length === 1 ? writes[0] : writes);
}
/** Publish only after the callback succeeds; it must await each append before returning. */
export async function publishFileAtomically(
fileSystem: FileSystem,
destinationPath: string,
content: string,
context: Context,
writeContent: (append: (content: string) => Promise<void>) => Promise<void>,
): Promise<void> {
const tempPath = `${destinationPath}.tmp`;
try {
fileValue(
await fileSystem.writeFile(tempPath, content, context),
`Failed to stage JSONL storage ${destinationPath}`,
);
fileValue(await fileSystem.writeFile(tempPath, "", context), `Failed to stage JSONL storage ${destinationPath}`);
await writeContent(async (content) => {
fileValue(
await fileSystem.appendFile(tempPath, content, context),
`Failed to append JSONL storage ${destinationPath}`,
);
});
fileValue(
await fileSystem.renameFile(tempPath, destinationPath, context),
`Failed to publish JSONL storage ${destinationPath}`,
@ -80,3 +102,17 @@ export async function publishFileAtomically(
throw error;
}
}
/** Stream a header and complete transactions through the shared atomic publisher. */
export async function publishJsonl(
fileSystem: FileSystem,
destinationPath: string,
header: JsonlStorageHeader,
context: Context,
writeTransactions: (append: (writes: readonly CommittedWrite[]) => Promise<void>) => Promise<void>,
): Promise<void> {
await publishFileAtomically(fileSystem, destinationPath, context, async (append) => {
await append(`${JSON.stringify(header)}\n`);
await writeTransactions((writes) => append(`${serializeJsonlTransaction(writes)}\n`));
});
}

644
packages/agent/src/harness/session/jsonl/legacy-v3.ts
View file

@ -3,13 +3,13 @@ import { uuidv7 } from "@earendil-works/pi-ai/utils/uuid";
import type { AgentMessage, ThinkingLevel } from "../../../types.ts";
import type { Context } from "../../context.ts";
import { createBranchSummaryMessage, createCompactionSummaryMessage } from "../../messages.ts";
import type { FileSystem } from "../../types.ts";
import type { FileSystem, TextLineReader } from "../../types.ts";
import { addUsage, emptyUsage } from "../../utils/usage.ts";
import type { CommittedEntryWrite, CommittedValueSetWrite, CommittedWrite } from "../commit.ts";
import type { CommittedEntryWrite, CommittedValueSetWrite } from "../commit.ts";
import type { JsonValue, LaneConfiguration } from "../types.ts";
import { branchTip, entryLabel, laneConfig, laneState, sessionName } from "../values.ts";
import { branchTip, entryLabel, laneConfig, laneState, sessionName, setValue } from "../values.ts";
import { type LegacyV3SessionHeader, parseJsonlSessionHeader } from "./codec.ts";
import { fileValue } from "./io.ts";
import { fileValue, readJsonlHeader } from "./io.ts";
import {
JSONL_FORMAT_VERSION,
JSONL_STORAGE_VERSION,
@ -70,7 +70,7 @@ interface LegacyV3ModelChangeEntry extends LegacyV3EntryBase {
interface LegacyV3ThinkingLevelChangeEntry extends LegacyV3EntryBase {
type: "thinking_level_change";
thinkingLevel: string;
thinkingLevel: ThinkingLevel;
}
interface LegacyV3ActiveToolsChangeEntry extends LegacyV3EntryBase {
@ -114,15 +114,38 @@ type DiscardedLegacyV3Entry =
type LegacyV3Entry = RetainedLegacyV3Entry | DiscardedLegacyV3Entry;
export interface NormalizedLegacyV3Records {
writes: CommittedWrite[];
importedUsage: Usage;
nextSeq: number;
interface LegacyV3IndexBase {
id: string;
parentId: string | null;
/** This node's new ID, or its parent's mapped ID when the node is discarded. */
mappedId: string | null;
}
export interface NormalizedLegacyV3Source {
header: JsonlStorageHeader;
writes: CommittedWrite[];
type RetainedLegacyV3IndexEntry = LegacyV3IndexBase & { mappedId: string; seq: number } & (
| { type: "message" | "custom" | "custom_message" }
| { type: "branch_summary"; fromId: string }
| { type: "compaction"; firstKeptEntryId: string }
);
/** Keep structure, labels, and configuration changes, but not conversation payloads. */
type LegacyV3IndexEntry =
| RetainedLegacyV3IndexEntry
| (LegacyV3IndexBase &
(
| Pick<LegacyV3LabelEntry, "type" | "targetId" | "label">
| Pick<LegacyV3ModelChangeEntry, "type" | "provider" | "modelId">
| Pick<LegacyV3ThinkingLevelChangeEntry, "type" | "thinkingLevel">
| Pick<LegacyV3ActiveToolsChangeEntry, "type" | "activeToolNames">
| { type: "session_info" }
));
type LegacyV3CompactionIndexEntry = Extract<RetainedLegacyV3IndexEntry, { type: "compaction" }>;
interface LegacyV3Inventory {
entries: ReadonlyMap<string, LegacyV3IndexEntry>;
importedUsage: Usage;
name: string | undefined;
finalId: string | null;
nextSeq: number;
}
@ -170,14 +193,14 @@ export async function normalizeLegacyV3Header(
};
}
function parseLegacyV3Entry(line: string, lineNumber: number): LegacyV3Entry {
function parseLegacyV3Entry(line: string): LegacyV3Entry {
let entry: LegacyV3Entry;
try {
entry = JSON.parse(line) as LegacyV3Entry;
} catch (error) {
throw new Error(`Invalid legacy v3 JSONL record at line ${lineNumber}: not valid JSON`, { cause: error });
throw new Error("Invalid legacy v3 JSONL record: not valid JSON", { cause: error });
}
const recordType: unknown = entry.type;
const recordType: unknown = entry?.type;
if (
recordType !== "message" &&
recordType !== "custom" &&
@ -190,7 +213,7 @@ function parseLegacyV3Entry(line: string, lineNumber: number): LegacyV3Entry {
recordType !== "session_info" &&
recordType !== "label"
) {
throw new Error(`Unsupported legacy v3 record type at line ${lineNumber}: ${String(recordType)}`);
throw new Error(`Unsupported legacy v3 record type: ${String(recordType)}`);
}
return entry;
}
@ -208,7 +231,9 @@ function importedCustomMessage(entry: LegacyV3CustomMessageEntry): AgentMessage
return message as unknown as AgentMessage;
}
function isRetainedEntry(entry: LegacyV3Entry): entry is RetainedLegacyV3Entry {
function isRetainedEntry<T extends LegacyV3Entry | LegacyV3IndexEntry>(
entry: T,
): entry is Extract<T, { type: RetainedLegacyV3Entry["type"] }> {
return (
entry.type !== "model_change" &&
entry.type !== "thinking_level_change" &&
@ -218,110 +243,64 @@ function isRetainedEntry(entry: LegacyV3Entry): entry is RetainedLegacyV3Entry {
);
}
class RetainedIdResolver {
/** Caches the reminted ID of each discarded legacy node's nearest retained ancestor, or null. */
private resolvedIds = new Map<string, string | null>();
private entriesById: Map<string, LegacyV3Entry>;
private remintedIds: Map<string, string>;
/**
* Resolve a legacy ID to its imported ID. Discarded records resolve to the minted ID of their
* nearest retained ancestor.
*/
type ResolveLegacyId = (legacyId: string | null) => string | null;
/**
* @param entriesById Complete inventory of legacy physical nodes, including discarded nodes.
* @param remintedIds Legacy-to-current ID map containing retained nodes only.
*/
constructor(entriesById: Map<string, LegacyV3Entry>, remintedIds: Map<string, string>) {
this.entriesById = entriesById;
this.remintedIds = remintedIds;
}
/**
* Resolve a legacy node to its reminted ID, or to its nearest retained ancestor when discarded.
* Returns null when the reference is null or no retained ancestor exists.
*/
resolve(legacyId: string | null): string | null {
const traversedIds: string[] = [];
const visitedIds = new Set<string>();
let currentId = legacyId;
let resolvedId: string | null = null;
while (currentId !== null) {
const remintedId = this.remintedIds.get(currentId);
if (remintedId !== undefined) {
resolvedId = remintedId;
break;
}
if (this.resolvedIds.has(currentId)) {
resolvedId = this.resolvedIds.get(currentId) ?? null;
break;
}
if (visitedIds.has(currentId)) throw new Error(`Cycle in legacy v3 parent chain at entry: ${currentId}`);
visitedIds.add(currentId);
const entry = this.entriesById.get(currentId);
if (entry === undefined) throw new Error(`Missing legacy v3 entry reference: ${currentId}`);
traversedIds.push(currentId);
currentId = entry.parentId;
}
for (const traversedId of traversedIds) this.resolvedIds.set(traversedId, resolvedId);
return resolvedId;
}
/** Parent mappings are already folded during the scan; later references need only a lookup. */
function createLegacyIdResolver(entries: ReadonlyMap<string, LegacyV3IndexEntry>): ResolveLegacyId {
return (legacyId) => {
if (legacyId === null) return null;
const mappedId = entries.get(legacyId)?.mappedId;
if (mappedId === undefined) throw new Error(`Missing legacy v3 entry reference: ${legacyId}`);
return mappedId;
};
}
function requireRetainedId(resolver: RetainedIdResolver, legacyId: string): string {
const importedId = resolver.resolve(legacyId);
if (importedId === null) throw new Error(`Legacy v3 entry reference has no retained ancestor: ${legacyId}`);
return importedId;
}
function resolveBranchSummaryFromId(resolver: RetainedIdResolver, legacyFromId: string): string | null {
function resolveBranchSummaryFromId(resolveLegacyId: ResolveLegacyId, legacyFromId: string): string | null {
// Legacy branchWithSummary() encoded a root source as the "root" sentinel instead of null.
return legacyFromId === "root" ? null : resolver.resolve(legacyFromId);
return legacyFromId === "root" ? null : resolveLegacyId(legacyFromId);
}
function projectContextMessages(entry: LegacyV3Entry, resolver: RetainedIdResolver): AgentMessage[] {
function projectContextMessage(entry: LegacyV3Entry, resolveLegacyId: ResolveLegacyId): AgentMessage | undefined {
switch (entry.type) {
case "message":
return [entry.message];
return entry.message;
case "custom_message":
return [importedCustomMessage(entry)];
return importedCustomMessage(entry);
case "branch_summary":
return entry.summary
? [
createBranchSummaryMessage(
entry.summary,
resolveBranchSummaryFromId(resolver, entry.fromId),
entry.timestamp,
),
]
: [];
? createBranchSummaryMessage(
entry.summary,
resolveBranchSummaryFromId(resolveLegacyId, entry.fromId),
entry.timestamp,
)
: undefined;
case "compaction":
return [createCompactionSummaryMessage(entry.summary, entry.tokensBefore, entry.timestamp)];
return createCompactionSummaryMessage(entry.summary, entry.tokensBefore, entry.timestamp);
case "custom":
case "model_change":
case "thinking_level_change":
case "active_tools_change":
case "session_info":
case "label":
return [];
return undefined;
}
}
function materializeRetainedTail(
compaction: LegacyV3CompactionEntry,
entriesById: ReadonlyMap<string, LegacyV3Entry>,
resolver: RetainedIdResolver,
): AgentMessage[] {
const reversedTail: LegacyV3Entry[] = [];
const visited = new Set<string>();
/** Walk physical ancestry, including discarded nodes and the exact kept boundary. */
function* retainedTailStructure(
compaction: LegacyV3CompactionIndexEntry,
entriesById: ReadonlyMap<string, LegacyV3IndexEntry>,
): Iterable<LegacyV3IndexEntry> {
let currentId = compaction.parentId;
while (currentId !== null) {
if (visited.has(currentId)) throw new Error(`Cycle in legacy v3 parent chain at entry: ${currentId}`);
visited.add(currentId);
const entry = entriesById.get(currentId);
if (entry === undefined) throw new Error(`Missing legacy v3 parent entry: ${currentId}`);
reversedTail.push(entry);
if (currentId === compaction.firstKeptEntryId) {
return reversedTail.reverse().flatMap((tailEntry) => projectContextMessages(tailEntry, resolver));
}
// The scan guarantees that every parent exists earlier in the file, so cycles are impossible.
const entry = entriesById.get(currentId)!;
yield entry;
if (currentId === compaction.firstKeptEntryId) return;
currentId = entry.parentId;
}
throw new Error(
@ -331,15 +310,15 @@ function materializeRetainedTail(
function normalizeRetainedEntry(
entry: RetainedLegacyV3Entry,
seq: number,
entriesById: ReadonlyMap<string, LegacyV3Entry>,
resolver: RetainedIdResolver,
indexed: RetainedLegacyV3IndexEntry,
retainedTail: AgentMessage[],
resolveLegacyId: ResolveLegacyId,
): CommittedEntryWrite {
const committedBase = {
kind: "entry" as const,
id: requireRetainedId(resolver, entry.id),
parentId: resolver.resolve(entry.parentId),
seq,
id: indexed.mappedId,
parentId: resolveLegacyId(indexed.parentId),
seq: indexed.seq,
timestamp: Date.parse(entry.timestamp),
};
if (entry.type === "message") {
@ -356,7 +335,7 @@ function normalizeRetainedEntry(
return {
...committedBase,
type: "branch_summary",
fromId: resolveBranchSummaryFromId(resolver, entry.fromId),
fromId: resolveBranchSummaryFromId(resolveLegacyId, entry.fromId),
summary: entry.summary,
details: entry.details,
usage: entry.usage,
@ -368,7 +347,7 @@ function normalizeRetainedEntry(
...committedBase,
type: "compaction",
summary: entry.summary,
retainedTail: materializeRetainedTail(entry, entriesById, resolver),
retainedTail,
tokensBefore: entry.tokensBefore,
details: entry.details,
usage: entry.usage,
@ -383,44 +362,33 @@ function normalizeRetainedEntry(
};
}
const THINKING_LEVELS = new Set<ThinkingLevel>(["off", "minimal", "low", "medium", "high", "xhigh", "max"]);
function selectedConfiguration(
entriesById: ReadonlyMap<string, LegacyV3Entry>,
entriesById: ReadonlyMap<string, LegacyV3IndexEntry>,
selectedId: string | null,
): LaneConfiguration | undefined {
const remaining = new Set<LegacyV3IndexEntry["type"]>([
"model_change",
"thinking_level_change",
"active_tools_change",
]);
let model: LaneConfiguration["model"] | undefined;
let thinkingLevel: ThinkingLevel | undefined;
let activeToolNames: string[] | undefined;
let sawModel = false;
let sawThinkingLevel = false;
let sawActiveToolNames = false;
const visited = new Set<string>();
let currentId = selectedId;
while (currentId !== null && (!sawModel || !sawThinkingLevel || !sawActiveToolNames)) {
if (visited.has(currentId)) throw new Error(`Cycle in legacy v3 parent chain at entry: ${currentId}`);
visited.add(currentId);
const entry = entriesById.get(currentId);
if (entry === undefined) throw new Error(`Missing legacy v3 entry reference: ${currentId}`);
if (entry.type === "model_change" && !sawModel) {
sawModel = true;
if (
typeof entry.provider === "string" &&
entry.provider.length !== 0 &&
typeof entry.modelId === "string" &&
entry.modelId.length !== 0
) {
model = { provider: entry.provider, modelId: entry.modelId };
}
} else if (entry.type === "thinking_level_change" && !sawThinkingLevel) {
sawThinkingLevel = true;
if (typeof entry.thinkingLevel === "string" && THINKING_LEVELS.has(entry.thinkingLevel as ThinkingLevel)) {
thinkingLevel = entry.thinkingLevel as ThinkingLevel;
}
} else if (entry.type === "active_tools_change" && !sawActiveToolNames) {
sawActiveToolNames = true;
if (Array.isArray(entry.activeToolNames) && entry.activeToolNames.every((name) => typeof name === "string")) {
activeToolNames = [...entry.activeToolNames];
while (currentId !== null && remaining.size !== 0) {
const entry = entriesById.get(currentId)!;
// Consume the nearest change even when invalid: older values must not become fallbacks.
if (remaining.delete(entry.type)) {
switch (entry.type) {
case "model_change":
model = { provider: entry.provider, modelId: entry.modelId };
break;
case "thinking_level_change":
thinkingLevel = entry.thinkingLevel;
break;
case "active_tools_change":
activeToolNames = [...entry.activeToolNames];
break;
}
}
currentId = entry.parentId;
@ -429,173 +397,285 @@ function selectedConfiguration(
return { model, thinkingLevel, activeToolNames: activeToolNames ?? [] };
}
function normalizeLegacyV3Values(
entries: readonly LegacyV3Entry[],
entriesById: ReadonlyMap<string, LegacyV3Entry>,
resolver: RetainedIdResolver,
firstSeq: number,
): CommittedValueSetWrite[] {
const writes: CommittedValueSetWrite[] = [];
let latestSessionInfo: LegacyV3SessionInfoEntry | undefined;
for (const entry of entries) {
if (entry.type === "session_info") latestSessionInfo = entry;
function indexLegacyV3Entry(
entry: LegacyV3Entry,
lineNumber: number,
seq: number,
entries: ReadonlyMap<string, LegacyV3IndexEntry>,
): LegacyV3IndexEntry {
// SessionManager appends after an existing leaf and writes extracted branches in parent order.
const mappedParentId = entry.parentId === null ? null : entries.get(entry.parentId)?.mappedId;
if (mappedParentId === undefined) {
throw new Error(
`Legacy v3 entry ${entry.id} has a missing or forward parent at line ${lineNumber}: ${entry.parentId}`,
);
}
if (latestSessionInfo?.name) {
writes.push({
kind: "value",
op: "set",
seq: firstSeq + writes.length,
namespace: sessionName.namespace,
key: sessionName.key,
value: latestSessionInfo.name,
});
const structure: LegacyV3IndexBase = {
id: entry.id,
parentId: entry.parentId,
mappedId: mappedParentId,
};
if (!isRetainedEntry(entry)) {
switch (entry.type) {
case "label":
return { ...structure, type: entry.type, targetId: entry.targetId, label: entry.label };
case "model_change":
return { ...structure, type: entry.type, provider: entry.provider, modelId: entry.modelId };
case "thinking_level_change":
return { ...structure, type: entry.type, thinkingLevel: entry.thinkingLevel };
case "active_tools_change":
return { ...structure, type: entry.type, activeToolNames: entry.activeToolNames };
case "session_info":
return { ...structure, type: entry.type };
}
}
// Null denotes the root, not a retained node identity that can be reminted.
if (entry.id === null) throw new Error("Legacy v3 entry reference has no retained ancestor: null");
const retained = {
...structure,
mappedId: uuidv7(Date.parse(entry.timestamp)),
seq,
};
switch (entry.type) {
case "branch_summary":
return { ...retained, type: entry.type, fromId: entry.fromId };
case "compaction":
return { ...retained, type: entry.type, firstKeptEntryId: entry.firstKeptEntryId };
default:
return { ...retained, type: entry.type };
}
}
function legacyEntryUsage(entry: LegacyV3Entry): Usage | undefined {
switch (entry.type) {
case "message":
return entry.message.role === "assistant" || entry.message.role === "toolResult"
? entry.message.usage
: undefined;
case "compaction":
case "branch_summary":
return entry.usage;
default:
return undefined;
}
}
async function readLegacyV3Inventory(reader: TextLineReader, context: Context): Promise<LegacyV3Inventory> {
const entries = new Map<string, LegacyV3IndexEntry>();
let nextSeq = 1;
let importedUsage = emptyUsage();
let name: string | undefined;
let finalId: string | null = null;
while (true) {
const line = fileValue(await reader.readLine(context), "Failed to read legacy v3 source");
if (line === undefined || !line.terminated) break;
const lineNumber = entries.size + 2;
const entry = parseLegacyV3Entry(line.text);
if (entries.has(entry.id)) throw new Error(`Duplicate legacy v3 entry id: ${entry.id}`);
const indexed = indexLegacyV3Entry(entry, lineNumber, nextSeq, entries);
entries.set(entry.id, indexed);
if (isRetainedEntry(indexed)) nextSeq++;
finalId = entry.id;
if (entry.type === "session_info") name = entry.name;
const usage = legacyEntryUsage(entry);
if (usage !== undefined) importedUsage = addUsage(importedUsage, usage);
}
return { entries, nextSeq, importedUsage, name, finalId };
}
function normalizeLegacyV3Values(inventory: LegacyV3Inventory): CommittedValueSetWrite[] {
const { entries, name, finalId } = inventory;
const resolveLegacyId = createLegacyIdResolver(entries);
let nextSeq = inventory.nextSeq;
const values: CommittedValueSetWrite[] = [];
// session name
if (name) values.push({ ...setValue(sessionName, name), seq: nextSeq++ });
// labels
const labels = new Map<string, string>();
for (const entry of entries) {
for (const entry of entries.values()) {
if (entry.type !== "label") continue;
const targetId = resolver.resolve(entry.targetId);
// Labels have no current address when their target has no retained ancestor.
const targetId = resolveLegacyId(entry.targetId);
if (targetId === null) continue;
// Legacy v3 treated both undefined and the empty string as clearing a label.
if (entry.label) labels.set(targetId, entry.label);
else labels.delete(targetId);
}
for (const [targetId, label] of labels) {
const address = entryLabel(targetId);
writes.push({
kind: "value",
op: "set",
seq: firstSeq + writes.length,
namespace: address.namespace,
key: address.key,
value: label,
});
values.push({ ...setValue(entryLabel(targetId), label), seq: nextSeq++ });
}
const finalEntry = entries.at(-1);
const tipAddress = branchTip("main");
writes.push({
kind: "value",
op: "set",
seq: firstSeq + writes.length,
namespace: tipAddress.namespace,
key: tipAddress.key,
value: finalEntry === undefined ? null : resolver.resolve(finalEntry.id),
});
const configuration = selectedConfiguration(entriesById, finalEntry?.id ?? null);
// branch tip
values.push({ ...setValue(branchTip("main"), resolveLegacyId(finalId)), seq: nextSeq++ });
// configuration
const configuration = selectedConfiguration(entries, finalId);
if (configuration !== undefined) {
const configAddress = laneConfig("main");
writes.push({
kind: "value",
op: "set",
seq: firstSeq + writes.length,
namespace: configAddress.namespace,
key: configAddress.key,
value: configuration,
});
const stateAddress = laneState("main");
writes.push({
kind: "value",
op: "set",
seq: firstSeq + writes.length,
namespace: stateAddress.namespace,
key: stateAddress.key,
value: { currentOperationId: null, lastOperationId: null, inbox: [] },
values.push({ ...setValue(laneConfig("main"), configuration), seq: nextSeq++ });
values.push({
...setValue(laneState("main"), { currentOperationId: null, lastOperationId: null, inbox: [] }),
seq: nextSeq++,
});
}
return writes;
return values;
}
function aggregateImportedUsage(entries: readonly LegacyV3Entry[]): Usage {
let aggregate = emptyUsage();
for (const entry of entries) {
let usage: Usage | undefined;
if (entry.type === "message") {
if (entry.message.role === "assistant") usage = entry.message.usage;
else if (entry.message.role === "toolResult") usage = entry.message.usage;
} else if (entry.type === "compaction" || entry.type === "branch_summary") {
usage = entry.usage;
}
if (usage !== undefined) aggregate = addUsage(aggregate, usage);
/**
* A captured legacy file exposed as repeatable logical v4 writes.
* Each pass reopens the path; callers must not replace or edit the source between passes.
* Structural indexes, label/configuration metadata, and derived current values survive between scans.
*/
export class LegacyV3Source {
readonly header: JsonlStorageHeader;
readonly importedUsage: Usage;
readonly nextSeq: number;
readonly values: readonly CommittedValueSetWrite[];
private readonly fileSystem: FileSystem;
private readonly path: string;
private readonly entries: ReadonlyMap<string, LegacyV3IndexEntry>;
private readonly resolveLegacyId: ResolveLegacyId;
private constructor(
fileSystem: FileSystem,
path: string,
header: JsonlStorageHeader,
entries: ReadonlyMap<string, LegacyV3IndexEntry>,
importedUsage: Usage,
values: readonly CommittedValueSetWrite[],
nextSeq: number,
) {
this.fileSystem = fileSystem;
this.path = path;
this.header = header;
this.entries = entries;
this.resolveLegacyId = createLegacyIdResolver(entries);
this.importedUsage = importedUsage;
this.values = values;
this.nextSeq = nextSeq;
}
return aggregate;
}
/** Normalize the currently supported v3 records without touching their source file. */
export function normalizeLegacyV3Records(recordLines: readonly string[]): NormalizedLegacyV3Records {
const entries = recordLines.map((line, index) => parseLegacyV3Entry(line, index + 2));
const entriesById = new Map<string, LegacyV3Entry>();
for (const entry of entries) {
if (entriesById.has(entry.id)) throw new Error(`Duplicate legacy v3 entry id: ${entry.id}`);
entriesById.set(entry.id, entry);
}
const retainedEntries = entries.filter(isRetainedEntry);
const remintedIds = new Map(retainedEntries.map((entry) => [entry.id, uuidv7(Date.parse(entry.timestamp))]));
const resolver = new RetainedIdResolver(entriesById, remintedIds);
const entryWrites = retainedEntries.map((entry, index) =>
normalizeRetainedEntry(entry, index + 1, entriesById, resolver),
);
const valueWrites = normalizeLegacyV3Values(entries, entriesById, resolver, entryWrites.length + 1);
const writes: CommittedWrite[] = [...entryWrites, ...valueWrites];
return {
writes,
importedUsage: aggregateImportedUsage(entries),
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}`,
/**
* Scan complete v3 records without modifying the file, ignoring an unterminated final line.
* Build parent mappings, assign IDs stable for this source instance, and derive current values
* and imported usage. Retain metadata, not conversation payloads or an open reader. writes()
* reopens the path to materialize captured records and resolve their payload-specific references.
*/
static async read(fileSystem: FileSystem, path: string, context: Context): Promise<LegacyV3Source> {
const reader = fileValue(
await fileSystem.openTextLineReader(path, context),
`Failed to open legacy v3 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}`);
try {
const parsed = await readJsonlHeader(reader, path, context);
if (parsed.format !== "v3-legacy") {
throw new Error(`Invalid legacy v3 JSONL storage ${path}: expected format 3 header`);
}
const inventory = await readLegacyV3Inventory(reader, context);
const values = normalizeLegacyV3Values(inventory);
return new LegacyV3Source(
fileSystem,
path,
await normalizeLegacyV3Header(fileSystem, parsed.header, context),
inventory.entries,
inventory.importedUsage,
values,
inventory.nextSeq + values.length,
);
} finally {
await reader.close(context);
}
}
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);
*entryStructures(): Iterable<Pick<CommittedEntryWrite, "id" | "parentId" | "seq">> {
for (const entry of this.entries.values()) {
if (!isRetainedEntry(entry)) continue;
yield { id: entry.mappedId, parentId: this.resolveLegacyId(entry.parentId), seq: entry.seq };
}
}
translateForkEntryId(legacyId: string): string {
const entry = this.entries.get(legacyId);
if (entry === undefined) throw new Error(`Legacy v3 fork entry does not exist: ${legacyId}`);
if (!isRetainedEntry(entry)) throw new Error(`Legacy v3 fork entry is not a retained entry: ${legacyId}`);
return entry.mappedId;
}
private collectRequiredTailMessageIds(isEntrySelected?: (id: string) => boolean): Set<string> {
const requiredIds = new Set<string>();
for (const entry of this.entries.values()) {
if (entry.type !== "compaction") continue;
const compactionIsSelected = isEntrySelected === undefined || isEntrySelected(entry.mappedId);
if (!compactionIsSelected) continue;
// Walk from the compaction's parent through firstKeptEntryId, inclusive.
for (const tailEntry of retainedTailStructure(entry, this.entries)) {
const canProduceContextMessage = isRetainedEntry(tailEntry) && tailEntry.type !== "custom";
if (canProduceContextMessage) requiredIds.add(tailEntry.id);
}
}
return requiredIds;
}
/**
* Stream normalized v4 entries, optionally filtered by reminted ID, followed by derived current values.
* Each pass owns its reader and message cache, sharing only the captured IDs and metadata.
*/
async *writes(
context: Context,
isEntrySelected?: (id: string) => boolean,
): AsyncIterable<CommittedEntryWrite | CommittedValueSetWrite> {
const requiredTailMessageIds = this.collectRequiredTailMessageIds(isEntrySelected);
// Keep needed context messages for this entire pass; tails may revisit old or shared branches.
const tailMessagesByLegacyId = new Map<string, AgentMessage>();
for await (const { entry, indexed } of this.readCapturedEntries(context)) {
if (requiredTailMessageIds.has(entry.id)) {
const message = projectContextMessage(entry, this.resolveLegacyId);
if (message !== undefined) tailMessagesByLegacyId.set(entry.id, message);
}
if (!isRetainedEntry(indexed) || !isRetainedEntry(entry)) continue;
if (isEntrySelected !== undefined && !isEntrySelected(indexed.mappedId)) continue;
const retainedTail: AgentMessage[] = [];
if (indexed.type === "compaction") {
for (const ancestor of retainedTailStructure(indexed, this.entries)) {
const message = tailMessagesByLegacyId.get(ancestor.id);
if (message !== undefined) retainedTail.push(message);
}
retainedTail.reverse();
}
yield normalizeRetainedEntry(entry, indexed, retainedTail, this.resolveLegacyId);
}
yield* this.values;
}
/** Replay only the captured prefix and verify its physical identities before materialization. */
private async *readCapturedEntries(
context: Context,
): AsyncIterable<{ entry: LegacyV3Entry; indexed: LegacyV3IndexEntry }> {
const reader = fileValue(
await this.fileSystem.openTextLineReader(this.path, context),
`Failed to reopen legacy v3 source ${this.path}`,
);
try {
const parsed = await readJsonlHeader(reader, this.path, context);
if (
parsed.format !== "v3-legacy" ||
parsed.header.id !== this.header.id ||
parsed.header.cwd !== this.header.cwd
) {
throw new Error("Legacy v3 source header changed");
}
for (const indexed of this.entries.values()) {
const line = fileValue(await reader.readLine(context), "Failed to reread legacy v3 source");
if (line === undefined || !line.terminated)
throw new Error("Legacy v3 source ended before captured entries");
const entry = parseLegacyV3Entry(line.text);
if (entry.id !== indexed.id || entry.type !== indexed.type) {
throw new Error("Legacy v3 source changed");
}
yield { entry, indexed };
}
} finally {
await reader.close(context);
}
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,
};
}

96
packages/agent/src/harness/session/jsonl/repo.ts
View file

@ -4,9 +4,9 @@ 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 { fileValue, serializeJsonlTransaction } from "./io.ts";
import { metadataFromLegacyV3Header, readNormalizedLegacyV3Source } from "./legacy-v3.ts";
import { type JsonlForkInput, runJsonlFork } from "./fork.ts";
import { fileValue } from "./io.ts";
import { LegacyV3Source, metadataFromLegacyV3Header } from "./legacy-v3.ts";
import { JsonlStorage } from "./storage.ts";
import {
JSONL_FORMAT_VERSION,
@ -42,8 +42,6 @@ function sessionFileName(createdAt: number, id: string): string {
return `${timestamp}_${encodeURIComponent(id)}.jsonl`;
}
type ResolvedJsonlForkSource = JsonlForkSource | { kind: "legacy-v3"; metadata: JsonlSessionMetadata };
/** File-backed format-4 session repository lifecycle. */
export class JsonlSessionRepo
implements SessionRepo<JsonlSessionMetadata, JsonlSessionCreateOptions, JsonlSessionListOptions>
@ -164,7 +162,7 @@ export class JsonlSessionRepo
let path: string | undefined;
let storage: JsonlStorage | undefined;
try {
const resolvedSource = await this.resolveForkSource(source, sourceStorage, context);
const input = await this.resolveForkInput(source, sourceStorage, context);
path = await this.resolveNewSessionPath(cwd, createdAt, id, context);
const header: Omit<JsonlStorageHeader, "nextSeq"> = {
v: JSONL_FORMAT_VERSION,
@ -175,7 +173,17 @@ export class JsonlSessionRepo
cwd,
parentSessionId: source.id,
};
storage = await this.createForkStorage(path, header, resolvedSource, options, context);
await runJsonlFork(
{
input,
fileSystem: this.fileSystem,
destinationPath: path,
destinationHeader: header,
fork: options,
},
context,
);
storage = await JsonlStorage.open({ fileSystem: this.fileSystem, path, now: this.now }, context);
const info = fileValue(await this.fileSystem.fileInfo(path, context), `Failed to read session ${path}`);
return this.publishOpenSession(metadataFromHeader(header, path, info.mtimeMs), storage, destinationKey);
} catch (error) {
@ -288,11 +296,11 @@ export class JsonlSessionRepo
if (idExists) throw new Error(`Session already exists: ${id}`);
}
private async resolveForkSource(
private async resolveForkInput(
source: JsonlSessionMetadata,
storage: JsonlStorage | undefined,
context: Context,
): Promise<ResolvedJsonlForkSource> {
): Promise<JsonlForkInput> {
if (storage !== undefined) {
if (storage.isLegacyV3()) {
throw new Error(
@ -302,72 +310,14 @@ export class JsonlSessionRepo
const nextSeq = await storage.captureForkNextSeq(context);
return { kind: "open", metadata: source, nextSeq };
}
if (await this.isLegacyV3ForkSource(source, context)) return { kind: "legacy-v3", metadata: source };
return { kind: "closed", metadata: source };
}
private async createForkStorage(
path: string,
header: Omit<JsonlStorageHeader, "nextSeq">,
source: ResolvedJsonlForkSource,
fork: ForkOptions,
context: Context,
): Promise<JsonlStorage> {
if (source.kind === "legacy-v3") {
return this.createLegacyV3ForkStorage(path, header, source.metadata, fork, context);
}
await runJsonlFork(
{
source,
fileSystem: this.fileSystem,
destinationPath: path,
destinationHeader: header,
fork,
},
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}`,
);
if (await this.isLegacyV3ForkSource(source, context)) {
const normalized = await LegacyV3Source.read(this.fileSystem, source.path, context);
if (normalized.header.id !== source.id || normalized.header.cwd !== source.cwd) {
throw new Error(`Session identity does not match header: ${source.id}`);
}
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);
return { kind: "legacy-v3", normalized };
}
return { kind: "closed", metadata: source };
}
private async isLegacyV3ForkSource(source: JsonlSessionMetadata, context: Context): Promise<boolean> {

102
packages/agent/src/harness/session/jsonl/storage.ts
View file

@ -1,4 +1,3 @@
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 { FileSystem } from "../../types.ts";
@ -17,15 +16,17 @@ import type {
Write,
} from "../types.ts";
import type { ListElement, ListReadOptions, StoredValue, Value, ValueList } from "../values.ts";
import { type LegacyV3SessionHeader, parseJsonlSessionHeader } from "./codec.ts";
import { fileValue, parseJsonlTransaction, publishFileAtomically, serializeJsonlTransaction } from "./io.ts";
import { normalizeLegacyV3Header, normalizeLegacyV3Records } from "./legacy-v3.ts";
import {
fileValue,
parseJsonlTransaction,
publishFileAtomically,
publishJsonl,
readJsonlHeader,
serializeJsonlTransaction,
} from "./io.ts";
import { LegacyV3Source } from "./legacy-v3.ts";
import { JSONL_STORAGE_VERSION, type JsonlStorageHeader, type JsonlStorageOptions } from "./types.ts";
function serializeStorage(header: JsonlStorageHeader, transactions: CommittedWrite[][]): string {
return `${[JSON.stringify(header), ...transactions.map(serializeJsonlTransaction)].join("\n")}\n`;
}
function splitCompleteLines(content: string): { lines: string[]; torn: boolean } {
if (content.endsWith("\n")) return { lines: content.slice(0, -1).split("\n"), torn: false };
const lastNewline = content.lastIndexOf("\n");
@ -33,13 +34,7 @@ function splitCompleteLines(content: string): { lines: string[]; torn: boolean }
return { lines: content.slice(0, lastNewline).split("\n"), torn: true };
}
type LegacyV3Backing = {
kind: "v3";
importedUsage: Usage;
baselineWrites: readonly CommittedWrite[];
};
type JsonlBacking = { kind: "v4" } | LegacyV3Backing;
type JsonlBacking = { kind: "v4" } | { kind: "v3"; source: LegacyV3Source };
/** JSONL storage backed by an injected filesystem capability. */
export class JsonlStorage implements Storage {
@ -69,30 +64,34 @@ export class JsonlStorage implements Storage {
): Promise<JsonlStorage> {
const storage = new JsonlStorage(options, header, { kind: "v4" });
const prepared = storage.storageState.prepareCommit(initialWrites, storage.now());
const transactions = prepared.writes.length === 0 ? [] : [prepared.writes];
await publishFileAtomically(options.fileSystem, options.path, serializeStorage(header, transactions), context);
await publishJsonl(options.fileSystem, options.path, header, context, async (append) => {
if (prepared.writes.length !== 0) await append(prepared.writes);
});
storage.storageState.applyValidated(prepared.writes);
return storage;
}
static async open(options: JsonlStorageOptions, context: Context): Promise<JsonlStorage> {
const reader = fileValue(
await options.fileSystem.openTextLineReader(options.path, context),
`Failed to read JSONL storage ${options.path}`,
);
const parsed = await readJsonlHeader(reader, options.path, context).finally(() => reader.close(context));
return parsed.format === "v3-legacy"
? JsonlStorage.openLegacyV3(options, context)
: JsonlStorage.openV4(options, parsed.header, context);
}
private static async openV4(
options: JsonlStorageOptions,
header: JsonlStorageHeader,
context: Context,
): Promise<JsonlStorage> {
const content = fileValue(
await options.fileSystem.readTextFile(options.path, context),
`Failed to read JSONL storage ${options.path}`,
);
const { lines, torn } = splitCompleteLines(content);
if (lines[0] === undefined || lines[0] === "") {
throw new Error(`Invalid JSONL storage ${options.path}: missing header`);
}
const parsedHeader = parseJsonlSessionHeader(lines[0]);
if (!parsedHeader.ok) {
throw new Error(`Invalid JSONL storage ${options.path}: invalid header`, { cause: parsedHeader.error });
}
if (parsedHeader.value.format === "v3-legacy") {
return JsonlStorage.openLegacyV3(options, parsedHeader.value.header, lines.slice(1), context);
}
const header = parsedHeader.value.header;
if (header.storageVersion !== JSONL_STORAGE_VERSION) {
throw new Error(`Session ${header.id} uses unsupported storage version ${header.storageVersion}`);
}
@ -106,27 +105,18 @@ export class JsonlStorage implements Storage {
}
}
if (header.nextSeq !== undefined) storage.storageState.advanceNextSeq(header.nextSeq);
if (torn) await publishFileAtomically(options.fileSystem, options.path, `${lines.join("\n")}\n`, context);
if (torn) {
await publishFileAtomically(options.fileSystem, options.path, context, (append) =>
append(`${lines.join("\n")}\n`),
);
}
return storage;
}
private static async openLegacyV3(
options: JsonlStorageOptions,
header: LegacyV3SessionHeader,
recordLines: readonly string[],
context: Context,
): Promise<JsonlStorage> {
const { writes, importedUsage, nextSeq } = normalizeLegacyV3Records(recordLines);
const targetHeader = {
...(await normalizeLegacyV3Header(options.fileSystem, header, context)),
nextSeq,
};
const storage = new JsonlStorage(options, targetHeader, {
kind: "v3",
importedUsage,
baselineWrites: writes,
});
storage.replayCommitted(writes);
private static async openLegacyV3(options: JsonlStorageOptions, context: Context): Promise<JsonlStorage> {
const source = await LegacyV3Source.read(options.fileSystem, options.path, context);
const storage = new JsonlStorage(options, { ...source.header, nextSeq: source.nextSeq }, { kind: "v3", source });
for await (const write of source.writes(context)) storage.replayCommitted([write]);
return storage;
}
@ -147,7 +137,7 @@ export class JsonlStorage implements Storage {
private async applyCommit(writes: Write[], context: Context): Promise<CommitResult> {
if (this.backing.kind === "v3" && writes.length !== 0) {
return this.upgradeLegacyV3ToV4(this.backing, writes, context);
return this.upgradeLegacyV3ToV4(this.backing.source, writes, context);
}
const prepared = this.storageState.prepareCommit(writes, this.now());
if (prepared.writes.length !== 0) {
@ -162,7 +152,7 @@ export class JsonlStorage implements Storage {
/** Atomically upgrade legacy v3 backing and preserve the first caller write as a v4 transaction. */
private async upgradeLegacyV3ToV4(
backing: LegacyV3Backing,
source: LegacyV3Source,
callerWrites: Write[],
context: Context,
): Promise<CommitResult> {
@ -171,7 +161,7 @@ export class JsonlStorage implements Storage {
[
insertUsage({
id: uuidv7(timestamp),
usage: backing.importedUsage,
usage: source.importedUsage,
adjustment: true,
details: { source: "v3-import" },
}),
@ -182,12 +172,10 @@ export class JsonlStorage implements Storage {
const nextSeq = prepared.result.firstSeq + prepared.writes.length;
const upgradedHeader = { ...this.header, nextSeq };
await publishFileAtomically(
this.fileSystem,
this.path,
serializeStorage(upgradedHeader, [...backing.baselineWrites.map((write) => [write]), prepared.writes]),
context,
);
await publishJsonl(this.fileSystem, this.path, upgradedHeader, context, async (append) => {
for await (const write of source.writes(context)) await append([write]);
await append(prepared.writes);
});
const stats = this.storageState.applyValidated(prepared.writes);
this.backing = { kind: "v4" };
@ -250,7 +238,7 @@ export class JsonlStorage implements Storage {
}
private withImportedUsage(stats: SessionStats): SessionStats {
return this.backing.kind === "v4" ? stats : { ...stats, usage: this.backing.importedUsage };
return this.backing.kind === "v4" ? stats : { ...stats, usage: this.backing.source.importedUsage };
}
isLegacyV3(): boolean {

108
packages/agent/test/harness/jsonl-io.test.ts Normal file
View file

@ -0,0 +1,108 @@
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { BACKGROUND_CONTEXT } from "../../src/harness/context.ts";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import { publishFileAtomically, publishJsonl } from "../../src/harness/session/jsonl/io.ts";
import type { JsonlStorageHeader } from "../../src/harness/session/jsonl/types.ts";
import { sessionName, setValue } from "../../src/harness/session/values.ts";
import { err, FileError } from "../../src/harness/types.ts";
let root: string;
let path: string;
let fileSystem: NodeExecutionEnv;
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), "pi-jsonl-io-"));
path = join(root, "session.jsonl");
fileSystem = new NodeExecutionEnv({ cwd: root });
});
afterEach(async () => {
vi.restoreAllMocks();
await rm(root, { recursive: true, force: true });
});
describe("atomic file publication", () => {
it("keeps the destination unchanged until all content has been written", async () => {
await writeFile(path, "original");
await publishFileAtomically(fileSystem, path, BACKGROUND_CONTEXT, async (append) => {
await append("first\n");
expect(await readFile(`${path}.tmp`, "utf8")).toBe("first\n");
expect(await readFile(path, "utf8")).toBe("original");
await append("second\n");
expect(await readFile(path, "utf8")).toBe("original");
});
expect(await readFile(path, "utf8")).toBe("first\nsecond\n");
await expect(readFile(`${path}.tmp`)).rejects.toMatchObject({ code: "ENOENT" });
});
it("discards partial content and preserves the original error when the callback fails", async () => {
await writeFile(path, "original");
const failure = new Error("content generation failed");
await expect(
publishFileAtomically(fileSystem, path, BACKGROUND_CONTEXT, async (append) => {
await append("partial");
throw failure;
}),
).rejects.toBe(failure);
expect(await readFile(path, "utf8")).toBe("original");
await expect(readFile(`${path}.tmp`)).rejects.toMatchObject({ code: "ENOENT" });
});
it.each(["writeFile", "appendFile", "renameFile"] as const)(
"preserves the destination and allows retry after %s fails",
async (method) => {
await writeFile(path, "original");
const failure = new FileError("unknown", "injected I/O failure", path);
vi.spyOn(fileSystem, method).mockResolvedValueOnce(err(failure));
await expect(
publishFileAtomically(fileSystem, path, BACKGROUND_CONTEXT, (append) => append("replacement")),
).rejects.toMatchObject({ cause: failure });
expect(await readFile(path, "utf8")).toBe("original");
await expect(readFile(`${path}.tmp`)).rejects.toMatchObject({ code: "ENOENT" });
await publishFileAtomically(fileSystem, path, BACKGROUND_CONTEXT, (append) => append("retry"));
expect(await readFile(path, "utf8")).toBe("retry");
},
);
});
describe("JSONL publication", () => {
const header: JsonlStorageHeader = {
v: 4,
kind: "header",
id: "session",
storageVersion: 1,
createdAt: 1_700_000_000_000,
cwd: "/workspace",
nextSeq: 4,
};
it("writes the header first and preserves single-write and multi-write transaction boundaries", async () => {
const first = { ...setValue(sessionName, "first"), seq: 1 };
const second = { ...setValue(sessionName, "second"), seq: 2 };
const third = { ...setValue(sessionName, "third"), seq: 3 };
await publishJsonl(fileSystem, path, header, BACKGROUND_CONTEXT, async (append) => {
await append([first]);
await append([second, third]);
});
expect(await readFile(path, "utf8")).toBe(
[JSON.stringify(header), JSON.stringify(first), JSON.stringify([second, third]), ""].join("\n"),
);
});
it("publishes a header-only file when the callback emits no transactions", async () => {
await publishJsonl(fileSystem, path, header, BACKGROUND_CONTEXT, async () => {});
expect(await readFile(path, "utf8")).toBe(`${JSON.stringify(header)}\n`);
});
});

207
packages/agent/test/harness/jsonl-v3-migration.test.ts
View file

@ -303,6 +303,109 @@ describe("JSONL v3 migration", () => {
expect(getOrThrow(await fileSystem.readTextFile(path, BACKGROUND_CONTEXT))).toBe(content);
});
it("forks a configured closed source at its main tip when entryId is omitted", async () => {
await writeLegacyV3Fixture([
{
type: "model_change",
id: "model",
parentId: null,
timestamp: new Date(NOW + 1_000).toISOString(),
provider: "anthropic",
modelId: "claude-sonnet-4-5",
},
{
type: "thinking_level_change",
id: "thinking",
parentId: "model",
timestamp: new Date(NOW + 2_000).toISOString(),
thinkingLevel: "high",
},
{
type: "message",
id: "tip",
parentId: "thinking",
timestamp: new Date(NOW + 3_000).toISOString(),
message: firstMessage,
},
]);
const [metadata] = await repo.list({ cwd: "/workspace" }, BACKGROUND_CONTEXT);
if (metadata === undefined) throw new Error("Legacy fixture was not discovered");
const fork = await repo.fork(
metadata,
{ id: "branch-fork", scope: "branch", branch: "main" },
BACKGROUND_CONTEXT,
);
const entries = await fork.findEntries({ order: "asc" }, BACKGROUND_CONTEXT);
expect(entries).toHaveLength(1);
expect(await mainTip(fork)).toBe(entries[0]!.id);
expect((await fork.getValue(storedValues.laneConfig("main"), BACKGROUND_CONTEXT))?.value).toEqual({
model: { provider: "anthropic", modelId: "claude-sonnet-4-5" },
thinkingLevel: "high",
activeToolNames: [],
});
expect((await fork.getValue(storedValues.laneState("main"), BACKGROUND_CONTEXT))?.value).toEqual({
currentOperationId: null,
lastOperationId: null,
inbox: [],
});
await fork.close(BACKGROUND_CONTEXT);
});
it("forks a configured closed source at an original legacy entry id", async () => {
await writeLegacyV3Fixture([
{
type: "model_change",
id: "model",
parentId: null,
timestamp: new Date(NOW + 1_000).toISOString(),
provider: "anthropic",
modelId: "claude-sonnet-4-5",
},
{
type: "thinking_level_change",
id: "thinking",
parentId: "model",
timestamp: new Date(NOW + 2_000).toISOString(),
thinkingLevel: "high",
},
{
type: "message",
id: "message-1",
parentId: "thinking",
timestamp: new Date(NOW + 3_000).toISOString(),
message: firstMessage,
},
{
type: "message",
id: "message-2",
parentId: "message-1",
timestamp: new Date(NOW + 4_000).toISOString(),
message: secondMessage,
},
{
type: "message",
id: "message-3",
parentId: "message-2",
timestamp: new Date(NOW + 5_000).toISOString(),
message: firstMessage,
},
]);
const [metadata] = await repo.list({ cwd: "/workspace" }, BACKGROUND_CONTEXT);
if (metadata === undefined) throw new Error("Legacy fixture was not discovered");
const fork = await repo.fork(
metadata,
{ id: "entry-fork", scope: "branch", branch: "main", entryId: "message-2" },
BACKGROUND_CONTEXT,
);
const entries = await fork.findEntries({ order: "asc" }, BACKGROUND_CONTEXT);
expect(entries).toHaveLength(2);
expect(entries[1]).toMatchObject({ parentId: entries[0]!.id, message: secondMessage });
expect(await mainTip(fork)).toBe(entries[1]!.id);
await fork.close(BACKGROUND_CONTEXT);
});
it("rejects an open v3 source until a non-empty commit persists its format-4 ids", async () => {
const { path, content, metadata } = await writeForkFixture();
const source = await repo.open(metadata, BACKGROUND_CONTEXT);
@ -765,48 +868,6 @@ describe("JSONL v3 migration", () => {
await session.close(BACKGROUND_CONTEXT);
});
it("omits an unsupported nearest value instead of falling back to an older change", async () => {
await writeLegacyV3Fixture([
{
type: "message",
id: "root",
parentId: null,
timestamp: new Date(firstTimestamp).toISOString(),
message: firstMessage,
},
{
type: "model_change",
id: "older-model",
parentId: "root",
timestamp: new Date(modelChangeTimestamp).toISOString(),
provider: "anthropic",
modelId: "older",
},
{
type: "model_change",
id: "invalid-model",
parentId: "older-model",
timestamp: new Date(thinkingChangeTimestamp).toISOString(),
provider: "",
modelId: "",
},
{
type: "message",
id: "tip",
parentId: "invalid-model",
timestamp: new Date(secondTimestamp).toISOString(),
message: secondMessage,
},
]);
const [metadata] = await repo.list({ cwd: "/workspace" }, BACKGROUND_CONTEXT);
if (metadata === undefined) throw new Error("Legacy fixture was not discovered");
const session = await repo.open(metadata, BACKGROUND_CONTEXT);
expect(await session.getValue(storedValues.laneConfig("main"), BACKGROUND_CONTEXT)).toBeUndefined();
expect(await session.getValue(storedValues.laneState("main"), BACKGROUND_CONTEXT)).toBeUndefined();
await session.close(BACKGROUND_CONTEXT);
});
it.each([
{
name: "missing model",
@ -833,26 +894,6 @@ describe("JSONL v3 migration", () => {
},
],
},
{
name: "invalid thinking level",
changes: [
{
type: "model_change",
id: "model",
parentId: "root",
timestamp: new Date(modelChangeTimestamp).toISOString(),
provider: "anthropic",
modelId: "selected",
},
{
type: "thinking_level_change",
id: "thinking",
parentId: "model",
timestamp: new Date(thinkingChangeTimestamp).toISOString(),
thinkingLevel: "unsupported",
},
],
},
])("leaves main data-only for $name", async ({ changes }) => {
await writeLegacyV3Fixture([
{
@ -880,48 +921,6 @@ describe("JSONL v3 migration", () => {
await session.close(BACKGROUND_CONTEXT);
});
it("normalizes malformed active-tool history without compatibility state", async () => {
await writeLegacyV3Fixture([
{
type: "model_change",
id: "model",
parentId: null,
timestamp: new Date(modelChangeTimestamp).toISOString(),
provider: "anthropic",
modelId: "selected",
},
{
type: "thinking_level_change",
id: "thinking",
parentId: "model",
timestamp: new Date(thinkingChangeTimestamp).toISOString(),
thinkingLevel: "high",
},
{
type: "active_tools_change",
id: "tools",
parentId: "thinking",
timestamp: new Date(activeToolsChangeTimestamp).toISOString(),
activeToolNames: ["read", 42],
},
]);
const [metadata] = await repo.list({ cwd: "/workspace" }, BACKGROUND_CONTEXT);
if (metadata === undefined) throw new Error("Legacy fixture was not discovered");
const session = await repo.open(metadata, BACKGROUND_CONTEXT);
expect((await session.getValue(storedValues.laneConfig("main"), BACKGROUND_CONTEXT))?.value).toEqual({
model: { provider: "anthropic", modelId: "selected" },
thinkingLevel: "high",
activeToolNames: [],
});
expect((await session.getValue(storedValues.laneState("main"), BACKGROUND_CONTEXT))?.value).toEqual({
currentOperationId: null,
lastOperationId: null,
inbox: [],
});
await session.close(BACKGROUND_CONTEXT);
});
it("resolves the main tip through trailing configuration changes", async () => {
await writeLegacyV3Fixture([
{

161
packages/agent/test/harness/jsonl-v3-stream.test.ts Normal file
View file

@ -0,0 +1,161 @@
import { beforeEach, describe, expect, it } from "vitest";
import { BACKGROUND_CONTEXT, type Context } from "../../src/harness/context.ts";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import type { CommittedEntryWrite, CommittedValueSetWrite } from "../../src/harness/session/commit.ts";
import type { LegacyV3SessionHeader } from "../../src/harness/session/jsonl/codec.ts";
import { LegacyV3Source } from "../../src/harness/session/jsonl/legacy-v3.ts";
import { type FileError, getOrThrow, ok, type Result, type TextLineReader } from "../../src/harness/types.ts";
import { createTempDir } from "./session-test-utils.ts";
const NOW = 1_700_000_000_000;
const HEADER: LegacyV3SessionHeader = {
type: "session",
version: 3,
id: "legacy",
timestamp: new Date(NOW).toISOString(),
cwd: "/workspace",
};
type OutputWrite = CommittedEntryWrite | CommittedValueSetWrite;
function entry(type: string, id: string, parentId: string | null, fields: Record<string, unknown> = {}) {
return { type, id, parentId, timestamp: new Date(NOW).toISOString(), ...fields };
}
function message(id: string, parentId: string | null, content = id) {
return entry("message", id, parentId, { message: { role: "user", content, timestamp: NOW } });
}
async function collect(source: LegacyV3Source, selected?: (id: string) => boolean): Promise<OutputWrite[]> {
const writes: OutputWrite[] = [];
for await (const write of source.writes(BACKGROUND_CONTEXT, selected)) writes.push(write);
return writes;
}
class ObservedEnv extends NodeExecutionEnv {
lineReads = 0;
opens = 0;
closes = 0;
override async openTextLineReader(path: string, context: Context): Promise<Result<TextLineReader, FileError>> {
const opened = await super.openTextLineReader(path, context);
if (!opened.ok) return opened;
this.opens++;
return ok({
readLine: (readContext: Context) => {
this.lineReads++;
return opened.value.readLine(readContext);
},
close: async (closeContext: Context) => {
this.closes++;
await opened.value.close(closeContext);
},
});
}
}
describe("streaming legacy v3 normalization", () => {
let fileSystem: ObservedEnv;
let fixtureId = 0;
beforeEach(() => {
fileSystem = new ObservedEnv({ cwd: createTempDir() });
});
async function fixture(records: readonly unknown[], suffix = "") {
const path = `legacy-${fixtureId++}.jsonl`;
const content = `${[HEADER, ...records].map((record) => JSON.stringify(record)).join("\n")}\n${suffix}`;
getOrThrow(await fileSystem.writeFile(path, content, BACKGROUND_CONTEXT));
return {
path,
content,
reader: fileSystem,
read: () => LegacyV3Source.read(fileSystem, path, BACKGROUND_CONTEXT),
};
}
it("materializes a selected compaction with its branch-local tail", async () => {
const { read, reader } = await fixture([
message("before", null),
entry("session_info", "boundary", "before"),
message("kept", "boundary", "Unicode é漢字"),
message("other", "before", "not on this branch"),
entry("compaction", "selected", "kept", {
summary: "selected",
firstKeptEntryId: "boundary",
tokensBefore: 20,
fromHook: true,
details: { a: 2 },
}),
]);
const source = await read();
const selectedId = [...source.entryStructures()].at(-1)!.id;
expect(await collect(source, () => false)).toEqual(source.values);
reader.lineReads = 0;
const writes = await collect(source, (id) => id === selectedId);
expect(writes).toHaveLength(2);
expect(writes[0]).toMatchObject({
type: "compaction",
id: selectedId,
seq: 4,
fromHook: true,
details: { a: 2 },
retainedTail: [{ role: "user", content: "Unicode é漢字", timestamp: NOW }],
});
expect(reader.lineReads).toBe(6); // Header plus each captured physical record, read once.
});
it("reuses an earlier cached message in selected compaction tails on different branches", async () => {
const { read, reader } = await fixture([
message("root", null),
message("left", "root"),
entry("compaction", "left-compaction", "left", {
summary: "left summary",
firstKeptEntryId: "root",
tokensBefore: 10,
}),
message("right", "root"),
entry("compaction", "right-compaction", "right", {
summary: "right summary",
firstKeptEntryId: "root",
tokensBefore: 20,
}),
]);
const source = await read();
const structures = [...source.entryStructures()];
const selected = new Set([structures[2]!.id, structures[4]!.id]);
reader.lineReads = 0;
const writes = await collect(source, (id) => selected.has(id));
expect(writes).toHaveLength(3);
expect(writes[0]).toMatchObject({
type: "compaction",
summary: "left summary",
retainedTail: [
{ role: "user", content: "root", timestamp: NOW },
{ role: "user", content: "left", timestamp: NOW },
],
});
expect(writes[1]).toMatchObject({
type: "compaction",
summary: "right summary",
retainedTail: [
{ role: "user", content: "root", timestamp: NOW },
{ role: "user", content: "right", timestamp: NOW },
],
});
expect(reader.lineReads).toBe(6);
expect(await collect(source, (id) => selected.has(id))).toEqual(writes);
});
it("ignores a torn tail and emits only the captured complete records after later appends", async () => {
const { path, read, reader } = await fixture([message("a", null)], JSON.stringify(message("torn", "a")));
const source = await read();
getOrThrow(
await fileSystem.appendFile(path, `\n${JSON.stringify(message("later", "torn"))}\n`, BACKGROUND_CONTEXT),
);
reader.lineReads = 0;
const writes = await collect(source);
expect(writes).toHaveLength(2);
expect(reader.lineReads).toBe(2); // Header plus the single captured complete record.
expect(source.nextSeq).toBe(3);
});
});

120
packages/agent/test/harness/text-line-reader.test.ts Normal file
View file

@ -0,0 +1,120 @@
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { BACKGROUND_CONTEXT, withAbortSignal } from "../../src/harness/context.ts";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import { getOrThrow, type TextLine, type TextLineReader } from "../../src/harness/types.ts";
let root: string;
let env: NodeExecutionEnv;
const readers: TextLineReader[] = [];
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), "pi-text-line-reader-"));
env = new NodeExecutionEnv({ cwd: root });
});
afterEach(async () => {
await Promise.all(readers.splice(0).map((reader) => reader.close(BACKGROUND_CONTEXT)));
await rm(root, { recursive: true, force: true });
});
async function openReader(content: string | Uint8Array): Promise<TextLineReader> {
await writeFile(join(root, "text.txt"), content);
const reader = getOrThrow(await env.openTextLineReader("text.txt", BACKGROUND_CONTEXT));
readers.push(reader);
return reader;
}
async function readLines(reader: TextLineReader): Promise<TextLine[]> {
const lines: TextLine[] = [];
while (true) {
const line = getOrThrow(await reader.readLine(BACKGROUND_CONTEXT));
if (line === undefined) return lines;
lines.push(line);
}
}
describe("TextLineReader", () => {
it("decodes Unicode, blank lines, and a torn final line", async () => {
const reader = await openReader("hé🙂\n\n\n終\ntorn");
expect(await readLines(reader)).toEqual([
{ text: "hé🙂", terminated: true },
{ text: "", terminated: true },
{ text: "", terminated: true },
{ text: "終", terminated: true },
{ text: "torn", terminated: false },
]);
expect(getOrThrow(await reader.readLine(BACKGROUND_CONTEXT))).toBeUndefined();
});
it("reads an empty file", async () => {
const reader = await openReader("");
expect(await readLines(reader)).toEqual([]);
});
it("decodes multibyte characters split across 64 KiB chunks", async () => {
const first = `${"a".repeat(64 * 1024 - 1)}🙂${"é".repeat(40_000)}\n`;
const reader = await openReader(`${first}終`);
expect(await readLines(reader)).toEqual([
{ text: first.slice(0, -1), terminated: true },
{ text: "終", terminated: false },
]);
});
it("replaces malformed and incomplete UTF-8", async () => {
const reader = await openReader(new Uint8Array([0xff, 0x0a, 0xe2, 0x82]));
expect(await readLines(reader)).toEqual([
{ text: "<22>", terminated: true },
{ text: "<22>", terminated: false },
]);
});
it("rejects an open with a pre-aborted context", async () => {
await writeFile(join(root, "text.txt"), "one\n");
const controller = new AbortController();
controller.abort();
const context = withAbortSignal(controller.signal, BACKGROUND_CONTEXT);
expect(await env.openTextLineReader("text.txt", context)).toMatchObject({
ok: false,
error: { code: "aborted" },
});
});
it("does not consume a buffered line when its context is pre-aborted", async () => {
const reader = await openReader("one\ntwo\n");
getOrThrow(await reader.readLine(BACKGROUND_CONTEXT));
const controller = new AbortController();
controller.abort();
const context = withAbortSignal(controller.signal, BACKGROUND_CONTEXT);
expect(await reader.readLine(context)).toMatchObject({ ok: false, error: { code: "aborted" } });
expect(getOrThrow(await reader.readLine(BACKGROUND_CONTEXT))?.text).toBe("two");
});
it("honors cancellation during a read and allows retry", async () => {
const reader = await openReader(`${"é".repeat(70_000)}\nlast`);
const controller = new AbortController();
const pending = reader.readLine(withAbortSignal(controller.signal, BACKGROUND_CONTEXT));
controller.abort();
expect(await pending).toMatchObject({ ok: false, error: { code: "aborted" } });
expect((await readLines(reader)).map((line) => line.text)).toEqual(["é".repeat(70_000), "last"]);
});
it("closes idempotently even with an aborted context and rejects later reads", async () => {
const reader = await openReader("one\ntwo\n");
getOrThrow(await reader.readLine(BACKGROUND_CONTEXT));
const controller = new AbortController();
controller.abort();
await expect(reader.close(withAbortSignal(controller.signal, BACKGROUND_CONTEXT))).resolves.toBeUndefined();
await expect(reader.close(BACKGROUND_CONTEXT)).resolves.toBeUndefined();
expect(await reader.readLine(BACKGROUND_CONTEXT)).toMatchObject({ ok: false, error: { code: "invalid" } });
});
it("returns a FileError for a missing file", async () => {
expect(await env.openTextLineReader("missing.txt", BACKGROUND_CONTEXT)).toMatchObject({
ok: false,
error: { code: "not_found", path: join(root, "missing.txt") },
});
});
});