From 2030ac7438443eebc3277bc191cf72cb48d4e5bd Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 16 Jul 2026 23:29:29 -0700 Subject: [PATCH] fix(acp): preserve legacy replay state during doctor repair (#109685) * fix(acp): move legacy replay migration to doctor * refactor(acp): use Kysely for replay migration * style(acp): remove redundant replay coercions * style(acp): format replay migration --- .../check-database-first-legacy-stores.mjs | 1 + src/acp/event-ledger.test.ts | 71 +-- src/acp/event-ledger.ts | 158 ------- src/acp/server.startup.test.ts | 82 ---- src/acp/server.ts | 30 +- src/commands/doctor.e2e-harness.ts | 4 + src/infra/state-migrations.acp-replay.test.ts | 332 +++++++++++++ src/infra/state-migrations.acp-replay.ts | 444 ++++++++++++++++++ src/infra/state-migrations.doctor.ts | 19 + src/infra/state-migrations.test.ts | 74 +++ src/infra/state-migrations.types.ts | 4 + 11 files changed, 882 insertions(+), 337 deletions(-) create mode 100644 src/infra/state-migrations.acp-replay.test.ts create mode 100644 src/infra/state-migrations.acp-replay.ts diff --git a/scripts/check-database-first-legacy-stores.mjs b/scripts/check-database-first-legacy-stores.mjs index a40d280196a6..19f4b067fef0 100644 --- a/scripts/check-database-first-legacy-stores.mjs +++ b/scripts/check-database-first-legacy-stores.mjs @@ -122,6 +122,7 @@ const allowedRuntimeMigrationPaths = [ "src/commands/doctor-usage-cost-cache.ts", "src/infra/session-state-migration.ts", "src/infra/state-migrations.ts", + "src/infra/state-migrations.acp-replay.ts", "src/infra/state-migrations.tui-last-session.ts", "src/infra/state-migrations.commitments.ts", "src/infra/state-migrations.managed-outgoing-images.ts", diff --git a/src/acp/event-ledger.test.ts b/src/acp/event-ledger.test.ts index 8f0ba16a2d42..d8fa799ae391 100644 --- a/src/acp/event-ledger.test.ts +++ b/src/acp/event-ledger.test.ts @@ -1,15 +1,10 @@ -/** Tests ACP event ledger recording, replay, retention, and SQLite migration. */ -import fs from "node:fs/promises"; +/** Tests ACP event ledger recording, replay, retention, and SQLite persistence. */ import path from "node:path"; import { afterEach, describe, expect, it } from "vitest"; import { requireNodeSqlite } from "../infra/node-sqlite.js"; import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js"; import { withTempDir } from "../test-helpers/temp-dir.js"; -import { - createInMemoryAcpEventLedger, - createSqliteAcpEventLedger, - migrateFileAcpEventLedgerToSqlite, -} from "./event-ledger.js"; +import { createInMemoryAcpEventLedger, createSqliteAcpEventLedger } from "./event-ledger.js"; describe("ACP event ledger", () => { afterEach(() => { @@ -153,68 +148,6 @@ describe("ACP event ledger", () => { }); }); - it("imports legacy file-backed replay state into SQLite", async () => { - await withTempDir({ prefix: "openclaw-acp-ledger-" }, async (dir) => { - const filePath = path.join(dir, "acp", "event-ledger.json"); - const databasePath = path.join(dir, "openclaw.sqlite"); - await fs.mkdir(path.dirname(filePath), { recursive: true }); - await fs.writeFile( - filePath, - JSON.stringify({ - version: 1, - sessions: { - "session-1": { - sessionId: "session-1", - sessionKey: "agent:main:work", - cwd: "/work", - complete: true, - createdAt: 1000, - updatedAt: 1000, - nextSeq: 2, - events: [ - { - seq: 1, - at: 1000, - sessionId: "session-1", - sessionKey: "agent:main:work", - runId: "run-1", - update: { - sessionUpdate: "agent_message_chunk", - content: { type: "text", text: "Answer" }, - }, - }, - ], - }, - }, - }), - "utf8", - ); - - const migrated = await migrateFileAcpEventLedgerToSqlite({ - filePath, - path: databasePath, - archiveSource: true, - }); - const sqlite = createSqliteAcpEventLedger({ path: databasePath }); - const replay = await sqlite.readReplay({ - sessionId: "session-1", - sessionKey: "agent:main:work", - }); - - expect(migrated).toEqual({ - importedSessions: 1, - importedEvents: 1, - archived: true, - }); - expect(replay.complete).toBe(true); - expect(replay.events[0]?.update).toEqual({ - sessionUpdate: "agent_message_chunk", - content: { type: "text", text: "Answer" }, - }); - await expect(fs.stat(`${filePath}.migrated`)).resolves.toBeTruthy(); - }); - }); - it("marks SQLite-backed replay incomplete when event retention truncates history", async () => { await withTempDir({ prefix: "openclaw-acp-ledger-" }, async (dir) => { const ledger = createSqliteAcpEventLedger({ diff --git a/src/acp/event-ledger.ts b/src/acp/event-ledger.ts index c60063f04f45..185cae2b65fa 100644 --- a/src/acp/event-ledger.ts +++ b/src/acp/event-ledger.ts @@ -1,12 +1,7 @@ /** Persistent/replayable ACP event ledger implementations for session rehydration. */ -import fs from "node:fs/promises"; -import path from "node:path"; import type { DatabaseSync } from "node:sqlite"; import type { ContentBlock, SessionUpdate } from "@agentclientprotocol/sdk"; import { resolveIntegerOption } from "@openclaw/acp-core/numeric-options"; -import { resolveStateDir } from "../config/paths.js"; -import { withFileLock } from "../infra/file-lock.js"; -import { readJsonFile } from "../infra/json-files.js"; import { openOpenClawStateDatabase, type OpenClawStateDatabaseOptions, @@ -18,16 +13,6 @@ const LEDGER_VERSION = 1; const DEFAULT_MAX_SESSIONS = 200; const DEFAULT_MAX_EVENTS_PER_SESSION = 5_000; const DEFAULT_MAX_SERIALIZED_BYTES = 16 * 1024 * 1024; -const FILE_LEDGER_LOCK_OPTIONS = { - retries: { - retries: 8, - factor: 2, - minTimeout: 50, - maxTimeout: 5_000, - randomize: true, - }, - stale: 15_000, -} as const; type AcpEventLedgerEntry = { seq: number; @@ -178,60 +163,6 @@ function normalizeEvent(raw: unknown): AcpEventLedgerEntry | undefined { }; } -function normalizeSession(raw: unknown): LedgerSession | undefined { - if (!isRecord(raw)) { - return undefined; - } - const sessionId = raw.sessionId; - const sessionKey = raw.sessionKey; - const cwd = raw.cwd; - const createdAt = raw.createdAt; - const updatedAt = raw.updatedAt; - const nextSeq = raw.nextSeq; - if ( - typeof sessionId !== "string" || - typeof sessionKey !== "string" || - typeof cwd !== "string" || - typeof createdAt !== "number" || - !Number.isFinite(createdAt) || - typeof updatedAt !== "number" || - !Number.isFinite(updatedAt) || - typeof nextSeq !== "number" || - !Number.isInteger(nextSeq) || - nextSeq < 1 - ) { - return undefined; - } - const events = Array.isArray(raw.events) - ? raw.events.map(normalizeEvent).filter((event): event is AcpEventLedgerEntry => Boolean(event)) - : []; - return { - sessionId, - sessionKey, - cwd, - complete: raw.complete === true, - createdAt, - updatedAt, - nextSeq, - events, - }; -} - -function normalizeStore(raw: unknown): LedgerStore { - if (!isRecord(raw) || raw.version !== LEDGER_VERSION || !isRecord(raw.sessions)) { - return createEmptyStore(); - } - const sessions: Record = {}; - for (const [sessionId, value] of Object.entries(raw.sessions)) { - const session = normalizeSession(value); - if (!session || session.sessionId !== sessionId) { - continue; - } - sessions[sessionId] = session; - } - return { version: LEDGER_VERSION, sessions }; -} - function getOrCreateSession( state: MutableLedgerState, params: { @@ -441,95 +372,6 @@ export function createInMemoryAcpEventLedger(options: LedgerOptions = {}): AcpEv }); } -/** Resolves the legacy file-backed ACP ledger path under the OpenClaw state directory. */ -export function resolveDefaultAcpEventLedgerPath(env: NodeJS.ProcessEnv = process.env): string { - return path.join(resolveStateDir(env), "acp", "event-ledger.json"); -} - -async function fileExists(filePath: string): Promise { - try { - await fs.access(filePath); - return true; - } catch { - return false; - } -} - -/** Migrates a legacy file ledger into the SQLite state database, preserving replay order. */ -export async function migrateFileAcpEventLedgerToSqlite( - params: { filePath: string; archiveSource?: boolean } & OpenClawStateDatabaseOptions, -): Promise<{ importedSessions: number; importedEvents: number; archived?: boolean }> { - if (!(await fileExists(params.filePath))) { - return { importedSessions: 0, importedEvents: 0 }; - } - - const legacy = await withFileLock(params.filePath, FILE_LEDGER_LOCK_OPTIONS, async () => - normalizeStore(await readJsonFile(params.filePath)), - ); - const sessions = Object.values(legacy.sessions); - if (sessions.length === 0) { - return { importedSessions: 0, importedEvents: 0 }; - } - - let importedSessions = 0; - let importedEvents = 0; - runOpenClawStateWriteTransaction((database) => { - const sessionExists = database.db.prepare( - "SELECT 1 FROM acp_replay_sessions WHERE session_id = ?", - ); - const insertSession = database.db.prepare( - `INSERT INTO acp_replay_sessions ( - session_id, session_key, cwd, complete, created_at, updated_at, next_seq - ) VALUES (?, ?, ?, ?, ?, ?, ?)`, - ); - const insertEvent = database.db.prepare( - `INSERT OR IGNORE INTO acp_replay_events ( - session_id, seq, at, session_key, run_id, update_json - ) VALUES (?, ?, ?, ?, ?, ?)`, - ); - for (const session of sessions) { - if (sessionExists.get(session.sessionId)) { - continue; - } - insertSession.run( - session.sessionId, - session.sessionKey, - session.cwd, - session.complete ? 1 : 0, - session.createdAt, - session.updatedAt, - session.nextSeq, - ); - importedSessions++; - for (const event of session.events) { - const result = insertEvent.run( - event.sessionId, - event.seq, - event.at, - event.sessionKey, - event.runId ?? null, - JSON.stringify(event.update), - ); - importedEvents += Number(result.changes); - } - } - }, params); - - if (params.archiveSource !== true || importedSessions === 0) { - return { importedSessions, importedEvents }; - } - const archivePath = `${params.filePath}.migrated`; - try { - if (!(await fileExists(archivePath))) { - await fs.rename(params.filePath, archivePath); - return { importedSessions, importedEvents, archived: true }; - } - } catch { - // The SQLite import succeeded; archiving is a best-effort cleanup. - } - return { importedSessions, importedEvents }; -} - function normalizeSqliteInteger(value: number | bigint | null): number { if (typeof value === "bigint") { return Number(value); diff --git a/src/acp/server.startup.test.ts b/src/acp/server.startup.test.ts index 90bdf11d475e..d75057276e80 100644 --- a/src/acp/server.startup.test.ts +++ b/src/acp/server.startup.test.ts @@ -41,7 +41,6 @@ const mockState = vi.hoisted(() => ({ startProxy: vi.fn(async (_configForTest: unknown) => null as unknown), stopProxy: vi.fn(async (_handle: unknown) => {}), closeOpenClawStateDatabase: vi.fn(), - migrateEventLedger: vi.fn(async () => ({ importedSessions: 0, importedEvents: 0 })), gatewayStopDeferred: null as { resolve: () => void; promise: Promise; @@ -185,8 +184,6 @@ vi.mock("../state/openclaw-state-db.js", () => ({ vi.mock("./event-ledger.js", () => ({ createSqliteAcpEventLedger: vi.fn(() => ({})), - migrateFileAcpEventLedgerToSqlite: () => mockState.migrateEventLedger(), - resolveDefaultAcpEventLedgerPath: vi.fn(() => "/tmp/acp-events.json"), })); vi.mock("../infra/net/proxy/proxy-lifecycle.js", () => ({ @@ -329,8 +326,6 @@ describe("serveAcpGateway startup", () => { mockState.startProxy.mockReset(); mockState.stopProxy.mockReset(); mockState.closeOpenClawStateDatabase.mockReset(); - mockState.migrateEventLedger.mockReset(); - mockState.migrateEventLedger.mockResolvedValue({ importedSessions: 0, importedEvents: 0 }); mockState.gatewayStopDeferred = null; mockState.startProxy.mockResolvedValue(null); mockState.stopProxy.mockResolvedValue(undefined); @@ -597,83 +592,6 @@ describe("serveAcpGateway startup", () => { } }); - it("waits for both ledger migration and Gateway teardown before closing", async () => { - let resolveMigration!: () => void; - const migrationPromise = new Promise<{ importedSessions: number; importedEvents: number }>( - (resolve) => { - resolveMigration = () => resolve({ importedSessions: 1, importedEvents: 1 }); - }, - ); - mockState.migrateEventLedger.mockImplementation(async () => await migrationPromise); - let resolveStop!: () => void; - const stopPromise = new Promise((resolve) => { - resolveStop = resolve; - }); - mockState.gatewayStopDeferred = { resolve: resolveStop, promise: stopPromise }; - const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); - - try { - const servePromise = serveAcpGateway({}); - await vi.waitFor(() => { - expect(mockState.gateways).toHaveLength(1); - }); - getMockGateway().emitHello(); - await vi.waitFor(() => { - expect(mockState.migrateEventLedger).toHaveBeenCalledOnce(); - }); - - signalHandlers.get("SIGTERM")?.(); - await Promise.resolve(); - expect(mockState.closeOpenClawStateDatabase).not.toHaveBeenCalled(); - - resolveMigration(); - await Promise.resolve(); - expect(mockState.closeOpenClawStateDatabase).not.toHaveBeenCalled(); - - resolveStop(); - await servePromise; - - expect(mockState.agentSideConnectionCtor).not.toHaveBeenCalled(); - expect(mockState.closeOpenClawStateDatabase).toHaveBeenCalledOnce(); - } finally { - onceSpy.mockRestore(); - } - }); - - it("closes after a pending ledger migration rejects during shutdown", async () => { - let rejectMigration!: (err: Error) => void; - const migrationPromise = new Promise<{ importedSessions: number; importedEvents: number }>( - (_resolve, reject) => { - rejectMigration = reject; - }, - ); - mockState.migrateEventLedger.mockImplementation(async () => await migrationPromise); - const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); - - try { - const servePromise = serveAcpGateway({}); - await vi.waitFor(() => { - expect(mockState.gateways).toHaveLength(1); - }); - getMockGateway().emitHello(); - await vi.waitFor(() => { - expect(mockState.migrateEventLedger).toHaveBeenCalledOnce(); - }); - - signalHandlers.get("SIGTERM")?.(); - await Promise.resolve(); - expect(mockState.closeOpenClawStateDatabase).not.toHaveBeenCalled(); - - rejectMigration(new Error("sqlite busy")); - await expect(servePromise).resolves.toBeUndefined(); - - expect(mockState.agentSideConnectionCtor).not.toHaveBeenCalled(); - expect(mockState.closeOpenClawStateDatabase).toHaveBeenCalledOnce(); - } finally { - onceSpy.mockRestore(); - } - }); - it("closes a real node:sqlite DatabaseSync handle through serveAcpGateway shutdown", async () => { // Use the real state-db module to open and verify a DatabaseSync handle — // this proves the full serveAcpGateway → shutdown → close path, not just diff --git a/src/acp/server.ts b/src/acp/server.ts index f5f6a2c63bb6..3e5d9a61432a 100644 --- a/src/acp/server.ts +++ b/src/acp/server.ts @@ -23,11 +23,7 @@ import { GatewayClient } from "../gateway/client.js"; import { isMainModule } from "../infra/is-main.js"; import { routeLogsToStderr } from "../logging/console.js"; import { closeOpenClawStateDatabase } from "../state/openclaw-state-db.js"; -import { - createSqliteAcpEventLedger, - migrateFileAcpEventLedgerToSqlite, - resolveDefaultAcpEventLedgerPath, -} from "./event-ledger.js"; +import { createSqliteAcpEventLedger } from "./event-ledger.js"; import { readSecretFromFile } from "./secret-file.js"; import { AcpGatewayAgent } from "./translator.js"; import { normalizeAcpProvenanceMode } from "./types.js"; @@ -54,7 +50,6 @@ export async function serveAcpGateway(opts: AcpServerOptions = {}): Promise | null = null; let onGatewayReadyResolve!: () => void; let onGatewayReadyReject!: (err: Error) => void; let gatewayReadySettled = false; @@ -134,13 +129,10 @@ export async function serveAcpGateway(opts: AcpServerOptions = {}): Promise {}) ?? Promise.resolve(); const gatewayStop = gateway.stopAndWait().catch((err: unknown) => { console.warn(`acp: gateway stop failed during shutdown: ${String(err)}`); }); - // Migration and transport teardown are independent. Close only after both - // settle so neither can touch or reopen the process-owned SQLite handle. - await Promise.all([startupSettlement, gatewayStop]); + await gatewayStop; closeStateDatabase(); onClosed(); }; @@ -177,24 +169,6 @@ export async function serveAcpGateway(opts: AcpServerOptions = {}): Promise { + const sourcePath = path.join(stateDir, "acp", "event-ledger.json"); + await fs.mkdir(path.dirname(sourcePath), { recursive: true }); + await fs.writeFile(sourcePath, JSON.stringify(value), "utf8"); + return sourcePath; +} + +describe("legacy ACP replay doctor migration", () => { + afterEach(() => { + closeOpenClawStateDatabaseForTest(); + }); + + it("detects legacy state only for explicit doctor repair", async () => { + await withTempDir({ prefix: "openclaw-acp-replay-migration-" }, async (stateDir) => { + await writeLegacyStore(stateDir); + expect(detectLegacyAcpReplayLedger({ stateDir }).hasLegacy).toBe(false); + expect( + detectLegacyAcpReplayLedger({ stateDir, doctorOnlyStateMigrations: true }).hasLegacy, + ).toBe(true); + }); + }); + + it("imports, verifies, and removes the retired JSON ledger", async () => { + await withTempDir({ prefix: "openclaw-acp-replay-migration-" }, async (stateDir) => { + const sourcePath = await writeLegacyStore(stateDir); + const result = await migrateLegacyAcpReplayLedger({ + detected: detectLegacyAcpReplayLedger({ + stateDir, + doctorOnlyStateMigrations: true, + }), + stateDir, + }); + + expect(result).toEqual({ + changes: [ + "Migrated 1 ACP replay session(s) and 1 event(s) → shared SQLite state", + `Removed retired ACP replay ledger ${sourcePath}`, + ], + warnings: [], + }); + await expect(fs.stat(sourcePath)).rejects.toMatchObject({ code: "ENOENT" }); + const replay = await createSqliteAcpEventLedger({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }).readReplay({ sessionId: "session-1", sessionKey: "agent:main:work" }); + expect(replay.complete).toBe(true); + expect(replay.events[0]?.update).toEqual({ + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Answer" }, + }); + const db = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }).db; + const aggregate = db + .prepare("SELECT estimated_bytes AS total FROM acp_replay_sessions WHERE session_id = ?") + .get("session-1") as { total: number | bigint }; + const groundTruth = db + .prepare( + `SELECT length(s.session_id) + length(s.session_key) + length(s.cwd) + 32 + + COALESCE(SUM(length(e.session_id) + length(e.session_key) + length(e.update_json) + + COALESCE(length(e.run_id), 0) + 32), 0) AS total + FROM acp_replay_sessions s + LEFT JOIN acp_replay_events e ON e.session_id = s.session_id + WHERE s.session_id = ? + GROUP BY s.session_id`, + ) + .get("session-1") as { total: number | bigint }; + expect(Number(aggregate.total)).toBe(Number(groundTruth.total)); + }); + }); + + it("resumes a claimed source without deleting a replacement ledger", async () => { + await withTempDir({ prefix: "openclaw-acp-replay-migration-" }, async (stateDir) => { + const sourcePath = await writeLegacyStore(stateDir); + const claimPath = `${sourcePath}.doctor-import`; + await fs.rename(sourcePath, claimPath); + const replacement = legacyStore(); + const replacementSession = replacement.sessions["session-1"]; + await writeLegacyStore(stateDir, { + ...replacement, + sessions: { + "session-2": { + ...replacementSession, + sessionId: "session-2", + events: replacementSession.events.map((event) => { + const next = structuredClone(event); + next.sessionId = "session-2"; + return next; + }), + }, + }, + }); + + const first = await migrateLegacyAcpReplayLedger({ + detected: detectLegacyAcpReplayLedger({ + stateDir, + doctorOnlyStateMigrations: true, + }), + stateDir, + }); + + expect(first.warnings).toEqual([ + `A newer ACP replay ledger remains at ${sourcePath}; rerun doctor to migrate it`, + ]); + await expect(fs.stat(claimPath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(fs.stat(sourcePath)).resolves.toBeDefined(); + + const second = await migrateLegacyAcpReplayLedger({ + detected: detectLegacyAcpReplayLedger({ + stateDir, + doctorOnlyStateMigrations: true, + }), + stateDir, + }); + expect(second.warnings).toEqual([]); + await expect(fs.stat(sourcePath)).rejects.toMatchObject({ code: "ENOENT" }); + const db = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }).db; + const rows = db + .prepare("SELECT session_id FROM acp_replay_sessions ORDER BY session_id") + .all() as Array<{ session_id: string }>; + expect(rows.map((row) => row.session_id)).toEqual(["session-1", "session-2"]); + }); + }); + + it("retains malformed state without partially importing it", async () => { + await withTempDir({ prefix: "openclaw-acp-replay-migration-" }, async (stateDir) => { + const sourcePath = await writeLegacyStore(stateDir, { + ...legacyStore(), + sessions: { broken: { sessionId: "broken" } }, + }); + const result = await migrateLegacyAcpReplayLedger({ + detected: detectLegacyAcpReplayLedger({ + stateDir, + doctorOnlyStateMigrations: true, + }), + stateDir, + }); + + expect(result.changes).toEqual([]); + expect(result.warnings[0]).toContain("legacy ACP replay session broken is invalid"); + await expect(fs.stat(sourcePath)).resolves.toBeDefined(); + await expect( + createSqliteAcpEventLedger({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }).readReplayBySessionId({ sessionId: "broken" }), + ).resolves.toEqual({ complete: false, events: [] }); + }); + }); + + it("removes a retry source when its prior import already exists", async () => { + await withTempDir({ prefix: "openclaw-acp-replay-migration-" }, async (stateDir) => { + const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir }; + const ledger = createSqliteAcpEventLedger({ env, now: () => 1_000 }); + await ledger.startSession({ + sessionId: "session-1", + sessionKey: "agent:main:work", + cwd: "/work", + complete: true, + }); + await ledger.recordUpdate({ + sessionId: "session-1", + sessionKey: "agent:main:work", + runId: "run-1", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Answer" }, + }, + }); + const db = openOpenClawStateDatabase({ env }).db; + db.exec(` + UPDATE acp_replay_events SET estimated_bytes = 0; + UPDATE acp_replay_sessions SET estimated_bytes = 0; + `); + const sourcePath = await writeLegacyStore(stateDir); + + const result = await migrateLegacyAcpReplayLedger({ + detected: detectLegacyAcpReplayLedger({ + stateDir, + doctorOnlyStateMigrations: true, + }), + stateDir, + }); + + expect(result.warnings).toEqual([]); + expect(result.changes).toContain( + "Kept 1 existing ACP replay session(s) from shared SQLite state", + ); + await expect(fs.stat(sourcePath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(ledger.readReplayBySessionId({ sessionId: "session-1" })).resolves.toMatchObject( + { complete: true, sessionKey: "agent:main:work" }, + ); + expect( + Number( + ( + db + .prepare( + "SELECT estimated_bytes AS total FROM acp_replay_sessions WHERE session_id = ?", + ) + .get("session-1") as { total: number | bigint } + ).total, + ), + ).toBeGreaterThan(0); + }); + }); + + it("retains a conflicting retry source instead of discarding changed events", async () => { + await withTempDir({ prefix: "openclaw-acp-replay-migration-" }, async (stateDir) => { + const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir }; + const ledger = createSqliteAcpEventLedger({ env, now: () => 2_000 }); + await ledger.startSession({ + sessionId: "session-1", + sessionKey: "agent:main:canonical", + cwd: "/current", + complete: true, + }); + const store = legacyStore(); + const original = store.sessions["session-1"]; + const sourcePath = await writeLegacyStore(stateDir, { + ...store, + sessions: { + "new-session": { + ...structuredClone(original), + sessionId: "new-session", + events: original.events.map((event) => ({ + ...structuredClone(event), + sessionId: "new-session", + })), + }, + "session-1": original, + }, + }); + + const result = await migrateLegacyAcpReplayLedger({ + detected: detectLegacyAcpReplayLedger({ + stateDir, + doctorOnlyStateMigrations: true, + }), + stateDir, + }); + + expect(result.changes).toEqual([]); + expect(result.warnings[0]).toContain( + "canonical ACP replay session session-1 conflicts with the legacy source", + ); + await expect(fs.stat(sourcePath)).resolves.toBeDefined(); + await expect(ledger.readReplayBySessionId({ sessionId: "session-1" })).resolves.toMatchObject( + { complete: true, sessionKey: "agent:main:canonical" }, + ); + await expect(ledger.readReplayBySessionId({ sessionId: "new-session" })).resolves.toEqual({ + complete: false, + events: [], + }); + }); + }); + + it("retains a source containing an impossible zero event sequence", async () => { + await withTempDir({ prefix: "openclaw-acp-replay-migration-" }, async (stateDir) => { + const store = legacyStore(); + store.sessions["session-1"].events[0]!.seq = 0; + const sourcePath = await writeLegacyStore(stateDir, store); + + const result = await migrateLegacyAcpReplayLedger({ + detected: detectLegacyAcpReplayLedger({ + stateDir, + doctorOnlyStateMigrations: true, + }), + stateDir, + }); + + expect(result.changes).toEqual([]); + expect(result.warnings[0]).toContain("contains an invalid event"); + await expect(fs.stat(sourcePath)).resolves.toBeDefined(); + }); + }); + + it("runtime ignores the retired JSON ledger until doctor imports it", async () => { + await withTempDir({ prefix: "openclaw-acp-replay-migration-" }, async (stateDir) => { + await writeLegacyStore(stateDir); + await expect( + createSqliteAcpEventLedger({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }).readReplayBySessionId({ sessionId: "session-1" }), + ).resolves.toEqual({ complete: false, events: [] }); + }); + }); +}); diff --git a/src/infra/state-migrations.acp-replay.ts b/src/infra/state-migrations.acp-replay.ts new file mode 100644 index 000000000000..6af77142f6d2 --- /dev/null +++ b/src/infra/state-migrations.acp-replay.ts @@ -0,0 +1,444 @@ +// Doctor-only import for the retired ACP replay JSON ledger. +import { createHash } from "node:crypto"; +import fsSync from "node:fs"; +import fs from "node:fs/promises"; +import path from "node:path"; +import type { DatabaseSync } from "node:sqlite"; +import { isDeepStrictEqual } from "node:util"; +import type { SessionUpdate } from "@agentclientprotocol/sdk"; +import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js"; +import { runOpenClawStateWriteTransaction } from "../state/openclaw-state-db.js"; +import { isRecord } from "../utils.js"; +import { withFileLock } from "./file-lock.js"; +import { + executeSqliteQuerySync, + executeSqliteQueryTakeFirstSync, + getNodeSqliteKysely, +} from "./kysely-sync.js"; +import type { LegacyStateDetection, MigrationMessages } from "./state-migrations.types.js"; + +const LEGACY_LEDGER_VERSION = 1; +const LEGACY_LEDGER_LOCK_OPTIONS = { + retries: { + retries: 8, + factor: 2, + minTimeout: 50, + maxTimeout: 5_000, + randomize: true, + }, + stale: 15_000, + staleRecovery: "fail-closed", +} as const; + +type LegacyAcpReplayEvent = { + seq: number; + at: number; + sessionId: string; + sessionKey: string; + runId?: string; + update: SessionUpdate; +}; + +type LegacyAcpReplaySession = { + sessionId: string; + sessionKey: string; + cwd: string; + complete: boolean; + createdAt: number; + updatedAt: number; + nextSeq: number; + events: LegacyAcpReplayEvent[]; +}; + +type LegacySourceIdentity = { + dev: number | bigint; + ino: number | bigint; + mtimeMs: number | bigint; + sha256: string; + size: number | bigint; +}; + +type AcpReplayMigrationDatabase = Pick< + OpenClawStateKyselyDatabase, + "acp_replay_events" | "acp_replay_sessions" +>; + +function resolveLegacyAcpReplayLedgerPath(stateDir: string): string { + return path.join(stateDir, "acp", "event-ledger.json"); +} + +function resolveLegacyAcpReplayClaimPath(sourcePath: string): string { + return `${sourcePath}.doctor-import`; +} + +/** Detect the retired ledger only when an explicit doctor flow opts in. */ +export function detectLegacyAcpReplayLedger(params: { + stateDir: string; + doctorOnlyStateMigrations?: boolean; +}): LegacyStateDetection["acpReplayLedger"] { + const sourcePath = resolveLegacyAcpReplayLedgerPath(params.stateDir); + const claimPath = resolveLegacyAcpReplayClaimPath(sourcePath); + return { + sourcePath, + hasLegacy: + params.doctorOnlyStateMigrations === true && + (fsSync.existsSync(sourcePath) || fsSync.existsSync(claimPath)), + }; +} + +function parseLegacyEvent(raw: unknown, sessionId: string): LegacyAcpReplayEvent { + if (!isRecord(raw) || !isRecord(raw.update)) { + throw new Error(`legacy ACP replay session ${sessionId} contains an invalid event`); + } + if ( + typeof raw.seq !== "number" || + !Number.isInteger(raw.seq) || + raw.seq < 1 || + typeof raw.at !== "number" || + !Number.isFinite(raw.at) || + raw.sessionId !== sessionId || + typeof raw.sessionKey !== "string" || + typeof raw.update.sessionUpdate !== "string" + ) { + throw new Error(`legacy ACP replay session ${sessionId} contains an invalid event`); + } + if (raw.runId !== undefined && (typeof raw.runId !== "string" || raw.runId.length === 0)) { + throw new Error(`legacy ACP replay session ${sessionId} contains an invalid run id`); + } + return { + seq: raw.seq, + at: raw.at, + sessionId, + sessionKey: raw.sessionKey, + ...(typeof raw.runId === "string" ? { runId: raw.runId } : {}), + update: structuredClone(raw.update) as SessionUpdate, + }; +} + +function parseLegacySession(raw: unknown, expectedSessionId: string): LegacyAcpReplaySession { + if ( + !isRecord(raw) || + raw.sessionId !== expectedSessionId || + typeof raw.sessionKey !== "string" || + typeof raw.cwd !== "string" || + typeof raw.complete !== "boolean" || + typeof raw.createdAt !== "number" || + !Number.isFinite(raw.createdAt) || + typeof raw.updatedAt !== "number" || + !Number.isFinite(raw.updatedAt) || + typeof raw.nextSeq !== "number" || + !Number.isInteger(raw.nextSeq) || + raw.nextSeq < 1 || + !Array.isArray(raw.events) + ) { + throw new Error(`legacy ACP replay session ${expectedSessionId} is invalid`); + } + const events = raw.events.map((event) => parseLegacyEvent(event, expectedSessionId)); + const sequences = new Set(events.map((event) => event.seq)); + const maxSeq = events.reduce((max, event) => Math.max(max, event.seq), 0); + if (sequences.size !== events.length || raw.nextSeq <= maxSeq) { + throw new Error(`legacy ACP replay session ${expectedSessionId} has invalid sequencing`); + } + return { + sessionId: expectedSessionId, + sessionKey: raw.sessionKey, + cwd: raw.cwd, + complete: raw.complete, + createdAt: raw.createdAt, + updatedAt: raw.updatedAt, + nextSeq: raw.nextSeq, + events: events.toSorted((left, right) => left.seq - right.seq), + }; +} + +function parseLegacyLedger(raw: string): LegacyAcpReplaySession[] { + const parsed = JSON.parse(raw) as unknown; + if (!isRecord(parsed) || parsed.version !== LEGACY_LEDGER_VERSION || !isRecord(parsed.sessions)) { + throw new Error("legacy ACP replay ledger must be a version 1 JSON object"); + } + return Object.entries(parsed.sessions).map(([sessionId, session]) => + parseLegacySession(session, sessionId), + ); +} + +function estimateSessionBytes(session: LegacyAcpReplaySession): number { + return session.sessionId.length + session.sessionKey.length + session.cwd.length + 32; +} + +function estimateEventBytes(event: LegacyAcpReplayEvent, updateJson: string): number { + return ( + event.sessionId.length + + event.sessionKey.length + + updateJson.length + + (event.runId?.length ?? 0) + + 32 + ); +} + +function sourceIdentity( + stat: Awaited>, + raw: string, +): LegacySourceIdentity { + return { + dev: stat.dev, + ino: stat.ino, + mtimeMs: stat.mtimeMs, + sha256: createHash("sha256").update(raw).digest("hex"), + size: stat.size, + }; +} + +function sourceIdentityMatches(left: LegacySourceIdentity, right: LegacySourceIdentity): boolean { + return ( + left.dev === right.dev && + left.ino === right.ino && + left.mtimeMs === right.mtimeMs && + left.sha256 === right.sha256 && + left.size === right.size + ); +} + +function reconcileCanonicalSession(db: DatabaseSync, session: LegacyAcpReplaySession): boolean { + const replayDb = getNodeSqliteKysely(db); + const stored = executeSqliteQueryTakeFirstSync( + db, + replayDb + .selectFrom("acp_replay_sessions") + .select([ + "session_key", + "cwd", + "complete", + "created_at", + "updated_at", + "next_seq", + "estimated_bytes", + ]) + .where("session_id", "=", session.sessionId), + ); + if ( + !stored || + stored.session_key !== session.sessionKey || + stored.cwd !== session.cwd || + stored.complete !== (session.complete ? 1 : 0) || + stored.created_at !== session.createdAt || + stored.updated_at !== session.updatedAt || + stored.next_seq !== session.nextSeq + ) { + return false; + } + + const storedEvents = executeSqliteQuerySync( + db, + replayDb + .selectFrom("acp_replay_events") + .select(["seq", "at", "session_key", "run_id", "update_json", "estimated_bytes"]) + .where("session_id", "=", session.sessionId) + .orderBy("seq", "asc"), + ).rows; + if (storedEvents.length !== session.events.length) { + return false; + } + + const expectedEventBytes: number[] = []; + for (const [index, event] of session.events.entries()) { + const storedEvent = storedEvents[index]; + if (!storedEvent) { + return false; + } + let storedUpdate: unknown; + try { + storedUpdate = JSON.parse(storedEvent.update_json); + } catch { + return false; + } + if ( + storedEvent.seq !== event.seq || + storedEvent.at !== event.at || + storedEvent.session_key !== event.sessionKey || + storedEvent.run_id !== (event.runId ?? null) || + !isDeepStrictEqual(storedUpdate, event.update) + ) { + return false; + } + expectedEventBytes.push(estimateEventBytes(event, JSON.stringify(event.update))); + } + + for (const [index, event] of session.events.entries()) { + const expectedBytes = expectedEventBytes[index]; + if (expectedBytes !== undefined && storedEvents[index]?.estimated_bytes !== expectedBytes) { + executeSqliteQuerySync( + db, + replayDb + .updateTable("acp_replay_events") + .set({ estimated_bytes: expectedBytes }) + .where("session_id", "=", session.sessionId) + .where("seq", "=", event.seq), + ); + } + } + const expectedSessionBytes = + estimateSessionBytes(session) + expectedEventBytes.reduce((sum, value) => sum + value, 0); + if (stored.estimated_bytes !== expectedSessionBytes) { + executeSqliteQuerySync( + db, + replayDb + .updateTable("acp_replay_sessions") + .set({ estimated_bytes: expectedSessionBytes }) + .where("session_id", "=", session.sessionId), + ); + } + return true; +} + +/** Import, verify, and remove the retired JSON ledger during explicit doctor repair. */ +export async function migrateLegacyAcpReplayLedger(params: { + detected: LegacyStateDetection["acpReplayLedger"]; + stateDir: string; +}): Promise { + const changes: string[] = []; + const warnings: string[] = []; + if (!params.detected.hasLegacy) { + return { changes, warnings }; + } + + try { + const result = await withFileLock( + params.detected.sourcePath, + LEGACY_LEDGER_LOCK_OPTIONS, + async () => { + const claimPath = resolveLegacyAcpReplayClaimPath(params.detected.sourcePath); + const resumedClaim = fsSync.existsSync(claimPath); + const activePath = resumedClaim ? claimPath : params.detected.sourcePath; + const before = await fs.lstat(activePath); + if (!before.isFile() || before.isSymbolicLink()) { + throw new Error("legacy ACP replay source is not a regular non-symlink file"); + } + const raw = await fs.readFile(activePath, "utf8"); + const identity = sourceIdentity(before, raw); + const sessions = parseLegacyLedger(raw); + let importedSessions = 0; + let importedEvents = 0; + let retainedSessions = 0; + let claimedThisRun = false; + + try { + if (!resumedClaim) { + await fs.rename(params.detected.sourcePath, claimPath); + claimedThisRun = true; + const claimedStat = await fs.lstat(claimPath); + const claimedRaw = await fs.readFile(claimPath, "utf8"); + if (!sourceIdentityMatches(identity, sourceIdentity(claimedStat, claimedRaw))) { + throw new Error("legacy ACP replay source changed while doctor was claiming it"); + } + } + + runOpenClawStateWriteTransaction( + ({ db }) => { + const replayDb = getNodeSqliteKysely(db); + const missingSessions: LegacyAcpReplaySession[] = []; + for (const session of sessions) { + const existing = executeSqliteQueryTakeFirstSync( + db, + replayDb + .selectFrom("acp_replay_sessions") + .select("session_id") + .where("session_id", "=", session.sessionId), + ); + if (existing) { + if (!reconcileCanonicalSession(db, session)) { + throw new Error( + `canonical ACP replay session ${session.sessionId} conflicts with the legacy source`, + ); + } + retainedSessions += 1; + continue; + } + missingSessions.push(session); + } + + for (const session of missingSessions) { + let estimatedBytes = estimateSessionBytes(session); + executeSqliteQuerySync( + db, + replayDb.insertInto("acp_replay_sessions").values({ + session_id: session.sessionId, + session_key: session.sessionKey, + cwd: session.cwd, + complete: session.complete ? 1 : 0, + created_at: session.createdAt, + updated_at: session.updatedAt, + next_seq: session.nextSeq, + estimated_bytes: estimatedBytes, + }), + ); + for (const event of session.events) { + const updateJson = JSON.stringify(event.update); + const eventBytes = estimateEventBytes(event, updateJson); + executeSqliteQuerySync( + db, + replayDb.insertInto("acp_replay_events").values({ + session_id: event.sessionId, + seq: event.seq, + at: event.at, + session_key: event.sessionKey, + run_id: event.runId ?? null, + update_json: updateJson, + estimated_bytes: eventBytes, + }), + ); + estimatedBytes += eventBytes; + importedEvents += 1; + } + executeSqliteQuerySync( + db, + replayDb + .updateTable("acp_replay_sessions") + .set({ estimated_bytes: estimatedBytes }) + .where("session_id", "=", session.sessionId), + ); + if (!reconcileCanonicalSession(db, session)) { + throw new Error( + `failed verifying imported ACP replay session ${session.sessionId}`, + ); + } + importedSessions += 1; + } + }, + { env: { ...process.env, OPENCLAW_STATE_DIR: params.stateDir } }, + ); + await fs.unlink(claimPath); + return { + importedSessions, + importedEvents, + retainedSessions, + pendingSource: fsSync.existsSync(params.detected.sourcePath), + }; + } catch (error) { + if (claimedThisRun && !fsSync.existsSync(params.detected.sourcePath)) { + await fs.rename(claimPath, params.detected.sourcePath).catch(() => {}); + } + throw error; + } + }, + ); + changes.push( + `Migrated ${result.importedSessions} ACP replay session(s) and ${result.importedEvents} event(s) → shared SQLite state`, + ); + if (result.retainedSessions > 0) { + changes.push( + `Kept ${result.retainedSessions} existing ACP replay session(s) from shared SQLite state`, + ); + } + changes.push(`Removed retired ACP replay ledger ${params.detected.sourcePath}`); + if (result.pendingSource) { + warnings.push( + `A newer ACP replay ledger remains at ${params.detected.sourcePath}; rerun doctor to migrate it`, + ); + } + } catch (error) { + warnings.push( + `Failed migrating legacy ACP replay ledger ${params.detected.sourcePath}: ${String(error)}`, + ); + } + return { changes, warnings }; +} diff --git a/src/infra/state-migrations.doctor.ts b/src/infra/state-migrations.doctor.ts index 94a946dbde2b..f2d77e06f6d9 100644 --- a/src/infra/state-migrations.doctor.ts +++ b/src/infra/state-migrations.doctor.ts @@ -27,6 +27,10 @@ import { repairOpenClawStateDatabaseSchema, type OpenClawStateDatabaseSchemaMigration, } from "../state/openclaw-state-db.js"; +import { + detectLegacyAcpReplayLedger, + migrateLegacyAcpReplayLedger, +} from "./state-migrations.acp-replay.js"; import { detectLegacyApnsRegistrations, migrateLegacyApnsRegistrations, @@ -392,6 +396,10 @@ export async function detectLegacyStateMigrations(params: { stateDir, doctorOnlyStateMigrations: params.doctorOnlyStateMigrations, }); + const acpReplayLedger = detectLegacyAcpReplayLedger({ + stateDir, + doctorOnlyStateMigrations: params.doctorOnlyStateMigrations, + }); const managedOutgoingImages = detectLegacyManagedOutgoingImages({ stateDir, doctorOnlyStateMigrations: params.doctorOnlyStateMigrations, @@ -570,6 +578,9 @@ export async function detectLegacyStateMigrations(params: { if (commitments.hasLegacy) { preview.push("- Commitments: legacy JSON file → shared SQLite state"); } + if (acpReplayLedger.hasLegacy) { + preview.push("- ACP replay ledger: legacy JSON file → shared SQLite state"); + } if (managedOutgoingImages.hasLegacy) { preview.push("- Managed outgoing images: legacy record JSON → shared SQLite state"); } @@ -675,6 +686,7 @@ export async function detectLegacyStateMigrations(params: { }, tuiLastSessions, commitments, + acpReplayLedger, managedOutgoingImages, apns, workspace, @@ -867,6 +879,10 @@ export async function runLegacyStateMigrations(params: { detected: detected.commitments, stateDir: detected.stateDir, }); + const acpReplayLedger = await migrateLegacyAcpReplayLedger({ + detected: detected.acpReplayLedger, + stateDir: detected.stateDir, + }); const managedOutgoingImages = migrateLegacyManagedOutgoingImages({ detected: detected.managedOutgoingImages, stateDir: detected.stateDir, @@ -931,6 +947,7 @@ export async function runLegacyStateMigrations(params: { updateCheck, tuiLastSessions, commitments, + acpReplayLedger, managedOutgoingImages, apns, workspace, @@ -954,6 +971,7 @@ export async function runLegacyStateMigrations(params: { ...currentConversationBindings.changes, ...tuiLastSessions.changes, ...commitments.changes, + ...acpReplayLedger.changes, ...managedOutgoingImages.changes, ...apns.changes, ...workspace.changes, @@ -984,6 +1002,7 @@ export async function runLegacyStateMigrations(params: { ...currentConversationBindings.warnings, ...tuiLastSessions.warnings, ...commitments.warnings, + ...acpReplayLedger.warnings, ...managedOutgoingImages.warnings, ...apns.warnings, ...workspace.warnings, diff --git a/src/infra/state-migrations.test.ts b/src/infra/state-migrations.test.ts index 98aac0bd7c49..1d928ec564ff 100644 --- a/src/infra/state-migrations.test.ts +++ b/src/infra/state-migrations.test.ts @@ -2124,6 +2124,80 @@ describe("state migrations", () => { await expectMissingPath(sourcePath); }); + it("routes explicit Doctor repair through the ACP replay SQLite importer", async () => { + const root = await createTempDir(); + const stateDir = path.join(root, ".openclaw"); + const env = createEnv(stateDir); + const cfg = createConfig(); + const sourcePath = path.join(stateDir, "acp", "event-ledger.json"); + await fs.mkdir(path.dirname(sourcePath), { recursive: true }); + await fs.writeFile( + sourcePath, + JSON.stringify({ + version: 1, + sessions: { + "doctor-acp-session": { + sessionId: "doctor-acp-session", + sessionKey: "agent:main:doctor-acp", + cwd: "/work", + complete: true, + createdAt: 1, + updatedAt: 2, + nextSeq: 2, + events: [ + { + seq: 1, + at: 2, + sessionId: "doctor-acp-session", + sessionKey: "agent:main:doctor-acp", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Doctor import" }, + }, + }, + ], + }, + }, + }), + "utf8", + ); + + const runtimeDetection = await detectLegacyStateMigrations({ + cfg, + env, + homedir: () => root, + }); + expect(runtimeDetection.acpReplayLedger.hasLegacy).toBe(false); + + const detected = await detectLegacyStateMigrations({ + cfg, + env, + homedir: () => root, + doctorOnlyStateMigrations: true, + }); + expect(detected.acpReplayLedger.hasLegacy).toBe(true); + expect(detected.preview).toContain( + "- ACP replay ledger: legacy JSON file → shared SQLite state", + ); + + const result = await runLegacyStateMigrations({ detected, config: cfg, env }); + + expect(result.warnings).toStrictEqual([]); + expect(result.changes).toContain( + "Migrated 1 ACP replay session(s) and 1 event(s) → shared SQLite state", + ); + const row = openOpenClawStateDatabase({ env }) + .db.prepare( + "SELECT session_key, estimated_bytes FROM acp_replay_sessions WHERE session_id = ?", + ) + .get("doctor-acp-session") as + | { session_key: string; estimated_bytes: number | bigint } + | undefined; + expect(row?.session_key).toBe("agent:main:doctor-acp"); + expect(Number(row?.estimated_bytes ?? 0)).toBeGreaterThan(0); + await expectMissingPath(sourcePath); + }); + it("routes explicit Doctor repair through the Web Push SQLite importer", async () => { const root = await createTempDir(); const stateDir = path.join(root, ".openclaw"); diff --git a/src/infra/state-migrations.types.ts b/src/infra/state-migrations.types.ts index 0da5252996b8..dfd0ec653fb3 100644 --- a/src/infra/state-migrations.types.ts +++ b/src/infra/state-migrations.types.ts @@ -101,6 +101,10 @@ export type LegacyStateDetection = { sourcePath: string; hasLegacy: boolean; }; + acpReplayLedger: { + sourcePath: string; + hasLegacy: boolean; + }; managedOutgoingImages: { sourceDir: string; hasLegacy: boolean;