diff --git a/packages/agent/src/harness/session/jsonl/storage.ts b/packages/agent/src/harness/session/jsonl/storage.ts index a11e1ca59..9dff8e516 100644 --- a/packages/agent/src/harness/session/jsonl/storage.ts +++ b/packages/agent/src/harness/session/jsonl/storage.ts @@ -1,6 +1,7 @@ -import type { Usage } from "@earendil-works/pi-ai"; +import { type Usage, uuidv7 } from "@earendil-works/pi-ai"; import type { Context } from "../../context.ts"; import type { FileError, FileSystem, Result } from "../../types.ts"; +import { insertUsage } from "../commit.ts"; import { type ForkDestinationSnapshot, type ForkSourceSnapshot, forkSnapshotWrites } from "../fork.ts"; import { type CommittedEntryWrite, @@ -103,7 +104,13 @@ async function publishFileAtomically( } } -type JsonlBacking = { kind: "v4" } | { kind: "v3"; importedUsage: Usage }; +type LegacyV3Backing = { + kind: "v3"; + importedUsage: Usage; + baselineWrites: readonly CommittedWrite[]; +}; + +type JsonlBacking = { kind: "v4" } | LegacyV3Backing; /** JSONL storage backed by an injected filesystem capability. */ export class JsonlStorage implements Storage { @@ -189,16 +196,17 @@ export class JsonlStorage implements Storage { recordLines: readonly string[], context: Context, ): Promise { - const normalized = normalizeLegacyV3Records(recordLines); + const { writes, importedUsage, nextSeq } = normalizeLegacyV3Records(recordLines); const targetHeader = { ...(await normalizeLegacyV3Header(options.fileSystem, header, context)), - nextSeq: normalized.nextSeq, + nextSeq, }; const storage = new JsonlStorage(options, targetHeader, { kind: "v3", - importedUsage: normalized.importedUsage, + importedUsage, + baselineWrites: writes, }); - storage.replayCommitted(normalized.writes); + storage.replayCommitted(writes); return storage; } @@ -219,7 +227,7 @@ export class JsonlStorage implements Storage { private async applyCommit(writes: Write[], context: Context): Promise { if (this.backing.kind === "v3" && writes.length !== 0) { - throw new Error("Legacy v3 storage is read-only"); + return this.upgradeLegacyV3ToV4(this.backing, writes, context); } const prepared = this.storageState.prepareCommit(writes, this.now()); if (prepared.writes.length !== 0) { @@ -236,6 +244,46 @@ export class JsonlStorage implements Storage { return prepared.result; } + /** Atomically upgrade legacy v3 backing and preserve the first caller write as a v4 transaction. */ + private async upgradeLegacyV3ToV4( + backing: LegacyV3Backing, + callerWrites: Write[], + context: Context, + ): Promise { + const timestamp = this.now(); + const prepared = this.storageState.prepareCommit( + [ + insertUsage({ + id: uuidv7(timestamp), + usage: backing.importedUsage, + adjustment: true, + details: { source: "v3-import" }, + }), + ...callerWrites, + ], + timestamp, + ); + + const nextSeq = prepared.result.firstSeq + prepared.writes.length; + const upgradedHeader = { ...this.header, nextSeq }; + const content = `${[ + JSON.stringify(upgradedHeader), + ...backing.baselineWrites.map((write) => JSON.stringify(write)), + JSON.stringify(prepared.writes), + ].join("\n")}\n`; + + await publishFileAtomically(this.fileSystem, this.path, content, context); + + this.storageState.applyValidated(prepared.writes); + this.backing = { kind: "v4" }; + // The first sequence belongs to the internal usage adjustment; return only caller-write sequences. + return { + ...prepared.result, + firstSeq: prepared.result.firstSeq + 1, + seqs: prepared.result.seqs.slice(1), + }; + } + getEntries(ids: string[], _context: Context): Promise> { if (this.state !== "open") return Promise.reject(new Error("JsonlStorage is closed")); return Promise.resolve(this.storageState.getEntries(ids)); diff --git a/packages/agent/test/harness/jsonl-v3-migration.test.ts b/packages/agent/test/harness/jsonl-v3-migration.test.ts index c073b0fcf..20d993ee5 100644 --- a/packages/agent/test/harness/jsonl-v3-migration.test.ts +++ b/packages/agent/test/harness/jsonl-v3-migration.test.ts @@ -3,6 +3,7 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { BACKGROUND_CONTEXT } from "../../src/harness/context.ts"; import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts"; import { + JSONL_FORMAT_VERSION, JSONL_STORAGE_VERSION, type JsonlSessionMetadata, JsonlSessionRepo, @@ -377,6 +378,109 @@ describe("JSONL v3 migration", () => { expect(getOrThrow(await fileSystem.readTextFile(path, BACKGROUND_CONTEXT))).toBe(content); }); + it("converts to v4 and preserves the first caller transaction with one usage adjustment", async () => { + // Seed a v3 session whose embedded usage must become a durable v4 ledger adjustment. + const usage = { + input: 10, + output: 5, + cacheRead: 2, + cacheWrite: 1, + totalTokens: 18, + cost: { input: 0.1, output: 0.05, cacheRead: 0.02, cacheWrite: 0.01, total: 0.18 }, + } satisfies Usage; + const message = { + role: "assistant", + content: [{ type: "text", text: "imported answer" }], + api: "anthropic-messages", + provider: "anthropic", + model: "claude-sonnet-4-5", + usage, + stopReason: "stop", + timestamp: NOW + 1_000, + } satisfies AgentMessage; + const { path } = await writeLegacyV3Fixture([ + { + type: "message", + id: "assistant", + parentId: null, + timestamp: new Date(NOW + 1_000).toISOString(), + message, + }, + ]); + // Capture the normalized read-only view so conversion can be checked for identity and stats stability. + const options = { fileSystem, path, now: () => NOW }; + const storage = await JsonlStorage.open(options, BACKGROUND_CONTEXT); + const importedEntries = await storage.scanEntries({ order: "asc" }, BACKGROUND_CONTEXT); + const statsBefore = await storage.getStats(BACKGROUND_CONTEXT); + + // The first caller write triggers conversion and must not expose the internal adjustment sequence. + const committed = await storage.commit( + [storedValues.setValue(storedValues.sessionName, "Converted session")], + BACKGROUND_CONTEXT, + ); + + expect(committed.seqs).toHaveLength(1); + expect(committed.firstSeq).toBe(committed.seqs[0]); + expect((await storage.getValue(storedValues.sessionName, BACKGROUND_CONTEXT))?.seq).toBe(committed.seqs[0]); + const usageRows = await storage.scanUsage({ order: "asc" }, BACKGROUND_CONTEXT); + expect(usageRows).toHaveLength(1); + const [adjustment] = usageRows; + if (adjustment === undefined) throw new Error("Import usage adjustment was not committed"); + expect(adjustment).toMatchObject({ + usage, + adjustment: true, + details: { source: "v3-import" }, + }); + expect(adjustment).not.toHaveProperty("entryId"); + expect(committed.seqs).not.toContain(adjustment.seq); + expect(await storage.getStats(BACKGROUND_CONTEXT)).toEqual(statsBefore); + + // Conversion replaces the legacy file and retains the adjustment and caller writes as its first transaction. + const convertedLines = getOrThrow(await fileSystem.readTextFile(path, BACKGROUND_CONTEXT)) + .trimEnd() + .split("\n"); + const headerLine = convertedLines[0]; + const transactionLine = convertedLines.at(-1); + if (headerLine === undefined || transactionLine === undefined) throw new Error("Converted file was not written"); + expect(JSON.parse(headerLine)).toMatchObject({ + v: JSONL_FORMAT_VERSION, + kind: "header", + id: "legacy", + storageVersion: JSONL_STORAGE_VERSION, + createdAt: NOW, + cwd: "/workspace", + }); + expect(JSON.parse(transactionLine)).toEqual([ + { kind: "usage", ...adjustment }, + { + kind: "value", + op: "set", + seq: committed.seqs[0], + namespace: storedValues.sessionName.namespace, + key: storedValues.sessionName.key, + value: "Converted session", + }, + ]); + await storage.close(BACKGROUND_CONTEXT); + + // Reopening through the ordinary v4 path must recover the complete normalized and newly committed state. + const reopened = await JsonlStorage.open(options, BACKGROUND_CONTEXT); + expect(await reopened.scanEntries({ order: "asc" }, BACKGROUND_CONTEXT)).toEqual(importedEntries); + const [importedEntry] = importedEntries; + if (importedEntry === undefined) throw new Error("Legacy message was not imported"); + expect((await reopened.getValue(storedValues.laneLeaf("main"), BACKGROUND_CONTEXT))?.value).toBe( + importedEntry.id, + ); + expect((await reopened.getValue(storedValues.laneState("main"), BACKGROUND_CONTEXT))?.value).toEqual({ + currentOperationId: null, + pendingNextRun: [], + }); + expect((await reopened.getValue(storedValues.sessionName, BACKGROUND_CONTEXT))?.value).toBe("Converted session"); + expect(await reopened.scanUsage({ order: "asc" }, BACKGROUND_CONTEXT)).toEqual([adjustment]); + expect(await reopened.getStats(BACKGROUND_CONTEXT)).toEqual(statsBefore); + await reopened.close(BACKGROUND_CONTEXT); + }); + describe("discarding legacy configuration changes", () => { const firstTimestamp = NOW + 1_000; const modelChangeTimestamp = NOW + 2_000;