mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-06 23:44:36 +00:00
feat(agent): upgrade legacy v3 sessions on first write
This commit is contained in:
parent
919ff8733a
commit
2f642c5d73
2 changed files with 159 additions and 7 deletions
|
|
@ -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<JsonlStorage> {
|
||||
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<CommitResult> {
|
||||
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<CommitResult> {
|
||||
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<Map<string, Entry>> {
|
||||
if (this.state !== "open") return Promise.reject(new Error("JsonlStorage is closed"));
|
||||
return Promise.resolve(this.storageState.getEntries(ids));
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue