import fs from "node:fs"; import path from "node:path"; import { DatabaseSync } from "node:sqlite"; import { gzipSync } from "node:zlib"; import { afterEach, describe, expect, it } from "vitest"; import { cleanupTempDirs, makeTempDir } from "../../test/helpers/temp-dir.js"; import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js"; import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js"; import { createDebugProxyCaptureReader } from "./store-readonly.js"; const cleanupDirs: string[] = []; afterEach(() => { cleanupTempDirs(cleanupDirs); }); function makeStateEnv(prefix: string): NodeJS.ProcessEnv { return { OPENCLAW_STATE_DIR: makeTempDir(cleanupDirs, prefix) }; } function createCaptureDatabase(env: NodeJS.ProcessEnv, schemaVersion: number): string { const databasePath = resolveOpenClawStateSqlitePath(env); fs.mkdirSync(path.dirname(databasePath), { recursive: true }); const db = new DatabaseSync(databasePath); try { db.exec(` PRAGMA user_version = ${schemaVersion}; CREATE TABLE workspace_attestations ( workspace_key TEXT PRIMARY KEY, attested_at_ms INTEGER NOT NULL, updated_at_ms INTEGER NOT NULL ) STRICT; CREATE TABLE capture_sessions ( id TEXT NOT NULL 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 ) STRICT; CREATE TABLE capture_blobs ( blob_id TEXT NOT NULL PRIMARY KEY, content_type TEXT, encoding TEXT NOT NULL, size_bytes INTEGER NOT NULL, sha256 TEXT NOT NULL, data BLOB NOT NULL, created_at INTEGER NOT NULL ) STRICT; CREATE TABLE capture_events ( id INTEGER NOT NULL 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 ) STRICT; CREATE INDEX capture_events_session_ts_idx ON capture_events(session_id, ts); CREATE INDEX capture_events_flow_idx ON capture_events(flow_id, ts); INSERT INTO workspace_attestations VALUES ('workspace-1', 100, 200); INSERT INTO capture_sessions VALUES ( 'session-1', 1, NULL, 'qa', 'openclaw', 'candidate', NULL ); `); db.prepare( `INSERT INTO capture_blobs ( blob_id, content_type, encoding, size_bytes, sha256, data, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?)`, ).run( "blob-1", "application/json", "gzip", 11, "fixture-sha", gzipSync(Buffer.from('{"ok":true}')), 2, ); db.prepare( `INSERT INTO capture_events ( id, session_id, ts, source_scope, source_process, protocol, direction, kind, flow_id, method, host, path, status, content_type, data_blob_id ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ).run( 1, "session-1", 3, "openclaw", "candidate", "https", "outbound", "request", "flow-1", "POST", "slack.com", "/api/chat.postMessage", 200, "application/json", "blob-1", ); } finally { db.close(); } return databasePath; } function inspectDatabase(databasePath: string) { const db = new DatabaseSync(databasePath, { readOnly: true }); try { return { fingerprint: JSON.stringify( db .prepare( `SELECT type, name, tbl_name AS tableName, sql FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%' ORDER BY type, name`, ) .all(), ), userVersion: db.prepare("PRAGMA user_version").get(), workspaceAttestations: db .prepare("SELECT * FROM workspace_attestations ORDER BY workspace_key") .all(), }; } finally { db.close(); } } describe("createDebugProxyCaptureReader", () => { it.each([3, 4, 5, 6])( "reads schema %s without changing its version, schema, or workspace attestations", (schemaVersion) => { const env = makeStateEnv(`openclaw-proxy-reader-v${schemaVersion}-`); const databasePath = createCaptureDatabase(env, schemaVersion); const before = inspectDatabase(databasePath); const reader = createDebugProxyCaptureReader({ env }); expect(reader.getSessionEvents("session-1")).toEqual([ expect.objectContaining({ id: 1, sessionId: "session-1", flowId: "flow-1", host: "slack.com", }), ]); expect(reader.readBlob("blob-1")).toBe('{"ok":true}'); expect(inspectDatabase(databasePath)).toEqual(before); }, ); it("returns empty results for a missing database without creating state artifacts", () => { const env = makeStateEnv("openclaw-proxy-reader-missing-"); const reader = createDebugProxyCaptureReader({ env }); const databasePath = resolveOpenClawStateSqlitePath(env); expect(reader.getSessionEvents("missing")).toEqual([]); expect(reader.readBlob("missing")).toBeNull(); expect(fs.existsSync(databasePath)).toBe(false); expect(fs.existsSync(path.dirname(databasePath))).toBe(false); }); it("observes a database created after the reader", () => { const env = makeStateEnv("openclaw-proxy-reader-late-db-"); const reader = createDebugProxyCaptureReader({ env }); expect(reader.getSessionEvents("session-1")).toEqual([]); createCaptureDatabase(env, 3); expect(reader.getSessionEvents("session-1")).toHaveLength(1); expect(reader.readBlob("blob-1")).toBe('{"ok":true}'); }); it("observes later committed WAL writes on subsequent calls", () => { const env = makeStateEnv("openclaw-proxy-reader-wal-"); const databasePath = createCaptureDatabase(env, 3); const writer = new DatabaseSync(databasePath); const reader = createDebugProxyCaptureReader({ env }); try { writer.exec("PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0;"); expect(reader.getSessionEvents("session-1")).toHaveLength(1); writer .prepare( `INSERT INTO capture_events ( id, session_id, ts, source_scope, source_process, protocol, direction, kind, flow_id ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, ) .run(2, "session-1", 4, "openclaw", "candidate", "https", "inbound", "response", "flow-1"); expect(reader.getSessionEvents("session-1").map((event) => event.id)).toEqual([2, 1]); } finally { writer.close(); } }); it("projects exact camelCase keys with timestamp, id, and limit ordering", () => { const env = makeStateEnv("openclaw-proxy-reader-ordering-"); const databasePath = createCaptureDatabase(env, 3); const writer = new DatabaseSync(databasePath); try { const insert = writer.prepare( `INSERT INTO capture_events ( id, 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ); insert.run( 2, "session-1", 5, "openclaw", "candidate", "https", "outbound", "request", "flow-2", "POST", "slack.com", "/api/chat.postMessage", 200, 1000, "application/json", '{"authorization":"redacted"}', '{"text":"older"}', "blob-1", "sha-2", "error-2", '{"attempt":2}', ); insert.run( 3, "session-1", 5, "openclaw", "candidate", "https", "inbound", "response", "flow-3", "GET", "slack.com", "/api/conversations.history", 201, 1001, "application/json", '{"accept":"application/json"}', '{"text":"newer"}', "blob-1", "sha-3", "error-3", '{"attempt":3}', ); insert.run( 4, "session-1", 6, "openclaw", "candidate", "https", "inbound", "response", "flow-4", null, null, null, null, null, null, null, null, null, null, null, null, ); } finally { writer.close(); } const events = createDebugProxyCaptureReader({ env }).getSessionEvents("session-1", 2); expect(events.map((event) => event.id)).toEqual([4, 3]); expect(events[1]).toEqual({ id: 3, sessionId: "session-1", ts: 5, sourceScope: "openclaw", sourceProcess: "candidate", protocol: "https", direction: "inbound", kind: "response", flowId: "flow-3", method: "GET", host: "slack.com", path: "/api/conversations.history", status: 201, closeCode: 1001, contentType: "application/json", headersJson: '{"accept":"application/json"}', dataText: '{"text":"newer"}', dataBlobId: "blob-1", dataSha256: "sha-3", errorText: "error-3", metaJson: '{"attempt":3}', }); }); it("propagates newer schema errors", () => { const env = makeStateEnv("openclaw-proxy-reader-newer-"); const databasePath = resolveOpenClawStateSqlitePath(env); fs.mkdirSync(path.dirname(databasePath), { recursive: true }); const db = new DatabaseSync(databasePath); db.exec(`PRAGMA user_version = ${OPENCLAW_STATE_SCHEMA_VERSION + 1};`); db.close(); expect(() => createDebugProxyCaptureReader({ env }).getSessionEvents("session-1")).toThrow( `uses newer schema version ${OPENCLAW_STATE_SCHEMA_VERSION + 1}`, ); }); it("propagates database corruption", () => { const env = makeStateEnv("openclaw-proxy-reader-corrupt-"); const databasePath = resolveOpenClawStateSqlitePath(env); fs.mkdirSync(path.dirname(databasePath), { recursive: true }); fs.writeFileSync(databasePath, "not a sqlite database"); expect(() => createDebugProxyCaptureReader({ env }).readBlob("blob-1")).toThrow(); }); it("propagates capture query failures", () => { const env = makeStateEnv("openclaw-proxy-reader-sql-"); const databasePath = resolveOpenClawStateSqlitePath(env); fs.mkdirSync(path.dirname(databasePath), { recursive: true }); const db = new DatabaseSync(databasePath); db.exec("PRAGMA user_version = 3;"); db.close(); expect(() => createDebugProxyCaptureReader({ env }).getSessionEvents("session-1")).toThrow( /no such table: capture_events/u, ); }); });