vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-22 11:21:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(acp): preserve legacy replay state during doctor repair (#109685)

Browse Source 
* 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
This commit is contained in:
Peter Steinberger
2026-07-16 23:29:29 -07:00
committed by GitHub
parent fb946267ae
commit 2030ac7438
11 changed files with 882 additions and 337 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
scripts/check-database-first-legacy-stores.mjs
View File

@@ -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",

71
src/acp/event-ledger.test.ts
View File

@@ -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({

158
src/acp/event-ledger.ts
View File

@@ -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<string, LedgerSession> = {};
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<boolean> {
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);

82
src/acp/server.startup.test.ts
View File

@@ -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<void>;
@@ -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<void>((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

30
src/acp/server.ts
View File

@@ -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<void
onClosed = resolve;
});
let stopped = false;
let startupWork: Promise<unknown> | null = null;
let onGatewayReadyResolve!: () => void;
let onGatewayReadyReject!: (err: Error) => void;
let gatewayReadySettled = false;
@@ -134,13 +129,10 @@ export async function serveAcpGateway(opts: AcpServerOptions = {}): Promise<void
const activeAgent = agent;
agent = null;
activeAgent?.shutdown();
const startupSettlement = startupWork?.catch(() => {}) ?? 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<void
},
}),
);
startupWork = migrateFileAcpEventLedgerToSqlite({
filePath: resolveDefaultAcpEventLedgerPath(process.env),
archiveSource: true,
});
try {
await startupWork;
} catch (err) {
if (stopped) {
return closed;
}
await shutdown();
throw err;
} finally {
startupWork = null;
}
if (stopped) {
return closed;
}
const eventLedger = createSqliteAcpEventLedger();
void new AgentSideConnection(

4
src/commands/doctor.e2e-harness.ts
View File

@@ -284,6 +284,10 @@ function createLegacyStateMigrationDetectionResult(params?: {
sourcePath: "/tmp/state/commitments/commitments.json",
hasLegacy: false,
},
acpReplayLedger: {
sourcePath: "/tmp/state/acp/event-ledger.json",
hasLegacy: false,
},
managedOutgoingImages: {
sourceDir: "/tmp/state/media/outgoing/records",
hasLegacy: false,

332
src/infra/state-migrations.acp-replay.test.ts Normal file
View File

@@ -0,0 +1,332 @@
// Covers fail-closed doctor import of the retired ACP replay JSON ledger.
import fs from "node:fs/promises";
import path from "node:path";
import { afterEach, describe, expect, it } from "vitest";
import { createSqliteAcpEventLedger } from "../acp/event-ledger.js";
import {
closeOpenClawStateDatabaseForTest,
openOpenClawStateDatabase,
} from "../state/openclaw-state-db.js";
import { withTempDir } from "../test-helpers/temp-dir.js";
import {
detectLegacyAcpReplayLedger,
migrateLegacyAcpReplayLedger,
} from "./state-migrations.acp-replay.js";
function legacyStore() {
return {
version: 1,
sessions: {
"session-1": {
sessionId: "session-1",
sessionKey: "agent:main:work",
cwd: "/work",
complete: true,
createdAt: 1_000,
updatedAt: 1_000,
nextSeq: 2,
events: [
{
seq: 1,
at: 1_000,
sessionId: "session-1",
sessionKey: "agent:main:work",
runId: "run-1",
update: {
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Answer" },
},
},
],
},
},
};
}
async function writeLegacyStore(stateDir: string, value: unknown = legacyStore()): Promise<string> {
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: [] });
});
});
});

444
src/infra/state-migrations.acp-replay.ts Normal file
View File

@@ -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<ReturnType<typeof fs.lstat>>,
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<AcpReplayMigrationDatabase>(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<MigrationMessages> {
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<AcpReplayMigrationDatabase>(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 };
}

19
src/infra/state-migrations.doctor.ts
View File

@@ -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,

74
src/infra/state-migrations.test.ts
View File

@@ -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");

4
src/infra/state-migrations.types.ts
View File

@@ -101,6 +101,10 @@ export type LegacyStateDetection = {
sourcePath: string;
hasLegacy: boolean;
};
acpReplayLedger: {
sourcePath: string;
hasLegacy: boolean;
};
managedOutgoingImages: {
sourceDir: string;
hasLegacy: boolean;