mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-09 15:59:30 +00:00
fix(sqlite): migrate legacy memory and proxy state
This commit is contained in:
Vincent Koc
parent
8413984e87
commit
7485dd6492
9 changed files with 1286 additions and 3 deletions
|
|
@ -389,6 +389,9 @@ The branch already has a real shared SQLite base:
|
|||
path/source row shape and serialized embedding compatibility. These tables
|
||||
are derived/search cache, not canonical transcript storage; they can be
|
||||
deleted and rebuilt from memory workspace files and configured sources.
|
||||
Opening a shipped generic-name memory index migrates its metadata, sources,
|
||||
chunks, and embedding cache into the canonical tables; derived FTS/vector
|
||||
tables are rebuilt under their canonical names.
|
||||
- Subagent run recovery state now lives in typed shared `subagent_runs` rows
|
||||
with indexed child, requester, and controller session keys. The old
|
||||
`subagents/runs.json` file is doctor migration input only.
|
||||
|
|
@ -1724,6 +1727,9 @@ Keep shared coordination state in `state/openclaw.sqlite`:
|
|||
WAL, and busy-timeout settings. Payload bytes are gzip-compressed in
|
||||
`capture_blobs.data`; there is no debug proxy runtime sidecar DB override,
|
||||
blob directory, or proxy-capture-only generated schema/codegen target.
|
||||
Doctor/startup migration imports shipped `debug-proxy/capture.sqlite` rows
|
||||
and referenced payload blobs, including active legacy DB/blob environment
|
||||
overrides, then archives those sources while leaving CA certificates intact.
|
||||
|
||||
This phase also deletes duplicate sidecar openers, permission helpers, WAL
|
||||
setup, filesystem pruning, and compatibility writers from those subsystems.
|
||||
|
|
|
|||
|
|
@ -1669,6 +1669,24 @@ describe("memory index", () => {
|
|||
expect(status.vector?.available).toBe(available);
|
||||
});
|
||||
|
||||
it("drops the shipped legacy vector table after loading sqlite-vec", async () => {
|
||||
const cfg = createCfg({ vectorEnabled: true });
|
||||
const manager = await getPersistentManager(cfg);
|
||||
const db = Reflect.get(manager, "db") as DatabaseSync;
|
||||
db.exec("CREATE TABLE chunks_vec (id TEXT PRIMARY KEY, embedding BLOB)");
|
||||
|
||||
const available = await manager.probeVectorStoreAvailability?.();
|
||||
if (!available) {
|
||||
return;
|
||||
}
|
||||
|
||||
expect(
|
||||
db
|
||||
.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'chunks_vec'")
|
||||
.get(),
|
||||
).toBeUndefined();
|
||||
});
|
||||
|
||||
it("probes sqlite vector store availability without initializing embeddings", async () => {
|
||||
forceNoProvider = true;
|
||||
const cfg = createCfg({
|
||||
|
|
|
|||
|
|
@ -145,6 +145,7 @@ type MemoryReindexRetryState = {
|
|||
|
||||
const META_KEY = "memory_index_meta_v1";
|
||||
const VECTOR_TABLE = MEMORY_INDEX_VECTOR_TABLE;
|
||||
const LEGACY_VECTOR_TABLE = "chunks_vec";
|
||||
const FTS_TABLE = MEMORY_INDEX_FTS_TABLE;
|
||||
const EMBEDDING_CACHE_TABLE = MEMORY_EMBEDDING_CACHE_TABLE;
|
||||
const SESSION_DIRTY_DEBOUNCE_MS = 5000;
|
||||
|
|
@ -167,6 +168,12 @@ const log = createSubsystemLogger("memory");
|
|||
const TEST_MEMORY_WATCH_FACTORY_KEY = Symbol.for("openclaw.test.memoryWatchFactory");
|
||||
const TEST_MEMORY_NATIVE_WATCH_FACTORY_KEY = Symbol.for("openclaw.test.memoryNativeWatchFactory");
|
||||
|
||||
function memoryTableExists(db: DatabaseSync, tableName: string): boolean {
|
||||
return Boolean(
|
||||
db.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?").get(tableName),
|
||||
);
|
||||
}
|
||||
|
||||
type NativeMemoryWatchPair = {
|
||||
dir: string;
|
||||
main: fsSync.FSWatcher;
|
||||
|
|
@ -642,6 +649,9 @@ export abstract class MemoryManagerSyncOps {
|
|||
}
|
||||
this.vector.extensionPath = loaded.extensionPath;
|
||||
this.vector.available = true;
|
||||
if (this.dropLegacyVectorTable()) {
|
||||
this.dirty = true;
|
||||
}
|
||||
return true;
|
||||
} catch (err) {
|
||||
const message = formatErrorMessage(err);
|
||||
|
|
@ -653,7 +663,7 @@ export abstract class MemoryManagerSyncOps {
|
|||
}
|
||||
|
||||
private ensureVectorTable(dimensions: number): void {
|
||||
if (this.vector.dims === dimensions) {
|
||||
if (this.vector.dims === dimensions && memoryTableExists(this.db, VECTOR_TABLE)) {
|
||||
return;
|
||||
}
|
||||
if (!this.dropVectorTable()) {
|
||||
|
|
@ -668,6 +678,19 @@ export abstract class MemoryManagerSyncOps {
|
|||
this.vector.dims = dimensions;
|
||||
}
|
||||
|
||||
private dropLegacyVectorTable(): boolean {
|
||||
if (!memoryTableExists(this.db, LEGACY_VECTOR_TABLE)) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
this.db.exec(`DROP TABLE ${LEGACY_VECTOR_TABLE}`);
|
||||
return true;
|
||||
} catch (err) {
|
||||
log.debug(`Failed to drop ${LEGACY_VECTOR_TABLE}: ${formatErrorMessage(err)}`);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private dropVectorTable(): boolean {
|
||||
try {
|
||||
this.db.exec(`DROP TABLE IF EXISTS ${VECTOR_TABLE}`);
|
||||
|
|
|
|||
175
packages/memory-host-sdk/src/host/memory-schema.test.ts
Normal file
175
packages/memory-host-sdk/src/host/memory-schema.test.ts
Normal file
|
|
@ -0,0 +1,175 @@
|
|||
// Memory schema tests cover canonical table creation and shipped-name migration.
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { ensureMemoryIndexSchema } from "./memory-schema.js";
|
||||
|
||||
describe("memory index schema", () => {
|
||||
it("migrates shipped generic tables into canonical memory tables", () => {
|
||||
const db = new DatabaseSync(":memory:");
|
||||
try {
|
||||
db.exec(`
|
||||
CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
||||
CREATE TABLE files (
|
||||
path TEXT PRIMARY KEY,
|
||||
source TEXT NOT NULL DEFAULT 'memory',
|
||||
hash TEXT NOT NULL,
|
||||
mtime INTEGER NOT NULL,
|
||||
size INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE chunks (
|
||||
id TEXT PRIMARY KEY,
|
||||
path TEXT NOT NULL,
|
||||
source TEXT NOT NULL DEFAULT 'memory',
|
||||
start_line INTEGER NOT NULL,
|
||||
end_line INTEGER NOT NULL,
|
||||
hash TEXT NOT NULL,
|
||||
model TEXT NOT NULL,
|
||||
text TEXT NOT NULL,
|
||||
embedding TEXT NOT NULL,
|
||||
updated_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE embedding_cache (
|
||||
provider TEXT NOT NULL,
|
||||
model TEXT NOT NULL,
|
||||
provider_key TEXT NOT NULL,
|
||||
hash TEXT NOT NULL,
|
||||
embedding TEXT NOT NULL,
|
||||
dims INTEGER,
|
||||
updated_at INTEGER NOT NULL,
|
||||
PRIMARY KEY (provider, model, provider_key, hash)
|
||||
);
|
||||
CREATE VIRTUAL TABLE chunks_fts USING fts5(
|
||||
text, id UNINDEXED, path UNINDEXED, source UNINDEXED, model UNINDEXED,
|
||||
start_line UNINDEXED, end_line UNINDEXED
|
||||
);
|
||||
INSERT INTO meta VALUES ('memory_index_meta_v1', '{"vectorDims":3}');
|
||||
INSERT INTO files VALUES ('MEMORY.md', 'memory', 'file-hash', 10, 20);
|
||||
INSERT INTO chunks VALUES (
|
||||
'chunk-1', 'MEMORY.md', 'memory', 1, 2, 'chunk-hash', 'embed-model',
|
||||
'remember this', '[1,0,0]', 30
|
||||
);
|
||||
INSERT INTO embedding_cache VALUES (
|
||||
'openai', 'embed-model', 'key', 'chunk-hash', '[1,0,0]', 3, 40
|
||||
);
|
||||
INSERT INTO chunks_fts VALUES (
|
||||
'remember this', 'chunk-1', 'MEMORY.md', 'memory', 'embed-model', 1, 2
|
||||
);
|
||||
`);
|
||||
|
||||
const result = ensureMemoryIndexSchema({
|
||||
db,
|
||||
cacheEnabled: true,
|
||||
ftsEnabled: true,
|
||||
});
|
||||
|
||||
expect(result.ftsAvailable).toBe(true);
|
||||
expect(db.prepare("SELECT * FROM memory_index_sources").all()).toEqual([
|
||||
{ path: "MEMORY.md", source: "memory", hash: "file-hash", mtime: 10, size: 20 },
|
||||
]);
|
||||
expect(db.prepare("SELECT id, text FROM memory_index_chunks").all()).toEqual([
|
||||
{ id: "chunk-1", text: "remember this" },
|
||||
]);
|
||||
expect(db.prepare("SELECT provider, hash FROM memory_embedding_cache").all()).toEqual([
|
||||
{ provider: "openai", hash: "chunk-hash" },
|
||||
]);
|
||||
expect(
|
||||
db
|
||||
.prepare(
|
||||
"SELECT name FROM sqlite_master WHERE type = 'table' AND name IN ('meta', 'files', 'chunks', 'embedding_cache', 'chunks_fts')",
|
||||
)
|
||||
.all(),
|
||||
).toEqual([]);
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("leaves unrelated generic tables untouched", () => {
|
||||
const db = new DatabaseSync(":memory:");
|
||||
try {
|
||||
db.exec(`
|
||||
CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL, owner TEXT);
|
||||
CREATE TABLE files (
|
||||
path TEXT PRIMARY KEY,
|
||||
source TEXT NOT NULL,
|
||||
hash TEXT NOT NULL,
|
||||
mtime INTEGER NOT NULL,
|
||||
size INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE chunks (
|
||||
id TEXT PRIMARY KEY,
|
||||
path TEXT NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
start_line INTEGER NOT NULL,
|
||||
end_line INTEGER NOT NULL,
|
||||
hash TEXT NOT NULL,
|
||||
model TEXT NOT NULL,
|
||||
text TEXT NOT NULL,
|
||||
embedding TEXT NOT NULL,
|
||||
updated_at INTEGER NOT NULL
|
||||
);
|
||||
`);
|
||||
|
||||
ensureMemoryIndexSchema({
|
||||
db,
|
||||
cacheEnabled: false,
|
||||
ftsEnabled: false,
|
||||
});
|
||||
|
||||
expect(
|
||||
db
|
||||
.prepare(
|
||||
"SELECT name FROM sqlite_master WHERE type = 'table' AND name IN ('meta', 'files', 'chunks') ORDER BY name",
|
||||
)
|
||||
.all(),
|
||||
).toEqual([{ name: "chunks" }, { name: "files" }, { name: "meta" }]);
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps legacy tables when canonical rows conflict", () => {
|
||||
const db = new DatabaseSync(":memory:");
|
||||
try {
|
||||
db.exec(`
|
||||
CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
||||
CREATE TABLE files (
|
||||
path TEXT PRIMARY KEY,
|
||||
source TEXT NOT NULL DEFAULT 'memory',
|
||||
hash TEXT NOT NULL,
|
||||
mtime INTEGER NOT NULL,
|
||||
size INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE chunks (
|
||||
id TEXT PRIMARY KEY,
|
||||
path TEXT NOT NULL,
|
||||
source TEXT NOT NULL DEFAULT 'memory',
|
||||
start_line INTEGER NOT NULL,
|
||||
end_line INTEGER NOT NULL,
|
||||
hash TEXT NOT NULL,
|
||||
model TEXT NOT NULL,
|
||||
text TEXT NOT NULL,
|
||||
embedding TEXT NOT NULL,
|
||||
updated_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE memory_index_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
||||
INSERT INTO meta VALUES ('memory_index_meta_v1', 'legacy');
|
||||
INSERT INTO memory_index_meta VALUES ('memory_index_meta_v1', 'canonical');
|
||||
`);
|
||||
|
||||
expect(() =>
|
||||
ensureMemoryIndexSchema({
|
||||
db,
|
||||
cacheEnabled: false,
|
||||
ftsEnabled: false,
|
||||
}),
|
||||
).toThrow("legacy memory meta rows conflict");
|
||||
expect(db.prepare("SELECT value FROM meta").get()).toEqual({ value: "legacy" });
|
||||
expect(db.prepare("SELECT value FROM memory_index_meta").get()).toEqual({
|
||||
value: "canonical",
|
||||
});
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
@ -12,6 +12,175 @@ export const MEMORY_INDEX_STATE_TABLE = "memory_index_state";
|
|||
export const MEMORY_INDEX_FTS_TABLE = "memory_index_chunks_fts";
|
||||
export const MEMORY_INDEX_VECTOR_TABLE = "memory_index_chunks_vec";
|
||||
|
||||
const LEGACY_MEMORY_INDEX_TRIGGERS = [
|
||||
"memory_files_revision_after_insert",
|
||||
"memory_files_revision_after_update",
|
||||
"memory_files_revision_after_delete",
|
||||
"memory_chunks_revision_after_insert",
|
||||
"memory_chunks_revision_after_update",
|
||||
"memory_chunks_revision_after_delete",
|
||||
] as const;
|
||||
|
||||
function tableHasExactColumns(
|
||||
db: DatabaseSync,
|
||||
tableName: string,
|
||||
expected: readonly string[],
|
||||
): boolean {
|
||||
const rows = db.prepare(`PRAGMA table_info(${tableName})`).all() as Array<{ name?: unknown }>;
|
||||
const columns = new Set(rows.flatMap((row) => (typeof row.name === "string" ? [row.name] : [])));
|
||||
return columns.size === expected.length && expected.every((column) => columns.has(column));
|
||||
}
|
||||
|
||||
function assertLegacyRowsCopied(db: DatabaseSync, query: string, tableName: string): void {
|
||||
const row = db.prepare(query).get() as { missing?: unknown } | undefined;
|
||||
if (Number(row?.missing ?? 0) > 0) {
|
||||
throw new Error(`legacy memory ${tableName} rows conflict with canonical memory index rows`);
|
||||
}
|
||||
}
|
||||
|
||||
function migrateLegacyMemoryIndexTables(db: DatabaseSync): void {
|
||||
const hasLegacyCoreTables =
|
||||
tableHasExactColumns(db, "meta", ["key", "value"]) &&
|
||||
tableHasExactColumns(db, "files", ["path", "source", "hash", "mtime", "size"]) &&
|
||||
tableHasExactColumns(db, "chunks", [
|
||||
"id",
|
||||
"path",
|
||||
"source",
|
||||
"start_line",
|
||||
"end_line",
|
||||
"hash",
|
||||
"model",
|
||||
"text",
|
||||
"embedding",
|
||||
"updated_at",
|
||||
]);
|
||||
if (!hasLegacyCoreTables) {
|
||||
return;
|
||||
}
|
||||
|
||||
db.exec("SAVEPOINT migrate_legacy_memory_index_tables");
|
||||
try {
|
||||
db.exec(`
|
||||
INSERT OR IGNORE INTO ${MEMORY_INDEX_META_TABLE} (key, value)
|
||||
SELECT key, value FROM meta;
|
||||
|
||||
INSERT OR IGNORE INTO ${MEMORY_INDEX_SOURCES_TABLE} (path, source, hash, mtime, size)
|
||||
SELECT path, source, hash, mtime, size FROM files;
|
||||
|
||||
INSERT OR IGNORE INTO ${MEMORY_INDEX_CHUNKS_TABLE} (
|
||||
id, path, source, start_line, end_line, hash, model, text, embedding, updated_at
|
||||
)
|
||||
SELECT id, path, source, start_line, end_line, hash, model, text, embedding, updated_at
|
||||
FROM chunks;
|
||||
`);
|
||||
assertLegacyRowsCopied(
|
||||
db,
|
||||
`SELECT COUNT(*) AS missing
|
||||
FROM meta AS legacy
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1 FROM ${MEMORY_INDEX_META_TABLE} AS canonical
|
||||
WHERE canonical.key = legacy.key AND canonical.value IS legacy.value
|
||||
)`,
|
||||
"meta",
|
||||
);
|
||||
assertLegacyRowsCopied(
|
||||
db,
|
||||
`SELECT COUNT(*) AS missing
|
||||
FROM files AS legacy
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1 FROM ${MEMORY_INDEX_SOURCES_TABLE} AS canonical
|
||||
WHERE canonical.path = legacy.path
|
||||
AND canonical.source IS legacy.source
|
||||
AND canonical.hash IS legacy.hash
|
||||
AND canonical.mtime IS legacy.mtime
|
||||
AND canonical.size IS legacy.size
|
||||
)`,
|
||||
"files",
|
||||
);
|
||||
assertLegacyRowsCopied(
|
||||
db,
|
||||
`SELECT COUNT(*) AS missing
|
||||
FROM chunks AS legacy
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1 FROM ${MEMORY_INDEX_CHUNKS_TABLE} AS canonical
|
||||
WHERE canonical.id = legacy.id
|
||||
AND canonical.path IS legacy.path
|
||||
AND canonical.source IS legacy.source
|
||||
AND canonical.start_line IS legacy.start_line
|
||||
AND canonical.end_line IS legacy.end_line
|
||||
AND canonical.hash IS legacy.hash
|
||||
AND canonical.model IS legacy.model
|
||||
AND canonical.text IS legacy.text
|
||||
AND canonical.embedding IS legacy.embedding
|
||||
AND canonical.updated_at IS legacy.updated_at
|
||||
)`,
|
||||
"chunks",
|
||||
);
|
||||
if (
|
||||
tableHasExactColumns(db, "embedding_cache", [
|
||||
"provider",
|
||||
"model",
|
||||
"provider_key",
|
||||
"hash",
|
||||
"embedding",
|
||||
"dims",
|
||||
"updated_at",
|
||||
])
|
||||
) {
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS ${MEMORY_EMBEDDING_CACHE_TABLE} (
|
||||
provider TEXT NOT NULL,
|
||||
model TEXT NOT NULL,
|
||||
provider_key TEXT NOT NULL,
|
||||
hash TEXT NOT NULL,
|
||||
embedding TEXT NOT NULL,
|
||||
dims INTEGER,
|
||||
updated_at INTEGER NOT NULL,
|
||||
PRIMARY KEY (provider, model, provider_key, hash)
|
||||
);
|
||||
INSERT OR IGNORE INTO ${MEMORY_EMBEDDING_CACHE_TABLE} (
|
||||
provider, model, provider_key, hash, embedding, dims, updated_at
|
||||
)
|
||||
SELECT provider, model, provider_key, hash, embedding, dims, updated_at
|
||||
FROM embedding_cache;
|
||||
`);
|
||||
assertLegacyRowsCopied(
|
||||
db,
|
||||
`SELECT COUNT(*) AS missing
|
||||
FROM embedding_cache AS legacy
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1 FROM ${MEMORY_EMBEDDING_CACHE_TABLE} AS canonical
|
||||
WHERE canonical.provider = legacy.provider
|
||||
AND canonical.model = legacy.model
|
||||
AND canonical.provider_key = legacy.provider_key
|
||||
AND canonical.hash = legacy.hash
|
||||
AND canonical.embedding IS legacy.embedding
|
||||
AND canonical.dims IS legacy.dims
|
||||
AND canonical.updated_at IS legacy.updated_at
|
||||
)`,
|
||||
"embedding_cache",
|
||||
);
|
||||
db.exec("DROP TABLE embedding_cache");
|
||||
}
|
||||
for (const trigger of LEGACY_MEMORY_INDEX_TRIGGERS) {
|
||||
db.exec(`DROP TRIGGER IF EXISTS ${trigger}`);
|
||||
}
|
||||
// FTS/vector tables are derived from canonical chunk rows. FTS can be
|
||||
// removed here; sqlite-vec cleanup waits until that extension is loaded.
|
||||
db.exec(`
|
||||
DROP TABLE IF EXISTS chunks_fts;
|
||||
DROP TABLE chunks;
|
||||
DROP TABLE files;
|
||||
DROP TABLE meta;
|
||||
RELEASE migrate_legacy_memory_index_tables;
|
||||
`);
|
||||
} catch (err) {
|
||||
db.exec("ROLLBACK TO migrate_legacy_memory_index_tables");
|
||||
db.exec("RELEASE migrate_legacy_memory_index_tables");
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
/** Ensure canonical memory index tables and the optional FTS table exist. */
|
||||
export function ensureMemoryIndexSchema(params: {
|
||||
db: DatabaseSync;
|
||||
|
|
@ -86,6 +255,7 @@ export function ensureMemoryIndexSchema(params: {
|
|||
CREATE INDEX IF NOT EXISTS idx_memory_index_chunks_source
|
||||
ON ${MEMORY_INDEX_CHUNKS_TABLE}(source);
|
||||
`);
|
||||
migrateLegacyMemoryIndexTables(params.db);
|
||||
if (params.cacheEnabled) {
|
||||
params.db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS ${MEMORY_EMBEDDING_CACHE_TABLE} (
|
||||
|
|
|
|||
|
|
@ -47,7 +47,10 @@ const rawSqliteAllowPathGroups = {
|
|||
"backup snapshot maintenance": ["src/commands/backup-verify.ts", "src/infra/backup-create.ts"],
|
||||
"agent auth profile read-only bootstrap": ["src/agents/auth-profiles/sqlite.ts"],
|
||||
"read-only SQLite status probes": ["src/commands/status.scan.shared.ts"],
|
||||
"doctor legacy state migration": ["src/infra/state-migrations.ts"],
|
||||
"doctor legacy state migration": [
|
||||
"src/infra/state-migrations.ts",
|
||||
"src/infra/state-migrations.debug-proxy.ts",
|
||||
],
|
||||
"shared database stores with direct DatabaseSync access": ["src/proxy-capture/store.sqlite.ts"],
|
||||
"Kysely-backed stores that own a DatabaseSync boundary": [
|
||||
"src/acp/event-ledger.ts",
|
||||
|
|
|
|||
|
|
@ -1,8 +1,10 @@
|
|||
// Doctor state migration tests cover legacy state moves, archive markers, and repair behavior.
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import { gunzipSync, gzipSync } from "node:zlib";
|
||||
import { afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import type { OpenClawConfig } from "../config/config.js";
|
||||
import type { SessionEntry } from "../config/sessions/types.js";
|
||||
|
|
@ -18,7 +20,10 @@ import {
|
|||
writePersistedInstalledPluginIndex,
|
||||
} from "../plugins/installed-plugin-index-store.js";
|
||||
import type { InstalledPluginInstallRecordInfo } from "../plugins/installed-plugin-index.js";
|
||||
import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
openOpenClawStateDatabase,
|
||||
} from "../state/openclaw-state-db.js";
|
||||
import { loadTaskFlowRegistryStateFromSqlite } from "../tasks/task-flow-registry.store.sqlite.js";
|
||||
import { loadTaskRegistryStateFromSqlite } from "../tasks/task-registry.store.sqlite.js";
|
||||
import {
|
||||
|
|
@ -349,6 +354,110 @@ function writeLegacyPluginStateSidecar(root: string): string {
|
|||
return sourcePath;
|
||||
}
|
||||
|
||||
function writeLegacyDebugProxyCaptureSidecar(
|
||||
root: string,
|
||||
overrides: { sourcePath?: string; blobDir?: string } = {},
|
||||
): {
|
||||
sourcePath: string;
|
||||
blobDir: string;
|
||||
blobId: string;
|
||||
} {
|
||||
const sourcePath = overrides.sourcePath ?? path.join(root, "debug-proxy", "capture.sqlite");
|
||||
const blobDir = overrides.blobDir ?? path.join(root, "debug-proxy", "blobs");
|
||||
fs.mkdirSync(path.dirname(sourcePath), { recursive: true });
|
||||
fs.mkdirSync(blobDir, { recursive: true });
|
||||
const payload = Buffer.from('{"legacy":true}');
|
||||
const sha256 = createHash("sha256").update(payload).digest("hex");
|
||||
const blobId = sha256.slice(0, 24);
|
||||
fs.writeFileSync(path.join(blobDir, `${blobId}.bin.gz`), gzipSync(payload));
|
||||
const sqlite = requireNodeSqlite();
|
||||
const db = new sqlite.DatabaseSync(sourcePath);
|
||||
try {
|
||||
db.exec(`
|
||||
CREATE TABLE capture_sessions (
|
||||
id TEXT PRIMARY KEY,
|
||||
started_at INTEGER NOT NULL,
|
||||
ended_at INTEGER,
|
||||
mode TEXT NOT NULL,
|
||||
source_scope TEXT NOT NULL,
|
||||
source_process TEXT NOT NULL,
|
||||
proxy_url TEXT,
|
||||
db_path TEXT NOT NULL,
|
||||
blob_dir TEXT NOT NULL
|
||||
);
|
||||
CREATE TABLE capture_events (
|
||||
id INTEGER PRIMARY KEY,
|
||||
session_id TEXT NOT NULL,
|
||||
ts INTEGER NOT NULL,
|
||||
source_scope TEXT NOT NULL,
|
||||
source_process TEXT NOT NULL,
|
||||
protocol TEXT NOT NULL,
|
||||
direction TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
flow_id TEXT NOT NULL,
|
||||
method TEXT,
|
||||
host TEXT,
|
||||
path TEXT,
|
||||
status INTEGER,
|
||||
close_code INTEGER,
|
||||
content_type TEXT,
|
||||
headers_json TEXT,
|
||||
data_text TEXT,
|
||||
data_blob_id TEXT,
|
||||
data_sha256 TEXT,
|
||||
error_text TEXT,
|
||||
meta_json TEXT
|
||||
);
|
||||
`);
|
||||
db.prepare(
|
||||
`INSERT INTO capture_sessions (
|
||||
id, started_at, ended_at, mode, source_scope, source_process, proxy_url, db_path, blob_dir
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
).run(
|
||||
"legacy-session",
|
||||
100,
|
||||
200,
|
||||
"proxy-run",
|
||||
"openclaw",
|
||||
"openclaw",
|
||||
"http://127.0.0.1:8080",
|
||||
sourcePath,
|
||||
blobDir,
|
||||
);
|
||||
db.prepare(
|
||||
`INSERT INTO capture_events (
|
||||
session_id, ts, source_scope, source_process, protocol, direction, kind, flow_id,
|
||||
method, host, path, status, close_code, content_type, headers_json, data_text,
|
||||
data_blob_id, data_sha256, error_text, meta_json
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
).run(
|
||||
"legacy-session",
|
||||
150,
|
||||
"openclaw",
|
||||
"openclaw",
|
||||
"https",
|
||||
"outbound",
|
||||
"request",
|
||||
"legacy-flow",
|
||||
"POST",
|
||||
"api.example.com",
|
||||
"/v1/test",
|
||||
null,
|
||||
null,
|
||||
"application/json",
|
||||
'{"content-type":"application/json"}',
|
||||
'{"legacy":true}',
|
||||
blobId,
|
||||
sha256,
|
||||
null,
|
||||
'{"provider":"test"}',
|
||||
);
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
return { sourcePath, blobDir, blobId };
|
||||
}
|
||||
|
||||
async function writeExistingPluginInstallIndex(
|
||||
root: string,
|
||||
installRecords: Record<string, InstalledPluginInstallRecordInfo>,
|
||||
|
|
@ -1576,6 +1685,225 @@ describe("doctor legacy state migrations", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("imports the shipped debug proxy capture sidecar into shared state", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const { sourcePath, blobDir, blobId } = writeLegacyDebugProxyCaptureSidecar(root);
|
||||
const certDir = path.join(root, "debug-proxy", "certs");
|
||||
fs.mkdirSync(certDir, { recursive: true });
|
||||
fs.writeFileSync(path.join(certDir, "ca.pem"), "keep");
|
||||
|
||||
const detected = await detectLegacyStateMigrations({
|
||||
cfg: {},
|
||||
env: { OPENCLAW_STATE_DIR: root } as NodeJS.ProcessEnv,
|
||||
});
|
||||
expect(detected.debugProxyCaptureSidecar).toEqual({
|
||||
sourcePath,
|
||||
blobDir,
|
||||
hasLegacy: true,
|
||||
});
|
||||
expect(detected.preview).toContain(
|
||||
`- Debug proxy capture sidecar: ${sourcePath} → shared SQLite state`,
|
||||
);
|
||||
|
||||
const result = await runLegacyStateMigrations({ detected });
|
||||
|
||||
expect(result.warnings).toStrictEqual([]);
|
||||
expect(result.changes).toContain(
|
||||
"Migrated 1 debug proxy capture session, 1 event, and 1 blob → shared SQLite state",
|
||||
);
|
||||
expect(fs.existsSync(sourcePath)).toBe(false);
|
||||
expect(fs.existsSync(`${sourcePath}.migrated`)).toBe(true);
|
||||
expect(fs.existsSync(blobDir)).toBe(false);
|
||||
expect(fs.existsSync(`${blobDir}.migrated`)).toBe(true);
|
||||
expect(fs.readFileSync(path.join(certDir, "ca.pem"), "utf8")).toBe("keep");
|
||||
|
||||
const state = openOpenClawStateDatabase({
|
||||
env: { OPENCLAW_STATE_DIR: root } as NodeJS.ProcessEnv,
|
||||
});
|
||||
expect(
|
||||
state.db.prepare("SELECT id, mode FROM capture_sessions WHERE id = ?").get("legacy-session"),
|
||||
).toEqual({ id: "legacy-session", mode: "proxy-run" });
|
||||
expect(
|
||||
state.db
|
||||
.prepare("SELECT flow_id, data_blob_id FROM capture_events WHERE session_id = ?")
|
||||
.get("legacy-session"),
|
||||
).toEqual({ flow_id: "legacy-flow", data_blob_id: blobId });
|
||||
const blob = state.db
|
||||
.prepare("SELECT data FROM capture_blobs WHERE blob_id = ?")
|
||||
.get(blobId) as { data?: Uint8Array } | undefined;
|
||||
expect(gunzipSync(Buffer.from(blob?.data ?? [])).toString("utf8")).toBe('{"legacy":true}');
|
||||
});
|
||||
|
||||
it("imports debug proxy capture storage from shipped environment overrides", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const sourcePath = path.join(root, "custom-capture", "capture.sqlite");
|
||||
const blobDir = path.join(root, "custom-capture", "blobs");
|
||||
writeLegacyDebugProxyCaptureSidecar(root, { sourcePath, blobDir });
|
||||
const env = {
|
||||
OPENCLAW_STATE_DIR: root,
|
||||
OPENCLAW_DEBUG_PROXY_DB_PATH: sourcePath,
|
||||
OPENCLAW_DEBUG_PROXY_BLOB_DIR: blobDir,
|
||||
} as NodeJS.ProcessEnv;
|
||||
|
||||
const detected = await detectLegacyStateMigrations({ cfg: {}, env });
|
||||
expect(detected.debugProxyCaptureSidecar).toEqual({
|
||||
sourcePath,
|
||||
blobDir,
|
||||
hasLegacy: true,
|
||||
});
|
||||
|
||||
const result = await runLegacyStateMigrations({ detected });
|
||||
|
||||
expect(result.warnings).toStrictEqual([]);
|
||||
expect(fs.existsSync(`${sourcePath}.migrated`)).toBe(true);
|
||||
expect(fs.existsSync(`${blobDir}.migrated`)).toBe(true);
|
||||
});
|
||||
|
||||
it("ignores a legacy debug proxy override that points at shared state", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const sharedStatePath = path.join(root, "state", "openclaw.sqlite");
|
||||
const state = openOpenClawStateDatabase({
|
||||
env: { OPENCLAW_STATE_DIR: root } as NodeJS.ProcessEnv,
|
||||
});
|
||||
state.db
|
||||
.prepare(
|
||||
`INSERT INTO capture_sessions (
|
||||
id, started_at, mode, source_scope, source_process
|
||||
) VALUES (?, ?, ?, ?, ?)`,
|
||||
)
|
||||
.run("shared-session", 100, "proxy-run", "openclaw", "openclaw");
|
||||
const detected = await detectLegacyStateMigrations({
|
||||
cfg: {},
|
||||
env: {
|
||||
OPENCLAW_STATE_DIR: root,
|
||||
OPENCLAW_DEBUG_PROXY_DB_PATH: sharedStatePath,
|
||||
} as NodeJS.ProcessEnv,
|
||||
});
|
||||
|
||||
expect(detected.debugProxyCaptureSidecar).toEqual({
|
||||
sourcePath: sharedStatePath,
|
||||
blobDir: path.join(root, "debug-proxy", "blobs"),
|
||||
hasLegacy: false,
|
||||
});
|
||||
const result = await runLegacyStateMigrations({ detected });
|
||||
|
||||
expect(result.warnings).toStrictEqual([]);
|
||||
expect(fs.existsSync(sharedStatePath)).toBe(true);
|
||||
expect(fs.existsSync(`${sharedStatePath}.migrated`)).toBe(false);
|
||||
expect(
|
||||
state.db.prepare("SELECT id FROM capture_sessions WHERE id = ?").get("shared-session"),
|
||||
).toEqual({ id: "shared-session" });
|
||||
});
|
||||
|
||||
it("preserves duplicate debug proxy events and retry idempotency", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const { sourcePath } = writeLegacyDebugProxyCaptureSidecar(root);
|
||||
const sqlite = requireNodeSqlite();
|
||||
const legacyDb = new sqlite.DatabaseSync(sourcePath);
|
||||
try {
|
||||
legacyDb.exec(`
|
||||
INSERT INTO capture_events (
|
||||
session_id, ts, source_scope, source_process, protocol, direction, kind, flow_id,
|
||||
method, host, path, status, close_code, content_type, headers_json, data_text,
|
||||
data_blob_id, data_sha256, error_text, meta_json
|
||||
)
|
||||
SELECT
|
||||
session_id, ts, source_scope, source_process, protocol, direction, kind, flow_id,
|
||||
method, host, path, status, close_code, content_type, headers_json, data_text,
|
||||
data_blob_id, data_sha256, error_text, meta_json
|
||||
FROM capture_events
|
||||
LIMIT 1;
|
||||
`);
|
||||
} finally {
|
||||
legacyDb.close();
|
||||
}
|
||||
const rename = failRenameOnce(sourcePath);
|
||||
const firstResult = await (async () => {
|
||||
try {
|
||||
return await runLegacyStateMigrationsForRoot(root);
|
||||
} finally {
|
||||
rename.mockRestore();
|
||||
}
|
||||
})();
|
||||
|
||||
expect(firstResult.warnings).toStrictEqual([
|
||||
`Failed archiving debug proxy capture sidecar ${sourcePath}: Error: forced archive failure`,
|
||||
]);
|
||||
expect(fs.existsSync(sourcePath)).toBe(true);
|
||||
const retryResult = await runLegacyStateMigrationsForRoot(root);
|
||||
|
||||
expect(retryResult.warnings).toStrictEqual([]);
|
||||
const state = openOpenClawStateDatabase({
|
||||
env: { OPENCLAW_STATE_DIR: root } as NodeJS.ProcessEnv,
|
||||
});
|
||||
expect(state.db.prepare("SELECT COUNT(*) AS count FROM capture_events").get()).toEqual({
|
||||
count: 2,
|
||||
});
|
||||
});
|
||||
|
||||
it("leaves debug proxy sources in place when a session id conflicts", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const { sourcePath, blobDir } = writeLegacyDebugProxyCaptureSidecar(root);
|
||||
const state = openOpenClawStateDatabase({
|
||||
env: { OPENCLAW_STATE_DIR: root } as NodeJS.ProcessEnv,
|
||||
});
|
||||
state.db
|
||||
.prepare(
|
||||
`INSERT INTO capture_sessions (
|
||||
id, started_at, mode, source_scope, source_process
|
||||
) VALUES (?, ?, ?, ?, ?)`,
|
||||
)
|
||||
.run("legacy-session", 999, "different", "openclaw", "openclaw");
|
||||
|
||||
const result = await runLegacyStateMigrationsForRoot(root);
|
||||
|
||||
expect(result.warnings).toStrictEqual([
|
||||
`Failed migrating debug proxy capture sidecar ${sourcePath}: session legacy-session already exists with different data`,
|
||||
]);
|
||||
expect(fs.existsSync(sourcePath)).toBe(true);
|
||||
expect(fs.existsSync(blobDir)).toBe(true);
|
||||
expect(state.db.prepare("SELECT COUNT(*) AS count FROM capture_events").get()).toEqual({
|
||||
count: 0,
|
||||
});
|
||||
});
|
||||
|
||||
it("retries debug proxy blob archival without duplicating imported events", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const { sourcePath, blobDir } = writeLegacyDebugProxyCaptureSidecar(root);
|
||||
const rename = failRenameOnce(blobDir);
|
||||
const firstResult = await (async () => {
|
||||
try {
|
||||
return await runLegacyStateMigrationsForRoot(root);
|
||||
} finally {
|
||||
rename.mockRestore();
|
||||
}
|
||||
})();
|
||||
|
||||
expect(firstResult.warnings).toStrictEqual([
|
||||
`Failed archiving debug proxy capture blobs ${blobDir}: Error: forced archive failure`,
|
||||
]);
|
||||
expect(fs.existsSync(`${sourcePath}.migrated`)).toBe(true);
|
||||
expect(fs.existsSync(blobDir)).toBe(true);
|
||||
|
||||
const retryDetected = await detectLegacyStateMigrations({
|
||||
cfg: {},
|
||||
env: { OPENCLAW_STATE_DIR: root } as NodeJS.ProcessEnv,
|
||||
});
|
||||
expect(retryDetected.debugProxyCaptureSidecar.hasLegacy).toBe(true);
|
||||
const retryResult = await runLegacyStateMigrations({ detected: retryDetected });
|
||||
|
||||
expect(retryResult.warnings).toStrictEqual([]);
|
||||
expect(retryResult.changes).toStrictEqual([
|
||||
`Archived debug proxy capture blobs → ${blobDir}.migrated`,
|
||||
]);
|
||||
const state = openOpenClawStateDatabase({
|
||||
env: { OPENCLAW_STATE_DIR: root } as NodeJS.ProcessEnv,
|
||||
});
|
||||
expect(state.db.prepare("SELECT COUNT(*) AS count FROM capture_events").get()).toEqual({
|
||||
count: 1,
|
||||
});
|
||||
});
|
||||
|
||||
it("archives the plugin-state rollback journal with the legacy database", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const sourcePath = writeLegacyPluginStateSidecar(root);
|
||||
|
|
|
|||
524
src/infra/state-migrations.debug-proxy.ts
Normal file
524
src/infra/state-migrations.debug-proxy.ts
Normal file
|
|
@ -0,0 +1,524 @@
|
|||
// Debug proxy state migration imports the shipped capture sidecar into shared SQLite state.
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import type { DatabaseSync, SQLInputValue } from "node:sqlite";
|
||||
import { gunzipSync } from "node:zlib";
|
||||
import { runOpenClawStateWriteTransaction } from "../state/openclaw-state-db.js";
|
||||
import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js";
|
||||
import { requireNodeSqlite } from "./node-sqlite.js";
|
||||
|
||||
const DEBUG_PROXY_SQLITE_SIDECAR_SUFFIXES = ["", "-shm", "-wal", "-journal"] as const;
|
||||
|
||||
export type LegacyDebugProxyCaptureDetection = {
|
||||
sourcePath: string;
|
||||
blobDir: string;
|
||||
hasLegacy: boolean;
|
||||
};
|
||||
|
||||
type LegacyCaptureSessionRow = {
|
||||
id: string;
|
||||
started_at: number | bigint;
|
||||
ended_at: number | bigint | null;
|
||||
mode: string;
|
||||
source_scope: string;
|
||||
source_process: string;
|
||||
proxy_url: string | null;
|
||||
};
|
||||
|
||||
type LegacyCaptureEventRow = {
|
||||
session_id: string;
|
||||
ts: number | bigint;
|
||||
source_scope: string;
|
||||
source_process: string;
|
||||
protocol: string;
|
||||
direction: string;
|
||||
kind: string;
|
||||
flow_id: string;
|
||||
method: string | null;
|
||||
host: string | null;
|
||||
path: string | null;
|
||||
status: number | bigint | null;
|
||||
close_code: number | bigint | null;
|
||||
content_type: string | null;
|
||||
headers_json: string | null;
|
||||
data_text: string | null;
|
||||
data_blob_id: string | null;
|
||||
data_sha256: string | null;
|
||||
error_text: string | null;
|
||||
meta_json: string | null;
|
||||
};
|
||||
|
||||
type LegacyCaptureBlobRow = {
|
||||
blobId: string;
|
||||
contentType: string | null;
|
||||
encoding: "gzip";
|
||||
sizeBytes: number;
|
||||
sha256: string;
|
||||
data: Buffer;
|
||||
createdAt: number;
|
||||
};
|
||||
|
||||
class LegacyDebugProxyBlobConflictError extends Error {
|
||||
constructor(readonly blobId: string) {
|
||||
super(`legacy debug proxy blob conflicts with shared state: ${blobId}`);
|
||||
}
|
||||
}
|
||||
|
||||
class LegacyDebugProxySessionConflictError extends Error {
|
||||
constructor(readonly sessionId: string) {
|
||||
super(`legacy debug proxy session conflicts with shared state: ${sessionId}`);
|
||||
}
|
||||
}
|
||||
|
||||
function fileExists(filePath: string): boolean {
|
||||
try {
|
||||
return fs.statSync(filePath).isFile();
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function dirExists(dirPath: string): boolean {
|
||||
try {
|
||||
return fs.statSync(dirPath).isDirectory();
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function resolveLegacyDebugProxyCapturePaths(
|
||||
stateDir: string,
|
||||
env: NodeJS.ProcessEnv,
|
||||
): {
|
||||
sourcePath: string;
|
||||
blobDir: string;
|
||||
} {
|
||||
const rootDir = path.join(stateDir, "debug-proxy");
|
||||
return {
|
||||
sourcePath: env.OPENCLAW_DEBUG_PROXY_DB_PATH?.trim() || path.join(rootDir, "capture.sqlite"),
|
||||
blobDir: env.OPENCLAW_DEBUG_PROXY_BLOB_DIR?.trim() || path.join(rootDir, "blobs"),
|
||||
};
|
||||
}
|
||||
|
||||
function hasPendingSqliteArchive(sourcePath: string): boolean {
|
||||
return (
|
||||
!fileExists(sourcePath) &&
|
||||
fileExists(`${sourcePath}.migrated`) &&
|
||||
DEBUG_PROXY_SQLITE_SIDECAR_SUFFIXES.some(
|
||||
(suffix) => suffix !== "" && fileExists(`${sourcePath}${suffix}`),
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
export function detectLegacyDebugProxyCaptureSidecar(
|
||||
stateDir: string,
|
||||
env: NodeJS.ProcessEnv = process.env,
|
||||
): LegacyDebugProxyCaptureDetection {
|
||||
const paths = resolveLegacyDebugProxyCapturePaths(stateDir, env);
|
||||
if (
|
||||
path.resolve(paths.sourcePath) ===
|
||||
path.resolve(resolveOpenClawStateSqlitePath({ ...env, OPENCLAW_STATE_DIR: stateDir }))
|
||||
) {
|
||||
return { ...paths, hasLegacy: false };
|
||||
}
|
||||
const hasArchivedDatabase = fileExists(`${paths.sourcePath}.migrated`);
|
||||
return {
|
||||
...paths,
|
||||
hasLegacy:
|
||||
fileExists(paths.sourcePath) ||
|
||||
hasPendingSqliteArchive(paths.sourcePath) ||
|
||||
(hasArchivedDatabase && dirExists(paths.blobDir)),
|
||||
};
|
||||
}
|
||||
|
||||
function listSqliteColumns(db: DatabaseSync, table: string): Set<string> {
|
||||
const rows = db.prepare(`PRAGMA table_info(${table})`).all() as Array<{ name?: unknown }>;
|
||||
return new Set(rows.flatMap((row) => (typeof row.name === "string" ? [row.name] : [])));
|
||||
}
|
||||
|
||||
function assertTableColumns(db: DatabaseSync, table: string, expected: readonly string[]): void {
|
||||
const columns = listSqliteColumns(db, table);
|
||||
const missing = expected.filter((column) => !columns.has(column));
|
||||
if (missing.length > 0) {
|
||||
throw new Error(`legacy ${table} table is missing ${missing.join(", ")}`);
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeSqliteInteger(value: number | bigint | null): number | null {
|
||||
return typeof value === "bigint" ? Number(value) : value;
|
||||
}
|
||||
|
||||
function readLegacyDebugProxyCapture(params: { sourcePath: string; blobDir: string }): {
|
||||
sessions: LegacyCaptureSessionRow[];
|
||||
events: LegacyCaptureEventRow[];
|
||||
blobs: LegacyCaptureBlobRow[];
|
||||
} {
|
||||
const sqlite = requireNodeSqlite();
|
||||
const db = new sqlite.DatabaseSync(params.sourcePath, { readOnly: true });
|
||||
try {
|
||||
assertTableColumns(db, "capture_sessions", [
|
||||
"id",
|
||||
"started_at",
|
||||
"ended_at",
|
||||
"mode",
|
||||
"source_scope",
|
||||
"source_process",
|
||||
"proxy_url",
|
||||
"db_path",
|
||||
"blob_dir",
|
||||
]);
|
||||
assertTableColumns(db, "capture_events", [
|
||||
"session_id",
|
||||
"ts",
|
||||
"source_scope",
|
||||
"source_process",
|
||||
"protocol",
|
||||
"direction",
|
||||
"kind",
|
||||
"flow_id",
|
||||
"method",
|
||||
"host",
|
||||
"path",
|
||||
"status",
|
||||
"close_code",
|
||||
"content_type",
|
||||
"headers_json",
|
||||
"data_text",
|
||||
"data_blob_id",
|
||||
"data_sha256",
|
||||
"error_text",
|
||||
"meta_json",
|
||||
]);
|
||||
const sessions = db
|
||||
.prepare(
|
||||
`SELECT id, started_at, ended_at, mode, source_scope, source_process, proxy_url
|
||||
FROM capture_sessions
|
||||
ORDER BY started_at ASC, id ASC`,
|
||||
)
|
||||
.all() as LegacyCaptureSessionRow[];
|
||||
const events = db
|
||||
.prepare(
|
||||
`SELECT
|
||||
session_id, ts, source_scope, source_process, protocol, direction, kind, flow_id,
|
||||
method, host, path, status, close_code, content_type, headers_json, data_text,
|
||||
data_blob_id, data_sha256, error_text, meta_json
|
||||
FROM capture_events
|
||||
ORDER BY ts ASC, id ASC`,
|
||||
)
|
||||
.all() as LegacyCaptureEventRow[];
|
||||
const sessionIds = new Set(sessions.map((session) => session.id));
|
||||
for (const event of events) {
|
||||
if (sessionIds.has(event.session_id)) {
|
||||
continue;
|
||||
}
|
||||
sessions.push({
|
||||
id: event.session_id,
|
||||
started_at: event.ts,
|
||||
ended_at: null,
|
||||
mode: "implicit",
|
||||
source_scope: event.source_scope,
|
||||
source_process: event.source_process,
|
||||
proxy_url: null,
|
||||
});
|
||||
sessionIds.add(event.session_id);
|
||||
}
|
||||
|
||||
const blobEvents = new Map<string, LegacyCaptureEventRow[]>();
|
||||
for (const event of events) {
|
||||
if (!event.data_blob_id) {
|
||||
continue;
|
||||
}
|
||||
const rows = blobEvents.get(event.data_blob_id) ?? [];
|
||||
rows.push(event);
|
||||
blobEvents.set(event.data_blob_id, rows);
|
||||
}
|
||||
const blobs: LegacyCaptureBlobRow[] = [];
|
||||
for (const [blobId, referencingEvents] of blobEvents) {
|
||||
const blobPath = path.join(params.blobDir, `${blobId}.bin.gz`);
|
||||
const data = fs.readFileSync(blobPath);
|
||||
const raw = gunzipSync(data);
|
||||
const sha256 = createHash("sha256").update(raw).digest("hex");
|
||||
if (sha256.slice(0, 24) !== blobId) {
|
||||
throw new Error(`legacy debug proxy blob hash mismatch: ${blobPath}`);
|
||||
}
|
||||
blobs.push({
|
||||
blobId,
|
||||
contentType: referencingEvents.find((event) => event.content_type)?.content_type ?? null,
|
||||
encoding: "gzip",
|
||||
sizeBytes: raw.byteLength,
|
||||
sha256,
|
||||
data,
|
||||
createdAt: Math.min(
|
||||
...referencingEvents.map((event) => normalizeSqliteInteger(event.ts) ?? 0),
|
||||
),
|
||||
});
|
||||
}
|
||||
return { sessions, events, blobs };
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
}
|
||||
|
||||
function eventValues(event: LegacyCaptureEventRow): SQLInputValue[] {
|
||||
return [
|
||||
event.session_id,
|
||||
normalizeSqliteInteger(event.ts),
|
||||
event.source_scope,
|
||||
event.source_process,
|
||||
event.protocol,
|
||||
event.direction,
|
||||
event.kind,
|
||||
event.flow_id,
|
||||
event.method,
|
||||
event.host,
|
||||
event.path,
|
||||
normalizeSqliteInteger(event.status),
|
||||
normalizeSqliteInteger(event.close_code),
|
||||
event.content_type,
|
||||
event.headers_json,
|
||||
event.data_text,
|
||||
event.data_blob_id,
|
||||
event.data_sha256,
|
||||
event.error_text,
|
||||
event.meta_json,
|
||||
];
|
||||
}
|
||||
|
||||
function eventKey(values: SQLInputValue[]): string {
|
||||
return JSON.stringify(values);
|
||||
}
|
||||
|
||||
function archiveLegacyDebugProxySqlite(params: {
|
||||
sourcePath: string;
|
||||
changes: string[];
|
||||
warnings: string[];
|
||||
}): void {
|
||||
const existingSources = DEBUG_PROXY_SQLITE_SIDECAR_SUFFIXES.map(
|
||||
(suffix) => `${params.sourcePath}${suffix}`,
|
||||
).filter(fileExists);
|
||||
if (existingSources.length === 0) {
|
||||
return;
|
||||
}
|
||||
const existingArchives = existingSources
|
||||
.map((sourcePath) => `${sourcePath}.migrated`)
|
||||
.filter(fileExists);
|
||||
if (existingArchives.length > 0) {
|
||||
params.warnings.push(
|
||||
`Left migrated debug proxy capture sidecar in place because archive already exists: ${existingArchives[0]}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
for (const sourcePath of existingSources) {
|
||||
try {
|
||||
fs.renameSync(sourcePath, `${sourcePath}.migrated`);
|
||||
} catch (err) {
|
||||
params.warnings.push(
|
||||
`Failed archiving debug proxy capture sidecar ${sourcePath}: ${String(err)}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
params.changes.push(
|
||||
`Archived debug proxy capture sidecar legacy source → ${params.sourcePath}.migrated`,
|
||||
);
|
||||
}
|
||||
|
||||
function archiveLegacyDebugProxyBlobs(params: {
|
||||
blobDir: string;
|
||||
changes: string[];
|
||||
warnings: string[];
|
||||
}): void {
|
||||
if (!dirExists(params.blobDir)) {
|
||||
return;
|
||||
}
|
||||
const archivePath = `${params.blobDir}.migrated`;
|
||||
if (dirExists(archivePath)) {
|
||||
params.warnings.push(
|
||||
`Left migrated debug proxy capture blobs in place because archive already exists: ${archivePath}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
fs.renameSync(params.blobDir, archivePath);
|
||||
params.changes.push(`Archived debug proxy capture blobs → ${archivePath}`);
|
||||
} catch (err) {
|
||||
params.warnings.push(
|
||||
`Failed archiving debug proxy capture blobs ${params.blobDir}: ${String(err)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
export function migrateLegacyDebugProxyCaptureSidecar(params: {
|
||||
stateDir: string;
|
||||
detected?: LegacyDebugProxyCaptureDetection;
|
||||
}): { changes: string[]; warnings: string[] } {
|
||||
const detected = params.detected ?? detectLegacyDebugProxyCaptureSidecar(params.stateDir);
|
||||
const changes: string[] = [];
|
||||
const warnings: string[] = [];
|
||||
if (!detected.hasLegacy) {
|
||||
return { changes, warnings };
|
||||
}
|
||||
if (!fileExists(detected.sourcePath)) {
|
||||
archiveLegacyDebugProxySqlite({ sourcePath: detected.sourcePath, changes, warnings });
|
||||
if (fileExists(`${detected.sourcePath}.migrated`)) {
|
||||
archiveLegacyDebugProxyBlobs({ blobDir: detected.blobDir, changes, warnings });
|
||||
}
|
||||
return { changes, warnings };
|
||||
}
|
||||
|
||||
let legacy: ReturnType<typeof readLegacyDebugProxyCapture>;
|
||||
try {
|
||||
legacy = readLegacyDebugProxyCapture(detected);
|
||||
} catch (err) {
|
||||
return {
|
||||
changes,
|
||||
warnings: [
|
||||
`Failed reading debug proxy capture sidecar ${detected.sourcePath}: ${String(err)}`,
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
try {
|
||||
runOpenClawStateWriteTransaction(
|
||||
({ db }) => {
|
||||
const selectBlob = db.prepare(
|
||||
`SELECT encoding, size_bytes AS sizeBytes, sha256, data
|
||||
FROM capture_blobs
|
||||
WHERE blob_id = ?`,
|
||||
);
|
||||
const insertBlob = db.prepare(
|
||||
`INSERT INTO capture_blobs (
|
||||
blob_id, content_type, encoding, size_bytes, sha256, data, created_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?)`,
|
||||
);
|
||||
for (const blob of legacy.blobs) {
|
||||
const existing = selectBlob.get(blob.blobId) as
|
||||
| { encoding?: unknown; sizeBytes?: unknown; sha256?: unknown; data?: Uint8Array }
|
||||
| undefined;
|
||||
if (existing) {
|
||||
if (
|
||||
existing.encoding !== blob.encoding ||
|
||||
Number(existing.sizeBytes) !== blob.sizeBytes ||
|
||||
existing.sha256 !== blob.sha256 ||
|
||||
!existing.data ||
|
||||
!Buffer.from(existing.data).equals(blob.data)
|
||||
) {
|
||||
throw new LegacyDebugProxyBlobConflictError(blob.blobId);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
insertBlob.run(
|
||||
blob.blobId,
|
||||
blob.contentType,
|
||||
blob.encoding,
|
||||
blob.sizeBytes,
|
||||
blob.sha256,
|
||||
blob.data,
|
||||
blob.createdAt,
|
||||
);
|
||||
}
|
||||
|
||||
const selectSession = db.prepare(
|
||||
`SELECT
|
||||
started_at AS startedAt,
|
||||
ended_at AS endedAt,
|
||||
mode,
|
||||
source_scope AS sourceScope,
|
||||
source_process AS sourceProcess,
|
||||
proxy_url AS proxyUrl
|
||||
FROM capture_sessions
|
||||
WHERE id = ?`,
|
||||
);
|
||||
const insertSession = db.prepare(
|
||||
`INSERT INTO capture_sessions (
|
||||
id, started_at, ended_at, mode, source_scope, source_process, proxy_url
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?)`,
|
||||
);
|
||||
for (const session of legacy.sessions) {
|
||||
const values = [
|
||||
session.id,
|
||||
normalizeSqliteInteger(session.started_at),
|
||||
normalizeSqliteInteger(session.ended_at),
|
||||
session.mode,
|
||||
session.source_scope,
|
||||
session.source_process,
|
||||
session.proxy_url,
|
||||
] satisfies SQLInputValue[];
|
||||
const existing = selectSession.get(session.id);
|
||||
if (existing) {
|
||||
const expected = {
|
||||
startedAt: values[1],
|
||||
endedAt: values[2],
|
||||
mode: values[3],
|
||||
sourceScope: values[4],
|
||||
sourceProcess: values[5],
|
||||
proxyUrl: values[6],
|
||||
};
|
||||
if (JSON.stringify(existing) !== JSON.stringify(expected)) {
|
||||
throw new LegacyDebugProxySessionConflictError(session.id);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
insertSession.run(...values);
|
||||
}
|
||||
|
||||
const existingEventCount = db.prepare(
|
||||
`SELECT COUNT(*) AS count
|
||||
FROM capture_events
|
||||
WHERE session_id IS ? AND ts IS ? AND source_scope IS ? AND source_process IS ?
|
||||
AND protocol IS ? AND direction IS ? AND kind IS ? AND flow_id IS ?
|
||||
AND method IS ? AND host IS ? AND path IS ? AND status IS ? AND close_code IS ?
|
||||
AND content_type IS ? AND headers_json IS ? AND data_text IS ? AND data_blob_id IS ?
|
||||
AND data_sha256 IS ? AND error_text IS ? AND meta_json IS ?
|
||||
`,
|
||||
);
|
||||
const insertEvent = db.prepare(
|
||||
`INSERT INTO capture_events (
|
||||
session_id, ts, source_scope, source_process, protocol, direction, kind, flow_id,
|
||||
method, host, path, status, close_code, content_type, headers_json, data_text,
|
||||
data_blob_id, data_sha256, error_text, meta_json
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
);
|
||||
const existingCounts = new Map<string, number>();
|
||||
const seenCounts = new Map<string, number>();
|
||||
for (const event of legacy.events) {
|
||||
const values = eventValues(event);
|
||||
const key = eventKey(values);
|
||||
const seenCount = (seenCounts.get(key) ?? 0) + 1;
|
||||
seenCounts.set(key, seenCount);
|
||||
let existingCount = existingCounts.get(key);
|
||||
if (existingCount === undefined) {
|
||||
const row = existingEventCount.get(...values) as { count?: unknown } | undefined;
|
||||
existingCount = Number(row?.count ?? 0);
|
||||
existingCounts.set(key, existingCount);
|
||||
}
|
||||
if (seenCount > existingCount) {
|
||||
insertEvent.run(...values);
|
||||
}
|
||||
}
|
||||
},
|
||||
{ env: { ...process.env, OPENCLAW_STATE_DIR: params.stateDir } },
|
||||
);
|
||||
changes.push(
|
||||
`Migrated ${legacy.sessions.length} debug proxy capture ${legacy.sessions.length === 1 ? "session" : "sessions"}, ${legacy.events.length} ${legacy.events.length === 1 ? "event" : "events"}, and ${legacy.blobs.length} ${legacy.blobs.length === 1 ? "blob" : "blobs"} → shared SQLite state`,
|
||||
);
|
||||
} catch (err) {
|
||||
const detail =
|
||||
err instanceof LegacyDebugProxyBlobConflictError
|
||||
? `blob ${err.blobId} already exists with different data`
|
||||
: err instanceof LegacyDebugProxySessionConflictError
|
||||
? `session ${err.sessionId} already exists with different data`
|
||||
: String(err);
|
||||
return {
|
||||
changes,
|
||||
warnings: [`Failed migrating debug proxy capture sidecar ${detected.sourcePath}: ${detail}`],
|
||||
};
|
||||
}
|
||||
|
||||
archiveLegacyDebugProxySqlite({ sourcePath: detected.sourcePath, changes, warnings });
|
||||
if (!fileExists(detected.sourcePath) && fileExists(`${detected.sourcePath}.migrated`)) {
|
||||
archiveLegacyDebugProxyBlobs({ blobDir: detected.blobDir, changes, warnings });
|
||||
}
|
||||
return { changes, warnings };
|
||||
}
|
||||
|
|
@ -71,6 +71,10 @@ import {
|
|||
import { requireNodeSqlite } from "./node-sqlite.js";
|
||||
import { parseRegistryNpmSpec } from "./npm-registry-spec.js";
|
||||
import { isWithinDir } from "./path-safety.js";
|
||||
import {
|
||||
detectLegacyDebugProxyCaptureSidecar,
|
||||
migrateLegacyDebugProxyCaptureSidecar,
|
||||
} from "./state-migrations.debug-proxy.js";
|
||||
import {
|
||||
ensureDir,
|
||||
existsDir,
|
||||
|
|
@ -116,6 +120,11 @@ export type LegacyStateDetection = {
|
|||
sourcePath: string;
|
||||
hasLegacy: boolean;
|
||||
};
|
||||
debugProxyCaptureSidecar: {
|
||||
sourcePath: string;
|
||||
blobDir: string;
|
||||
hasLegacy: boolean;
|
||||
};
|
||||
stateSchema: {
|
||||
hasLegacy: boolean;
|
||||
preview: string[];
|
||||
|
|
@ -2877,6 +2886,7 @@ export async function detectLegacyStateMigrations(params: {
|
|||
);
|
||||
const pluginInstallIndexPath = resolveLegacyInstalledPluginIndexStorePath({ stateDir });
|
||||
const hasPluginInstallIndex = fileExists(pluginInstallIndexPath);
|
||||
const debugProxyCaptureSidecar = detectLegacyDebugProxyCaptureSidecar(stateDir, env);
|
||||
const stateSchemaMigrations = detectOpenClawStateDatabaseSchemaMigrations({
|
||||
env: { ...env, OPENCLAW_STATE_DIR: stateDir },
|
||||
});
|
||||
|
|
@ -2941,6 +2951,11 @@ export async function detectLegacyStateMigrations(params: {
|
|||
if (hasPluginInstallIndex) {
|
||||
preview.push(`- Plugin install index: ${pluginInstallIndexPath} → shared SQLite state`);
|
||||
}
|
||||
if (debugProxyCaptureSidecar.hasLegacy) {
|
||||
preview.push(
|
||||
`- Debug proxy capture sidecar: ${debugProxyCaptureSidecar.sourcePath} → shared SQLite state`,
|
||||
);
|
||||
}
|
||||
if (stateSchemaMigrations.length > 0) {
|
||||
preview.push("- Shared SQLite schema: agent database registry primary key → agent_id,path");
|
||||
preview.push(
|
||||
|
|
@ -3005,6 +3020,7 @@ export async function detectLegacyStateMigrations(params: {
|
|||
sourcePath: pluginInstallIndexPath,
|
||||
hasLegacy: hasPluginInstallIndex,
|
||||
},
|
||||
debugProxyCaptureSidecar,
|
||||
stateSchema: {
|
||||
hasLegacy: stateSchemaMigrations.length > 0,
|
||||
preview: stateSchemaMigrations.map((migration) => migration.path),
|
||||
|
|
@ -3526,6 +3542,10 @@ export async function runLegacyStateMigrations(params: {
|
|||
const pluginInstallIndex = await migrateLegacyInstalledPluginIndex({
|
||||
stateDir: detected.stateDir,
|
||||
});
|
||||
const debugProxyCaptureSidecar = migrateLegacyDebugProxyCaptureSidecar({
|
||||
stateDir: detected.stateDir,
|
||||
detected: detected.debugProxyCaptureSidecar,
|
||||
});
|
||||
const taskStateSidecars = await migrateLegacyTaskStateSidecars({
|
||||
stateDir: detected.stateDir,
|
||||
});
|
||||
|
|
@ -3559,6 +3579,7 @@ export async function runLegacyStateMigrations(params: {
|
|||
...stateSchema.changes,
|
||||
...pluginStateSidecar.changes,
|
||||
...pluginInstallIndex.changes,
|
||||
...debugProxyCaptureSidecar.changes,
|
||||
...taskStateSidecars.changes,
|
||||
...deliveryQueues.changes,
|
||||
...execApprovals.changes,
|
||||
|
|
@ -3573,6 +3594,7 @@ export async function runLegacyStateMigrations(params: {
|
|||
...stateSchema.warnings,
|
||||
...pluginStateSidecar.warnings,
|
||||
...pluginInstallIndex.warnings,
|
||||
...debugProxyCaptureSidecar.warnings,
|
||||
...taskStateSidecars.warnings,
|
||||
...deliveryQueues.warnings,
|
||||
...execApprovals.warnings,
|
||||
|
|
@ -3886,6 +3908,10 @@ export async function autoMigrateLegacyState(params: {
|
|||
const pluginInstallIndex = await migrateLegacyInstalledPluginIndex({
|
||||
stateDir: detected.stateDir,
|
||||
});
|
||||
const debugProxyCaptureSidecar = migrateLegacyDebugProxyCaptureSidecar({
|
||||
stateDir: detected.stateDir,
|
||||
detected: detected.debugProxyCaptureSidecar,
|
||||
});
|
||||
const taskStateSidecars = await migrateLegacyTaskStateSidecars({
|
||||
stateDir: detected.stateDir,
|
||||
});
|
||||
|
|
@ -3907,6 +3933,7 @@ export async function autoMigrateLegacyState(params: {
|
|||
...acpSessionMetadata.changes,
|
||||
...pluginStateSidecar.changes,
|
||||
...pluginInstallIndex.changes,
|
||||
...debugProxyCaptureSidecar.changes,
|
||||
...taskStateSidecars.changes,
|
||||
...deliveryQueues.changes,
|
||||
...execApprovals.changes,
|
||||
|
|
@ -3920,6 +3947,7 @@ export async function autoMigrateLegacyState(params: {
|
|||
...acpSessionMetadata.warnings,
|
||||
...pluginStateSidecar.warnings,
|
||||
...pluginInstallIndex.warnings,
|
||||
...debugProxyCaptureSidecar.warnings,
|
||||
...taskStateSidecars.warnings,
|
||||
...deliveryQueues.warnings,
|
||||
...execApprovals.warnings,
|
||||
|
|
@ -3935,6 +3963,7 @@ export async function autoMigrateLegacyState(params: {
|
|||
acpSessionMetadata.changes.length > 0 ||
|
||||
pluginStateSidecar.changes.length > 0 ||
|
||||
pluginInstallIndex.changes.length > 0 ||
|
||||
debugProxyCaptureSidecar.changes.length > 0 ||
|
||||
taskStateSidecars.changes.length > 0 ||
|
||||
deliveryQueues.changes.length > 0 ||
|
||||
execApprovals.changes.length > 0 ||
|
||||
|
|
@ -3952,6 +3981,7 @@ export async function autoMigrateLegacyState(params: {
|
|||
!detected.pluginPlans?.hasLegacy &&
|
||||
!detected.pluginStateSidecar.hasLegacy &&
|
||||
!detected.pluginInstallIndex.hasLegacy &&
|
||||
!detected.debugProxyCaptureSidecar.hasLegacy &&
|
||||
!detected.stateSchema.hasLegacy &&
|
||||
!detected.taskStateSidecars.hasLegacy &&
|
||||
!detected.deliveryQueues.hasLegacy &&
|
||||
|
|
@ -3989,6 +4019,10 @@ export async function autoMigrateLegacyState(params: {
|
|||
const pluginInstallIndex = await migrateLegacyInstalledPluginIndex({
|
||||
stateDir: detected.stateDir,
|
||||
});
|
||||
const debugProxyCaptureSidecar = migrateLegacyDebugProxyCaptureSidecar({
|
||||
stateDir: detected.stateDir,
|
||||
detected: detected.debugProxyCaptureSidecar,
|
||||
});
|
||||
const taskStateSidecars = await migrateLegacyTaskStateSidecars({
|
||||
stateDir: detected.stateDir,
|
||||
});
|
||||
|
|
@ -4022,6 +4056,7 @@ export async function autoMigrateLegacyState(params: {
|
|||
...acpSessionMetadata.changes,
|
||||
...pluginStateSidecar.changes,
|
||||
...pluginInstallIndex.changes,
|
||||
...debugProxyCaptureSidecar.changes,
|
||||
...taskStateSidecars.changes,
|
||||
...deliveryQueues.changes,
|
||||
...execApprovals.changes,
|
||||
|
|
@ -4039,6 +4074,7 @@ export async function autoMigrateLegacyState(params: {
|
|||
...acpSessionMetadata.warnings,
|
||||
...pluginStateSidecar.warnings,
|
||||
...pluginInstallIndex.warnings,
|
||||
...debugProxyCaptureSidecar.warnings,
|
||||
...taskStateSidecars.warnings,
|
||||
...deliveryQueues.warnings,
|
||||
...execApprovals.warnings,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue