From f332cd9717dd8443c36b189bac41efacc7586e92 Mon Sep 17 00:00:00 2001 From: ikma Date: Thu, 6 Aug 2026 22:49:32 -0400 Subject: [PATCH 1/3] fix: recover committed conversation leaf --- README.md | 18 +- src/conversation/conversation-registry.ts | 16 +- src/conversation/conversation.ts | 65 +++++- src/conversation/file-history-store.ts | 52 +++-- src/server/server-runner.ts | 6 +- src/session/pi-session.ts | 68 +++++- .../conversation-recovery.test.ts | 218 ++++++++++++++++++ .../conversation-registry.test.ts | 42 ++-- test/conversation/file-history-store.test.ts | 100 ++++++-- test/openai/chat-service.test.ts | 21 +- 10 files changed, 526 insertions(+), 80 deletions(-) create mode 100644 test/conversation/conversation-recovery.test.ts diff --git a/README.md b/README.md index 2d9e9c5..f6aefe1 100644 --- a/README.md +++ b/README.md @@ -44,13 +44,17 @@ themes, or ambient context. `ConversationRegistry` resolves user and chat identity to one `Conversation`. The conversation checks the caller's visible history, reserves one turn, streams -through its `SessionAgent`, and commits by saving the next visible history. -Model failure, active abort or cancellation, and history-save failure restore the -prior Pi branch before the registry permits a new session for that identity. - -`FileConversationHistoryStore` atomically replaces mode-0600 history files in a -mode-0700 directory. It stores only visible user and assistant role/content -pairs. +through its `SessionAgent`, and commits by saving the next visible history with +the post-response Pi leaf. Model failure, active abort or cancellation, and +history-save failure restore the prior Pi branch before the registry permits a +new session for that identity. + +`FileConversationHistoryStore` atomically replaces mode-0600 versioned snapshots +in a mode-0700 directory. Each snapshot stores visible user and assistant +role/content pairs plus the committed Pi leaf. After a process restart, Stein +restores that leaf before Pi builds model context, excluding entries abandoned +before snapshot replacement. This is process-crash recovery, not a power-loss +durability or multi-process safety guarantee. ## Serve OpenAI-compatible chat diff --git a/src/conversation/conversation-registry.ts b/src/conversation/conversation-registry.ts index 30a3c92..53fb971 100644 --- a/src/conversation/conversation-registry.ts +++ b/src/conversation/conversation-registry.ts @@ -2,9 +2,11 @@ import { createHash } from "node:crypto"; import { SessionAgent } from "../session/session-agent.ts"; import { Conversation, + type ConversationHistorySnapshot, type ConversationHistoryStore, type ConversationMessage, type ConversationTurn, + validateConversationHistorySnapshot, } from "./conversation.ts"; export type ConversationIdentity = Readonly<{ @@ -15,6 +17,7 @@ export type ConversationIdentity = Readonly<{ export type ConversationAgentFactory = ( identity: ConversationIdentity, + snapshot: ConversationHistorySnapshot, ) => Promise; export class ConversationRegistry { @@ -42,19 +45,20 @@ export class ConversationRegistry { if (existing) return existing; let created: Promise; - created = this.#historyStore.load(identity.conversationId).then((history) => - new Conversation({ + created = this.#historyStore.load(identity.conversationId).then((snapshot) => { + validateConversationHistorySnapshot(snapshot); + return new Conversation({ conversationId: identity.conversationId, - history, + history: snapshot.messages, historyStore: this.#historyStore, - createAgent: () => this.#createAgent(identity), + createAgent: () => this.#createAgent(identity, snapshot), evict: () => { if (this.#conversations.get(identity.conversationId) === created) { this.#conversations.delete(identity.conversationId); } }, - }) - ); + }); + }); this.#conversations.set(identity.conversationId, created); created.catch(() => { if (this.#conversations.get(identity.conversationId) === created) { diff --git a/src/conversation/conversation.ts b/src/conversation/conversation.ts index d24c8ca..06a263b 100644 --- a/src/conversation/conversation.ts +++ b/src/conversation/conversation.ts @@ -12,16 +12,51 @@ export type VisibleConversationMessage = Readonly<{ content: string; }>; +export type ConversationHistorySnapshot = Readonly<{ + version: 1; + messages: readonly VisibleConversationMessage[]; + committedLeafId: string | null; +}>; + export interface ConversationHistoryStore { - load(conversationId: string): Promise; + load(conversationId: string): Promise; // Success is the turn commit point; abort before replacement must reject. save( conversationId: string, - history: readonly VisibleConversationMessage[], + snapshot: ConversationHistorySnapshot, signal: AbortSignal, ): Promise; } +export function emptyConversationHistorySnapshot(): ConversationHistorySnapshot { + return { version: 1, messages: [], committedLeafId: null }; +} + +export function validateConversationHistorySnapshot( + value: ConversationHistorySnapshot, +): void { + if (value.version !== 1 || !Array.isArray(value.messages)) { + throw new Error("Invalid visible conversation history snapshot"); + } + if (!value.messages.every(isVisibleConversationMessage)) { + throw new Error("Invalid visible conversation history messages"); + } + if (value.committedLeafId !== null && !value.committedLeafId.trim()) { + throw new Error("Invalid committed Pi session leaf"); + } + if ((value.messages.length === 0) !== (value.committedLeafId === null)) { + throw new Error("Visible conversation history and committed Pi leaf disagree"); + } + if ( + value.messages.length % 2 !== 0 || + value.messages.some((message, index) => + message.role !== (index % 2 === 0 ? "user" : "assistant") + ) + ) { + throw new Error("Visible conversation history is not a sequence of committed turns"); + } +} + export type ConversationTurn = Readonly<{ conversationId: string; deltas: AsyncIterable; @@ -95,15 +130,22 @@ export class Conversation { async commit( userMessage: ConversationMessage, assistantContent: string, + committedLeafId: string | null, signal: AbortSignal, ): Promise { - const history: VisibleConversationMessage[] = [ + const messages: VisibleConversationMessage[] = [ ...this.#history, { role: "user", content: userMessage.content }, { role: "assistant", content: assistantContent }, ]; - await this.#historyStore.save(this.conversationId, history, signal); - this.#history = history; + const snapshot: ConversationHistorySnapshot = { + version: 1, + messages, + committedLeafId, + }; + validateConversationHistorySnapshot(snapshot); + await this.#historyStore.save(this.conversationId, snapshot, signal); + this.#history = messages; this.#active = false; } @@ -177,9 +219,11 @@ class ActiveConversationTurn implements ConversationTurn { if (this.#phase !== "responding") return; this.#phase = "saving"; + const committedLeafId = this.#agent.checkpoint(); this.#savePromise = this.#conversation.commit( this.#userMessage, output, + committedLeafId, this.#abortController.signal, ); await this.#savePromise; @@ -209,6 +253,17 @@ class ActiveConversationTurn implements ConversationTurn { } } +function isVisibleConversationMessage( + value: unknown, +): value is VisibleConversationMessage { + if (typeof value !== "object" || value === null) return false; + const record = value as Record; + return ( + (record.role === "user" || record.role === "assistant") && + typeof record.content === "string" + ); +} + function snapshotUserMessage(message: ConversationMessage): ConversationMessage { return { role: "user", diff --git a/src/conversation/file-history-store.ts b/src/conversation/file-history-store.ts index c4c97b7..552e01c 100644 --- a/src/conversation/file-history-store.ts +++ b/src/conversation/file-history-store.ts @@ -2,9 +2,11 @@ import { randomUUID } from "node:crypto"; import { renameSync } from "node:fs"; import { chmod, mkdir, readFile, rm, writeFile } from "node:fs/promises"; import { isAbsolute, join } from "node:path"; -import type { - ConversationHistoryStore, - VisibleConversationMessage, +import { + emptyConversationHistorySnapshot, + type ConversationHistorySnapshot, + type ConversationHistoryStore, + validateConversationHistorySnapshot, } from "./conversation.ts"; export class FileConversationHistoryStore implements ConversationHistoryStore { @@ -17,31 +19,44 @@ export class FileConversationHistoryStore implements ConversationHistoryStore { this.#directory = directory; } - async load(conversationId: string): Promise { + async load(conversationId: string): Promise { let contents: string; try { contents = await readFile(this.#path(conversationId), "utf8"); } catch (error) { - if (isMissing(error)) return []; + if (isMissing(error)) return emptyConversationHistorySnapshot(); throw error; } - const value: unknown = JSON.parse(contents); - if (!Array.isArray(value) || !value.every(isVisibleMessage)) { - throw new Error(`Invalid visible conversation history for ${conversationId}`); + let value: unknown; + try { + value = JSON.parse(contents); + } catch (error) { + throw new Error("Invalid visible conversation history JSON", { cause: error }); + } + if (Array.isArray(value)) { + throw new Error( + "Legacy array-only visible history requires an explicit migration or reset", + ); } - return value.map(({ role, content }) => ({ role, content })); + if (!isConversationHistorySnapshot(value)) { + throw new Error(`Invalid visible conversation history snapshot for ${conversationId}`); + } + validateConversationHistorySnapshot(value); + return { + version: 1, + messages: value.messages.map(({ role, content }) => ({ role, content })), + committedLeafId: value.committedLeafId, + }; } async save( conversationId: string, - history: readonly VisibleConversationMessage[], + snapshot: ConversationHistorySnapshot, signal: AbortSignal, ): Promise { const destination = this.#path(conversationId); - if (!history.every(isVisibleMessage)) { - throw new Error(`Invalid visible conversation history for ${conversationId}`); - } + validateConversationHistorySnapshot(snapshot); signal.throwIfAborted(); await mkdir(this.#directory, { recursive: true, mode: 0o700 }); @@ -50,7 +65,7 @@ export class FileConversationHistoryStore implements ConversationHistoryStore { signal.throwIfAborted(); const temporary = `${destination}.${process.pid}.${randomUUID()}.tmp`; try { - await writeFile(temporary, `${JSON.stringify(history, null, 2)}\n`, { + await writeFile(temporary, `${JSON.stringify(snapshot, null, 2)}\n`, { encoding: "utf8", flag: "wx", mode: 0o600, @@ -71,12 +86,15 @@ export class FileConversationHistoryStore implements ConversationHistoryStore { } } -function isVisibleMessage(value: unknown): value is VisibleConversationMessage { +function isConversationHistorySnapshot( + value: unknown, +): value is ConversationHistorySnapshot { if (typeof value !== "object" || value === null) return false; const record = value as Record; return ( - (record.role === "user" || record.role === "assistant") && - typeof record.content === "string" + record.version === 1 && + Array.isArray(record.messages) && + (record.committedLeafId === null || typeof record.committedLeafId === "string") ); } diff --git a/src/server/server-runner.ts b/src/server/server-runner.ts index 5ac67a6..1646653 100644 --- a/src/server/server-runner.ts +++ b/src/server/server-runner.ts @@ -26,7 +26,11 @@ export async function runServer(config: ServerConfig): Promise { bearerToken: bearerToken.trim(), modelId, historyStore: new FileConversationHistoryStore(config.sessionDirectory), - createAgent: ({ conversationId }) => createSession(conversationId), + createAgent: ({ conversationId }, snapshot) => + createSession(conversationId, { + committedLeafId: snapshot.committedLeafId, + committedMessageRoles: snapshot.messages.map(({ role }) => role), + }), }); const server = Bun.serve({ hostname: config.hostname, diff --git a/src/session/pi-session.ts b/src/session/pi-session.ts index cf60a64..28502f4 100644 --- a/src/session/pi-session.ts +++ b/src/session/pi-session.ts @@ -1,3 +1,4 @@ +import { existsSync } from "node:fs"; import { isAbsolute, join } from "node:path"; import { createAgentSession, @@ -10,6 +11,11 @@ import { SessionAgent } from "./session-agent.ts"; export type ModelDescription = Readonly<{ provider: string; id: string }>; +export type PiSessionRecovery = Readonly<{ + committedLeafId: string | null; + committedMessageRoles: readonly ("user" | "assistant")[]; +}>; + export type PiSessionFactoryConfig = Readonly<{ model: ModelDescription; systemPrompt: string; @@ -20,7 +26,10 @@ export type PiSessionFactoryConfig = Readonly<{ export function createPiSessionFactory(config: PiSessionFactoryConfig) { validateConfig(config); - return async (conversationId: string): Promise => { + return async ( + conversationId: string, + recovery?: PiSessionRecovery, + ): Promise => { validateConversationId(conversationId); const modelRuntime = await ModelRuntime.create({ authPath: join(config.agentDirectory, "auth.json"), @@ -48,6 +57,12 @@ export function createPiSessionFactory(config: PiSessionFactoryConfig) { }); await resourceLoader.reload(); const sessionFile = join(config.sessionDirectory, `${conversationId}.jsonl`); + const sessionManager = openSessionManagerForRecovery( + sessionFile, + config.sessionDirectory, + config.workspaceDirectory, + recovery, + ); const { session, extensionsResult, modelFallbackMessage } = await createAgentSession({ cwd: config.workspaceDirectory, agentDir: config.agentDirectory, @@ -55,11 +70,7 @@ export function createPiSessionFactory(config: PiSessionFactoryConfig) { model, noTools: "all", resourceLoader, - sessionManager: SessionManager.open( - sessionFile, - config.sessionDirectory, - config.workspaceDirectory, - ), + sessionManager, settingsManager, }); if (extensionsResult.errors.length > 0 || modelFallbackMessage) { @@ -71,6 +82,51 @@ export function createPiSessionFactory(config: PiSessionFactoryConfig) { }; } +export function openSessionManagerForRecovery( + sessionFile: string, + sessionDirectory: string, + workspaceDirectory: string, + recovery?: PiSessionRecovery, +): SessionManager { + const sessionExists = existsSync(sessionFile); + if (recovery !== undefined && recovery.committedLeafId !== null && !sessionExists) { + throw new Error("Committed visible history has no Pi session file"); + } + + const sessionManager = SessionManager.open( + sessionFile, + sessionDirectory, + workspaceDirectory, + ); + if (recovery === undefined) return sessionManager; + + if (recovery.committedLeafId === null) { + if (recovery.committedMessageRoles.length > 0) { + throw new Error("Visible conversation history does not match committed Pi root"); + } + sessionManager.resetLeaf(); + return sessionManager; + } + try { + sessionManager.branch(recovery.committedLeafId); + } catch (error) { + throw new Error( + `Committed Pi session leaf is unavailable: ${recovery.committedLeafId}`, + { cause: error }, + ); + } + const branchMessageRoles = sessionManager.getBranch() + .filter((entry) => entry.type === "message") + .map((entry) => entry.message.role); + if ( + branchMessageRoles.length !== recovery.committedMessageRoles.length || + branchMessageRoles.some((role, index) => role !== recovery.committedMessageRoles[index]) + ) { + throw new Error("Visible conversation history does not match committed Pi branch"); + } + return sessionManager; +} + function validateConfig(config: PiSessionFactoryConfig): void { requireText("model.provider", config.model.provider); requireText("model.id", config.model.id); diff --git a/test/conversation/conversation-recovery.test.ts b/test/conversation/conversation-recovery.test.ts new file mode 100644 index 0000000..f14190c --- /dev/null +++ b/test/conversation/conversation-recovery.test.ts @@ -0,0 +1,218 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { mkdir, mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { SessionManager } from "@earendil-works/pi-coding-agent"; +import type { ConversationHistorySnapshot } from "../../src/conversation/conversation.ts"; +import { FileConversationHistoryStore } from "../../src/conversation/file-history-store.ts"; +import { openSessionManagerForRecovery } from "../../src/session/pi-session.ts"; + +const roots: string[] = []; +const activeSignal = (): AbortSignal => new AbortController().signal; +type SessionMessage = Parameters[0]; + +type Harness = Readonly<{ + sessionDirectory: string; + workspaceDirectory: string; + sessionFile: string; + historyStore: FileConversationHistoryStore; + conversationId: string; +}>; + +async function harness(): Promise { + const root = await mkdtemp(join(tmpdir(), "ghl-conversation-recovery-")); + roots.push(root); + const sessionDirectory = join(root, "sessions"); + const workspaceDirectory = join(root, "workspace"); + await Promise.all([ + mkdir(sessionDirectory, { recursive: true }), + mkdir(workspaceDirectory, { recursive: true }), + ]); + const conversationId = "a".repeat(64); + return { + sessionDirectory, + workspaceDirectory, + sessionFile: join(sessionDirectory, `${conversationId}.jsonl`), + historyStore: new FileConversationHistoryStore(sessionDirectory), + conversationId, + }; +} + +function appendTurn(manager: SessionManager, label: string): string { + manager.appendMessage({ role: "user", content: `${label} user`, timestamp: Date.now() }); + return manager.appendMessage({ + role: "assistant", + content: [{ type: "text", text: `${label} assistant` }], + api: "openai-completions", + provider: "test", + model: "deterministic", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + } as SessionMessage); +} + +function snapshot(label: string, committedLeafId: string): ConversationHistorySnapshot { + return { + version: 1, + messages: [ + { role: "user", content: `${label} user` }, + { role: "assistant", content: `${label} assistant` }, + ], + committedLeafId, + }; +} + +function reopen( + state: Harness, + snapshot: ConversationHistorySnapshot, +): SessionManager { + return openSessionManagerForRecovery( + state.sessionFile, + state.sessionDirectory, + state.workspaceDirectory, + { + committedLeafId: snapshot.committedLeafId, + committedMessageRoles: snapshot.messages.map(({ role }) => role), + }, + ); +} + +function context(manager: SessionManager): string { + return JSON.stringify(manager.buildSessionContext().messages); +} + +afterEach(async () => { + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); +}); + +describe("committed conversation recovery", () => { + test("crash before snapshot replacement restores the previous committed leaf", async () => { + const state = await harness(); + const manager = SessionManager.open( + state.sessionFile, + state.sessionDirectory, + state.workspaceDirectory, + ); + const committedLeafId = appendTurn(manager, "committed"); + await state.historyStore.save( + state.conversationId, + snapshot("committed", committedLeafId), + activeSignal(), + ); + const abandonedLeafId = appendTurn(manager, "abandoned"); + + const stored = await state.historyStore.load(state.conversationId); + const recovered = reopen(state, stored); + + expect(recovered.getLeafId()).toBe(committedLeafId); + expect(recovered.getLeafId()).not.toBe(abandonedLeafId); + expect(context(recovered)).toContain("committed assistant"); + expect(context(recovered)).not.toContain("abandoned assistant"); + }); + + test("crash after snapshot replacement restores the new committed leaf", async () => { + const state = await harness(); + const manager = SessionManager.open( + state.sessionFile, + state.sessionDirectory, + state.workspaceDirectory, + ); + appendTurn(manager, "first"); + const committedLeafId = appendTurn(manager, "second"); + const committed: ConversationHistorySnapshot = { + version: 1, + messages: [ + { role: "user", content: "first user" }, + { role: "assistant", content: "first assistant" }, + { role: "user", content: "second user" }, + { role: "assistant", content: "second assistant" }, + ], + committedLeafId, + }; + await state.historyStore.save(state.conversationId, committed, activeSignal()); + + const stored = await state.historyStore.load(state.conversationId); + const recovered = reopen(state, stored); + + expect(recovered.getLeafId()).toBe(committedLeafId); + expect(context(recovered)).toContain("first assistant"); + expect(context(recovered)).toContain("second assistant"); + }); + + test("missing snapshot resets an abandoned first turn to committed root", async () => { + const state = await harness(); + const manager = SessionManager.open( + state.sessionFile, + state.sessionDirectory, + state.workspaceDirectory, + ); + appendTurn(manager, "abandoned first"); + + const stored = await state.historyStore.load(state.conversationId); + const recovered = reopen(state, stored); + + expect(stored).toEqual({ version: 1, messages: [], committedLeafId: null }); + expect(recovered.getLeafId()).toBeNull(); + expect(recovered.buildSessionContext().messages).toEqual([]); + }); + + test("leaves the standalone session path unchanged without recovery metadata", async () => { + const state = await harness(); + const manager = SessionManager.open( + state.sessionFile, + state.sessionDirectory, + state.workspaceDirectory, + ); + const latestLeafId = appendTurn(manager, "standalone"); + + const reopened = openSessionManagerForRecovery( + state.sessionFile, + state.sessionDirectory, + state.workspaceDirectory, + ); + + expect(reopened.getLeafId()).toBe(latestLeafId); + expect(context(reopened)).toContain("standalone assistant"); + }); + + test("fails closed when snapshot messages do not match the committed branch", async () => { + const state = await harness(); + const manager = SessionManager.open( + state.sessionFile, + state.sessionDirectory, + state.workspaceDirectory, + ); + appendTurn(manager, "first"); + const secondLeafId = appendTurn(manager, "second"); + + expect(() => reopen(state, snapshot("first", secondLeafId))).toThrow( + "Visible conversation history does not match committed Pi branch", + ); + }); + + test("fails closed for a missing session file and an unknown committed leaf", async () => { + const missing = await harness(); + expect(() => reopen(missing, snapshot("missing", "missing-leaf"))).toThrow( + "Committed visible history has no Pi session file", + ); + + const unknown = await harness(); + const manager = SessionManager.open( + unknown.sessionFile, + unknown.sessionDirectory, + unknown.workspaceDirectory, + ); + appendTurn(manager, "existing"); + expect(() => reopen(unknown, snapshot("unknown", "unknown-leaf"))).toThrow( + "Committed Pi session leaf is unavailable", + ); + }); +}); diff --git a/test/conversation/conversation-registry.test.ts b/test/conversation/conversation-registry.test.ts index a0ea684..1ed84c7 100644 --- a/test/conversation/conversation-registry.test.ts +++ b/test/conversation/conversation-registry.test.ts @@ -1,9 +1,10 @@ import { describe, expect, test } from "bun:test"; import { ConversationConflictError, + emptyConversationHistorySnapshot, + type ConversationHistorySnapshot, type ConversationHistoryStore, type ConversationMessage, - type VisibleConversationMessage, } from "../../src/conversation/conversation.ts"; import { ConversationRegistry, @@ -13,19 +14,21 @@ import { SessionAgent } from "../../src/session/session-agent.ts"; import { FakePiSession } from "../support/fake-pi-session.ts"; class MemoryHistoryStore implements ConversationHistoryStore { - readonly histories = new Map(); + readonly snapshots = new Map(); failure: unknown; shouldFail = false; saveGate: Promise | undefined; saveStarted: (() => void) | undefined; - async load(conversationId: string): Promise { - return structuredClone(this.histories.get(conversationId) ?? []); + async load(conversationId: string): Promise { + return structuredClone( + this.snapshots.get(conversationId) ?? emptyConversationHistorySnapshot(), + ); } async save( conversationId: string, - history: readonly VisibleConversationMessage[], + snapshot: ConversationHistorySnapshot, signal: AbortSignal, ): Promise { signal.throwIfAborted(); @@ -33,7 +36,7 @@ class MemoryHistoryStore implements ConversationHistoryStore { if (this.saveGate) await this.saveGate; signal.throwIfAborted(); if (this.shouldFail) throw this.failure; - this.histories.set(conversationId, structuredClone(history)); + this.snapshots.set(conversationId, structuredClone(snapshot)); } } @@ -42,16 +45,18 @@ function harness( historyStore = new MemoryHistoryStore(), ) { const identities: ConversationIdentity[] = []; + const committedLeaves: Array = []; const sessions: FakePiSession[] = []; - const registry = new ConversationRegistry(async (identity) => { + const registry = new ConversationRegistry(async (identity, snapshot) => { identities.push(identity); + committedLeaves.push(snapshot.committedLeafId); const session = new FakePiSession(); session.defaultResponse = ["hello", " world"]; configure?.(session, sessions.length); sessions.push(session); return new SessionAgent({ conversationId: identity.conversationId, session }); }, historyStore); - return { registry, identities, sessions, historyStore }; + return { registry, identities, committedLeaves, sessions, historyStore }; } const user = ( @@ -75,10 +80,14 @@ describe("ConversationRegistry", () => { const { registry, identities, sessions, historyStore } = harness(); const first = await registry.start("user", "chat", [user("first")]); expect(await consume(first.deltas)).toBe("hello world"); - expect(historyStore.histories.get(first.conversationId)).toEqual([ - { role: "user", content: "first" }, - { role: "assistant", content: "hello world" }, - ]); + expect(historyStore.snapshots.get(first.conversationId)).toEqual({ + version: 1, + messages: [ + { role: "user", content: "first" }, + { role: "assistant", content: "hello world" }, + ], + committedLeafId: "session-1", + }); const second = await registry.start("user", "chat", [ user("first"), @@ -108,6 +117,7 @@ describe("ConversationRegistry", () => { ]); expect(await consume(continued.deltas)).toBe("hello world"); + expect(restarted.committedLeaves).toEqual(["session-1"]); expect(restarted.sessions).toHaveLength(1); }); @@ -159,7 +169,7 @@ describe("ConversationRegistry", () => { data: { checkpoint: null }, }); expect(sessions[0]?.disposeCount).toBe(1); - expect(historyStore.histories.size).toBe(0); + expect(historyStore.snapshots.size).toBe(0); const retry = await registry.start("user", "chat", [user("retry")]); expect(await consume(retry.deltas)).toBe("hello world"); @@ -201,7 +211,7 @@ describe("ConversationRegistry", () => { await expect(consume(failed.deltas)).rejects.toThrow("deterministic history failure"); expect(sessions[0]?.sessionManager.branches).toEqual([null]); expect(sessions[0]?.disposeCount).toBe(1); - expect(historyStore.histories.size).toBe(0); + expect(historyStore.snapshots.size).toBe(0); }); test("waits for rollback when aborted during visible-history save", async () => { @@ -233,7 +243,7 @@ describe("ConversationRegistry", () => { const aborting = active.abort().then(() => { abortFinished = true; }); await Promise.resolve(); expect(abortFinished).toBe(false); - expect(historyStore.histories.size).toBe(0); + expect(historyStore.snapshots.size).toBe(0); releaseSave(); await rollbackStarted; @@ -244,7 +254,7 @@ describe("ConversationRegistry", () => { expect(await failed).toHaveProperty("name", "AbortError"); expect(sessions[0]?.sessionManager.branches).toEqual([null]); expect(sessions[0]?.disposeCount).toBe(1); - expect(historyStore.histories.size).toBe(0); + expect(historyStore.snapshots.size).toBe(0); historyStore.saveGate = undefined; historyStore.saveStarted = undefined; diff --git a/test/conversation/file-history-store.test.ts b/test/conversation/file-history-store.test.ts index c260571..c0b6869 100644 --- a/test/conversation/file-history-store.test.ts +++ b/test/conversation/file-history-store.test.ts @@ -2,6 +2,10 @@ import { afterEach, describe, expect, test } from "bun:test"; import { mkdir, mkdtemp, readFile, readdir, rm, stat, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import type { + ConversationHistorySnapshot, + VisibleConversationMessage, +} from "../../src/conversation/conversation.ts"; import { FileConversationHistoryStore } from "../../src/conversation/file-history-store.ts"; const roots: string[] = []; @@ -13,19 +17,40 @@ async function temporaryDirectory(): Promise { return root; } +function snapshot( + committedLeafId: string, + messages: readonly VisibleConversationMessage[], +): ConversationHistorySnapshot { + return { version: 1, messages, committedLeafId }; +} + +const firstMessages = [ + { role: "user" as const, content: "fictional question" }, + { role: "assistant" as const, content: "fictional response" }, +]; +const replacementMessages = [ + ...firstMessages, + { role: "user" as const, content: "fictional follow-up" }, + { role: "assistant" as const, content: "fictional continuation" }, +]; + afterEach(async () => { await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); }); describe("FileConversationHistoryStore", () => { - test("atomically replaces private visible history", async () => { + test("atomically replaces a versioned private history snapshot", async () => { const directory = join(await temporaryDirectory(), "history"); const store = new FileConversationHistoryStore(directory); const conversationId = "a".repeat(64); - const first = [{ role: "user" as const, content: "fictional question" }]; - const replacement = [{ role: "assistant" as const, content: "fictional response" }]; + const first = snapshot("first-leaf", firstMessages); + const replacement = snapshot("replacement-leaf", replacementMessages); - expect(await store.load(conversationId)).toEqual([]); + expect(await store.load(conversationId)).toEqual({ + version: 1, + messages: [], + committedLeafId: null, + }); await store.save(conversationId, first, activeSignal()); await store.save(conversationId, replacement, activeSignal()); @@ -37,36 +62,85 @@ describe("FileConversationHistoryStore", () => { expect(await readdir(directory)).toEqual([`${conversationId}.visible-history.json`]); }); - test("does not replace visible history after cancellation", async () => { + test("does not replace a committed snapshot after cancellation", async () => { const directory = join(await temporaryDirectory(), "history"); const store = new FileConversationHistoryStore(directory); const conversationId = "b".repeat(64); - const committed = [{ role: "user" as const, content: "fictional question" }]; + const committed = snapshot("committed-leaf", firstMessages); await store.save(conversationId, committed, activeSignal()); const controller = new AbortController(); controller.abort(); await expect(store.save( conversationId, - [{ role: "assistant", content: "cancelled response" }], + snapshot("cancelled-leaf", replacementMessages), controller.signal, )).rejects.toHaveProperty("name", "AbortError"); expect(await store.load(conversationId)).toEqual(committed); expect(await readdir(directory)).toEqual([`${conversationId}.visible-history.json`]); }); - test("rejects unsafe identities and malformed persisted history", async () => { + test("rejects unsafe identities and malformed snapshot metadata", async () => { const directory = join(await temporaryDirectory(), "history"); const store = new FileConversationHistoryStore(directory); await expect(store.load("../other-conversation")).rejects.toThrow("SHA-256"); await mkdir(directory, { recursive: true }); const conversationId = "b".repeat(64); - await writeFile( - join(directory, `${conversationId}.visible-history.json`), - JSON.stringify([{ role: "system", content: "not visible history" }]), + const path = join(directory, `${conversationId}.visible-history.json`); + await writeFile(path, JSON.stringify({ + version: 1, + messages: [{ role: "system", content: "invalid" }], + committedLeafId: "leaf", + })); + await expect(store.load(conversationId)).rejects.toThrow( + "Invalid visible conversation history messages", + ); + + await writeFile(path, JSON.stringify({ + version: 1, + messages: [], + committedLeafId: "impossible-leaf", + })); + await expect(store.load(conversationId)).rejects.toThrow( + "history and committed Pi leaf disagree", + ); + + await writeFile(path, JSON.stringify({ + version: 1, + messages: firstMessages, + committedLeafId: null, + })); + await expect(store.load(conversationId)).rejects.toThrow( + "history and committed Pi leaf disagree", + ); + + await writeFile(path, JSON.stringify({ + version: 1, + messages: [{ role: "assistant", content: "out of order" }], + committedLeafId: "leaf", + })); + await expect(store.load(conversationId)).rejects.toThrow( + "not a sequence of committed turns", + ); + }); + + test("rejects corrupt and legacy array-only history without guessing", async () => { + const directory = join(await temporaryDirectory(), "history"); + const store = new FileConversationHistoryStore(directory); + const conversationId = "d".repeat(64); + const path = join(directory, `${conversationId}.visible-history.json`); + await mkdir(directory, { recursive: true }); + + await writeFile(path, "not-json"); + await expect(store.load(conversationId)).rejects.toThrow( + "Invalid visible conversation history JSON", + ); + + await writeFile(path, JSON.stringify(firstMessages)); + await expect(store.load(conversationId)).rejects.toThrow( + "Legacy array-only visible history requires an explicit migration or reset", ); - await expect(store.load(conversationId)).rejects.toThrow("Invalid visible conversation history"); }); test("cleans temporary output when atomic replacement fails", async () => { @@ -77,7 +151,7 @@ describe("FileConversationHistoryStore", () => { await expect(store.save( conversationId, - [{ role: "user", content: "question" }], + snapshot("leaf", firstMessages), activeSignal(), )).rejects.toBeDefined(); expect(await readdir(directory)).toEqual([`${conversationId}.visible-history.json`]); diff --git a/test/openai/chat-service.test.ts b/test/openai/chat-service.test.ts index 6fc0b43..3862de0 100644 --- a/test/openai/chat-service.test.ts +++ b/test/openai/chat-service.test.ts @@ -1,8 +1,9 @@ import { describe, expect, test } from "bun:test"; -import type { - ConversationHistoryStore, - ConversationMessage, - VisibleConversationMessage, +import { + emptyConversationHistorySnapshot, + type ConversationHistorySnapshot, + type ConversationHistoryStore, + type ConversationMessage, } from "../../src/conversation/conversation.ts"; import type { ConversationIdentity } from "../../src/conversation/conversation-registry.ts"; import { OpenAIChatService } from "../../src/openai/chat-service.ts"; @@ -10,17 +11,19 @@ import { SessionAgent } from "../../src/session/session-agent.ts"; import { FakePiSession } from "../support/fake-pi-session.ts"; class MemoryHistoryStore implements ConversationHistoryStore { - readonly histories = new Map(); + readonly snapshots = new Map(); - async load(conversationId: string): Promise { - return structuredClone(this.histories.get(conversationId) ?? []); + async load(conversationId: string): Promise { + return structuredClone( + this.snapshots.get(conversationId) ?? emptyConversationHistorySnapshot(), + ); } async save( conversationId: string, - history: readonly VisibleConversationMessage[], + snapshot: ConversationHistorySnapshot, ): Promise { - this.histories.set(conversationId, structuredClone(history)); + this.snapshots.set(conversationId, structuredClone(snapshot)); } } From 5795e55ceaeda13910c390832fd364de589b3938 Mon Sep 17 00:00:00 2001 From: ikma Date: Thu, 6 Aug 2026 23:25:51 -0400 Subject: [PATCH 2/3] fix: keep recovery keyed to committed leaf --- src/server/server-runner.ts | 5 +---- src/session/pi-session.ts | 13 ----------- .../conversation-recovery.test.ts | 22 ++----------------- 3 files changed, 3 insertions(+), 37 deletions(-) diff --git a/src/server/server-runner.ts b/src/server/server-runner.ts index 1646653..1c8744e 100644 --- a/src/server/server-runner.ts +++ b/src/server/server-runner.ts @@ -27,10 +27,7 @@ export async function runServer(config: ServerConfig): Promise { modelId, historyStore: new FileConversationHistoryStore(config.sessionDirectory), createAgent: ({ conversationId }, snapshot) => - createSession(conversationId, { - committedLeafId: snapshot.committedLeafId, - committedMessageRoles: snapshot.messages.map(({ role }) => role), - }), + createSession(conversationId, { committedLeafId: snapshot.committedLeafId }), }); const server = Bun.serve({ hostname: config.hostname, diff --git a/src/session/pi-session.ts b/src/session/pi-session.ts index 28502f4..cc25db3 100644 --- a/src/session/pi-session.ts +++ b/src/session/pi-session.ts @@ -13,7 +13,6 @@ export type ModelDescription = Readonly<{ provider: string; id: string }>; export type PiSessionRecovery = Readonly<{ committedLeafId: string | null; - committedMessageRoles: readonly ("user" | "assistant")[]; }>; export type PiSessionFactoryConfig = Readonly<{ @@ -101,9 +100,6 @@ export function openSessionManagerForRecovery( if (recovery === undefined) return sessionManager; if (recovery.committedLeafId === null) { - if (recovery.committedMessageRoles.length > 0) { - throw new Error("Visible conversation history does not match committed Pi root"); - } sessionManager.resetLeaf(); return sessionManager; } @@ -115,15 +111,6 @@ export function openSessionManagerForRecovery( { cause: error }, ); } - const branchMessageRoles = sessionManager.getBranch() - .filter((entry) => entry.type === "message") - .map((entry) => entry.message.role); - if ( - branchMessageRoles.length !== recovery.committedMessageRoles.length || - branchMessageRoles.some((role, index) => role !== recovery.committedMessageRoles[index]) - ) { - throw new Error("Visible conversation history does not match committed Pi branch"); - } return sessionManager; } diff --git a/test/conversation/conversation-recovery.test.ts b/test/conversation/conversation-recovery.test.ts index f14190c..bb60c50 100644 --- a/test/conversation/conversation-recovery.test.ts +++ b/test/conversation/conversation-recovery.test.ts @@ -20,7 +20,7 @@ type Harness = Readonly<{ }>; async function harness(): Promise { - const root = await mkdtemp(join(tmpdir(), "ghl-conversation-recovery-")); + const root = await mkdtemp(join(tmpdir(), "stein-conversation-recovery-")); roots.push(root); const sessionDirectory = join(root, "sessions"); const workspaceDirectory = join(root, "workspace"); @@ -78,10 +78,7 @@ function reopen( state.sessionFile, state.sessionDirectory, state.workspaceDirectory, - { - committedLeafId: snapshot.committedLeafId, - committedMessageRoles: snapshot.messages.map(({ role }) => role), - }, + { committedLeafId: snapshot.committedLeafId }, ); } @@ -183,21 +180,6 @@ describe("committed conversation recovery", () => { expect(context(reopened)).toContain("standalone assistant"); }); - test("fails closed when snapshot messages do not match the committed branch", async () => { - const state = await harness(); - const manager = SessionManager.open( - state.sessionFile, - state.sessionDirectory, - state.workspaceDirectory, - ); - appendTurn(manager, "first"); - const secondLeafId = appendTurn(manager, "second"); - - expect(() => reopen(state, snapshot("first", secondLeafId))).toThrow( - "Visible conversation history does not match committed Pi branch", - ); - }); - test("fails closed for a missing session file and an unknown committed leaf", async () => { const missing = await harness(); expect(() => reopen(missing, snapshot("missing", "missing-leaf"))).toThrow( From 329f08ac25dbb0dd4902ed33410c315702bfcdc8 Mon Sep 17 00:00:00 2001 From: c0da Date: Thu, 6 Aug 2026 23:56:39 -0400 Subject: [PATCH 3/3] test: cover recovery after Pi retry --- .../conversation-recovery.test.ts | 49 ++++++++++++++++++- 1 file changed, 47 insertions(+), 2 deletions(-) diff --git a/test/conversation/conversation-recovery.test.ts b/test/conversation/conversation-recovery.test.ts index bb60c50..971646d 100644 --- a/test/conversation/conversation-recovery.test.ts +++ b/test/conversation/conversation-recovery.test.ts @@ -40,9 +40,17 @@ async function harness(): Promise { function appendTurn(manager: SessionManager, label: string): string { manager.appendMessage({ role: "user", content: `${label} user`, timestamp: Date.now() }); + return appendAssistant(manager, `${label} assistant`, "stop"); +} + +function appendAssistant( + manager: SessionManager, + content: string, + stopReason: "stop" | "error", +): string { return manager.appendMessage({ role: "assistant", - content: [{ type: "text", text: `${label} assistant` }], + content: [{ type: "text", text: content }], api: "openai-completions", provider: "test", model: "deterministic", @@ -54,7 +62,8 @@ function appendTurn(manager: SessionManager, label: string): string { totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason, + errorMessage: stopReason === "error" ? "fictional retryable failure" : undefined, timestamp: Date.now(), } as SessionMessage); } @@ -144,6 +153,42 @@ describe("committed conversation recovery", () => { expect(context(recovered)).toContain("second assistant"); }); + test("restores a committed retry without interpreting Pi message cardinality", async () => { + const state = await harness(); + const manager = SessionManager.open( + state.sessionFile, + state.sessionDirectory, + state.workspaceDirectory, + ); + manager.appendMessage({ + role: "user", + content: "retried user", + timestamp: Date.now(), + }); + appendAssistant(manager, "retryable error", "error"); + const committedLeafId = appendAssistant(manager, "retried assistant", "stop"); + const committed: ConversationHistorySnapshot = { + version: 1, + messages: [ + { role: "user", content: "retried user" }, + { role: "assistant", content: "retried assistant" }, + ], + committedLeafId, + }; + await state.historyStore.save(state.conversationId, committed, activeSignal()); + + const stored = await state.historyStore.load(state.conversationId); + const recovered = reopen(state, stored); + const persistedMessageRoles = recovered.getBranch() + .filter((entry) => entry.type === "message") + .map((entry) => entry.message.role); + + expect(stored.messages).toHaveLength(2); + expect(persistedMessageRoles).toEqual(["user", "assistant", "assistant"]); + expect(recovered.getLeafId()).toBe(committedLeafId); + expect(context(recovered)).toContain("retried assistant"); + }); + test("missing snapshot resets an abandoned first turn to committed root", async () => { const state = await harness(); const manager = SessionManager.open(