From 9a4f12c94da43f8a024c3a312ff4a48c2db2f83b Mon Sep 17 00:00:00 2001 From: chhhee10 Date: Fri, 11 Sep 2026 21:22:19 +0530 Subject: [PATCH 1/3] fix(openclaw): support SQLite transcripts --- CHANGELOG.md | 8 + Cargo.lock | 6 +- Cargo.toml | 2 +- __tests__/lib/download-session.test.ts | 129 +++++- __tests__/lib/openclaw-projects.test.ts | 151 +++++- __tests__/lib/openclaw-sessions.test.ts | 140 +++++- crates/failproofaid/src/main.rs | 102 ++++- crates/fpai-collect/src/cursor.rs | 13 + crates/fpai-collect/src/filetail.rs | 2 + crates/fpai-collect/src/sources/mod.rs | 8 +- .../fpai-collect/src/sources/openclaw/mod.rs | 16 +- .../src/sources/openclaw/sqlite.rs | 433 ++++++++++++++++++ crates/fpai-collect/tests/openclaw_source.rs | 272 +++++++++++ lib/download-session.ts | 71 ++- lib/openclaw-db.ts | 271 +++++++++++ lib/openclaw-projects.ts | 72 ++- lib/openclaw-sessions.ts | 92 +++- package.json | 2 +- 18 files changed, 1692 insertions(+), 98 deletions(-) create mode 100644 crates/fpai-collect/src/sources/openclaw/sqlite.rs create mode 100644 lib/openclaw-db.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index fd447dcc9..9d655262a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## 1.0.4-beta.8 — 2026-09-11 + +### Fixes + +- Restore OpenClaw 2026.9.2+ observability after live transcripts moved from per-session JSONL files into per-agent SQLite databases. `failproofaid` now discovers every agent profile, tails each SQLite transcript incrementally, handles transcript rewrites without duplicating delivery, and retains legacy JSONL compatibility. + +- Restore OpenClaw sessions in the local dashboard's Projects view and session viewer. SQLite and legacy sessions are merged per agent, live SQLite copies win over archived duplicates, missing channels group under `local`, and downloads export the original `event_json` records as JSONL. + ## 1.0.4-beta.7 — 2026-09-11 ### Fixes diff --git a/Cargo.lock b/Cargo.lock index ffbeae800..7e5d8da4c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -202,7 +202,7 @@ dependencies = [ [[package]] name = "failproofaid" -version = "1.0.4-beta.7" +version = "1.0.4-beta.8" dependencies = [ "fpai-collect", "fpai-ipc", @@ -265,7 +265,7 @@ dependencies = [ [[package]] name = "fpai-collect" -version = "1.0.4-beta.7" +version = "1.0.4-beta.8" dependencies = [ "notify", "reqwest", @@ -280,7 +280,7 @@ dependencies = [ [[package]] name = "fpai-ipc" -version = "1.0.4-beta.7" +version = "1.0.4-beta.8" dependencies = [ "libc", "proptest", diff --git a/Cargo.toml b/Cargo.toml index 5445233aa..31df1bcb4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "3" members = ["crates/*"] [workspace.package] -version = "1.0.4-beta.7" +version = "1.0.4-beta.8" edition = "2024" license-file = "LICENSE" repository = "https://github.com/FailproofAI/failproofai" diff --git a/__tests__/lib/download-session.test.ts b/__tests__/lib/download-session.test.ts index 032ee20c1..b6b800a3e 100644 --- a/__tests__/lib/download-session.test.ts +++ b/__tests__/lib/download-session.test.ts @@ -22,7 +22,9 @@ describe("lib/download-session: isValidSessionId", () => { }); it("accepts ses_* IDs for opencode and rejects UUIDs", () => { - expect(isValidSessionId("opencode", "ses_21ad60d14ffewMeRRKMLdS7vOI")).toBe(true); + expect(isValidSessionId("opencode", "ses_21ad60d14ffewMeRRKMLdS7vOI")).toBe( + true, + ); expect(isValidSessionId("opencode", VALID_UUID)).toBe(false); expect(isValidSessionId("opencode", "ses_with-dash")).toBe(false); }); @@ -42,11 +44,17 @@ describe("lib/download-session: resolveDownloadSource", () => { vi.doUnmock("@/lib/cursor-sessions"); vi.doUnmock("@/lib/pi-sessions"); vi.doUnmock("@/lib/opencode-sessions"); + vi.doUnmock("@/lib/openclaw-sessions"); + vi.doUnmock("@/lib/openclaw-db"); }); it("throws RangeError on invalid session id", async () => { - await expect(resolveDownloadSource("codex", "proj", "garbage")).rejects.toBeInstanceOf(RangeError); - await expect(resolveDownloadSource("opencode", "proj", VALID_UUID)).rejects.toBeInstanceOf(RangeError); + await expect( + resolveDownloadSource("codex", "proj", "garbage"), + ).rejects.toBeInstanceOf(RangeError); + await expect( + resolveDownloadSource("opencode", "proj", VALID_UUID), + ).rejects.toBeInstanceOf(RangeError); }); it("Claude: resolves under the projects root via resolveSessionFilePath", async () => { @@ -72,7 +80,9 @@ describe("lib/download-session: resolveDownloadSource", () => { try { vi.resetModules(); ({ resolveDownloadSource } = await import("@/lib/download-session")); - await expect(resolveDownloadSource("claude", "../etc", VALID_UUID)).rejects.toBeInstanceOf(RangeError); + await expect( + resolveDownloadSource("claude", "../etc", VALID_UUID), + ).rejects.toBeInstanceOf(RangeError); } finally { delete process.env.CLAUDE_PROJECTS_PATH; rmSync(root, { recursive: true, force: true }); @@ -89,7 +99,8 @@ describe("lib/download-session: resolveDownloadSource", () => { async (cli, modulePath, fnName) => { vi.doMock(modulePath, () => ({ [fnName]: () => "/tmp/fake.jsonl" })); vi.resetModules(); - const { resolveDownloadSource: rds } = await import("@/lib/download-session"); + const { resolveDownloadSource: rds } = + await import("@/lib/download-session"); const result = await rds(cli, "proj", VALID_UUID); expect(result).toEqual({ kind: "file", path: "/tmp/fake.jsonl" }); }, @@ -100,25 +111,55 @@ describe("lib/download-session: resolveDownloadSource", () => { ["copilot", "@/lib/copilot-sessions", "findCopilotTranscript"], ["cursor", "@/lib/cursor-sessions", "findCursorTranscript"], ["pi", "@/lib/pi-sessions", "findPiTranscript"], - ] as const)("%s: returns null when transcript is missing", async (cli, modulePath, fnName) => { - vi.doMock(modulePath, () => ({ [fnName]: () => null })); - vi.resetModules(); - const { resolveDownloadSource: rds } = await import("@/lib/download-session"); - const result = await rds(cli, "proj", VALID_UUID); - expect(result).toBeNull(); - }); + ] as const)( + "%s: returns null when transcript is missing", + async (cli, modulePath, fnName) => { + vi.doMock(modulePath, () => ({ [fnName]: () => null })); + vi.resetModules(); + const { resolveDownloadSource: rds } = + await import("@/lib/download-session"); + const result = await rds(cli, "proj", VALID_UUID); + expect(result).toBeNull(); + }, + ); it("OpenCode: emits a JSON document mirroring the SQLite session/message/part structure", async () => { const exportPayload = { - session: { id: "ses_abc", project_id: "proj_1", slug: null, directory: "/tmp/p", title: "T", time_created: 1, time_updated: 2 }, - messages: [{ id: "m1", session_id: "ses_abc", time_created: 1, time_updated: 1, data: { role: "user" } }], - parts: [{ id: "p1", message_id: "m1", session_id: "ses_abc", time_created: 1, time_updated: 1, data: { type: "text", text: "hi" } }], + session: { + id: "ses_abc", + project_id: "proj_1", + slug: null, + directory: "/tmp/p", + title: "T", + time_created: 1, + time_updated: 2, + }, + messages: [ + { + id: "m1", + session_id: "ses_abc", + time_created: 1, + time_updated: 1, + data: { role: "user" }, + }, + ], + parts: [ + { + id: "p1", + message_id: "m1", + session_id: "ses_abc", + time_created: 1, + time_updated: 1, + data: { type: "text", text: "hi" }, + }, + ], }; vi.doMock("@/lib/opencode-sessions", () => ({ getOpenCodeSessionExport: async () => exportPayload, })); vi.resetModules(); - const { resolveDownloadSource: rds } = await import("@/lib/download-session"); + const { resolveDownloadSource: rds } = + await import("@/lib/download-session"); const result = await rds("opencode", "proj", "ses_abc123"); expect(result).toEqual({ kind: "synthesized", @@ -133,10 +174,51 @@ describe("lib/download-session: resolveDownloadSource", () => { getOpenCodeSessionExport: async () => null, })); vi.resetModules(); - const { resolveDownloadSource: rds } = await import("@/lib/download-session"); + const { resolveDownloadSource: rds } = + await import("@/lib/download-session"); const result = await rds("opencode", "proj", "ses_missing"); expect(result).toBeNull(); }); + + it("OpenClaw: synthesizes JSONL from SQLite event_json rows", async () => { + vi.doMock("@/lib/openclaw-sessions", () => ({ + findOpenClawTranscript: () => "/tmp/archived.jsonl", + listOpenClawAgents: () => ["main"], + openclawHome: () => "/tmp/openclaw", + })); + vi.doMock("@/lib/openclaw-db", () => ({ + readOpenClawSqliteTranscript: async () => ({ + eventJsonLines: ['{"type":"session"}', '{"type":"message"}'], + }), + })); + vi.resetModules(); + const { resolveDownloadSource: rds } = + await import("@/lib/download-session"); + expect(await rds("openclaw", "ignored", VALID_UUID)).toEqual({ + kind: "synthesized", + body: '{"type":"session"}\n{"type":"message"}\n', + contentType: "application/x-ndjson", + extension: "jsonl", + }); + }); + + it("OpenClaw: falls back to an archived JSONL file", async () => { + vi.doMock("@/lib/openclaw-sessions", () => ({ + findOpenClawTranscript: () => "/tmp/archived.jsonl", + listOpenClawAgents: () => ["main"], + openclawHome: () => "/tmp/openclaw", + })); + vi.doMock("@/lib/openclaw-db", () => ({ + readOpenClawSqliteTranscript: async () => null, + })); + vi.resetModules(); + const { resolveDownloadSource: rds } = + await import("@/lib/download-session"); + expect(await rds("openclaw", "ignored", VALID_UUID)).toEqual({ + kind: "file", + path: "/tmp/archived.jsonl", + }); + }); }); describe("lib/download-session: end-to-end fixture (codex)", () => { @@ -170,10 +252,17 @@ describe("lib/download-session: end-to-end fixture (codex)", () => { it("locates a codex transcript on disk and returns its path", async () => { const dir = join(tmpHome, ".codex", "sessions", "2026", "05", "05"); mkdirSync(dir, { recursive: true }); - const filePath = join(dir, `rollout-2026-05-05T00-00-00-${VALID_UUID}.jsonl`); - writeFileSync(filePath, '{"timestamp":"2026-05-05T00:00:00.000Z","type":"session_meta","payload":{}}\n'); + const filePath = join( + dir, + `rollout-2026-05-05T00-00-00-${VALID_UUID}.jsonl`, + ); + writeFileSync( + filePath, + '{"timestamp":"2026-05-05T00:00:00.000Z","type":"session_meta","payload":{}}\n', + ); - const { resolveDownloadSource: rds } = await import("@/lib/download-session"); + const { resolveDownloadSource: rds } = + await import("@/lib/download-session"); const result = await rds("codex", "ignored", VALID_UUID); expect(result).toEqual({ kind: "file", path: filePath }); }); diff --git a/__tests__/lib/openclaw-projects.test.ts b/__tests__/lib/openclaw-projects.test.ts index 0dd7af07e..dc43bdc87 100644 --- a/__tests__/lib/openclaw-projects.test.ts +++ b/__tests__/lib/openclaw-projects.test.ts @@ -23,12 +23,72 @@ const UUID_CLI = "f9e8516e-fed2-4e54-acbe-7a20aefc6cfa"; // project row when grouping was channel-only. Its id contains a hyphen, which // is why the name parser can't just split the slug. const UUID_WEATHER = "bb222222-3333-4444-5555-666666666666"; +const UUID_SQLITE = "cc333333-4444-5555-6666-777777777777"; let home: string | undefined; const prev = process.env.OPENCLAW_HOME; function writeSession(dir: string, uuid: string): void { - writeFileSync(join(dir, `${uuid}.jsonl`), JSON.stringify({ type: "session", cwd: "/x" }) + "\n"); + writeFileSync( + join(dir, `${uuid}.jsonl`), + JSON.stringify({ type: "session", cwd: "/x" }) + "\n", + ); +} + +async function writeSqliteSession( + root: string, + agentId: string, + uuid: string, + updatedAt: number, + channel: string | null = null, +): Promise { + try { + const { DatabaseSync } = (await import("node:sqlite")) as unknown as { + DatabaseSync: new (path: string) => { + exec(sql: string): void; + prepare(sql: string): { run(...params: unknown[]): void }; + close(): void; + }; + }; + const dir = join(root, "agents", agentId, "agent"); + mkdirSync(dir, { recursive: true }); + const db = new DatabaseSync(join(dir, "openclaw-agent.sqlite")); + db.exec(` + CREATE TABLE session_windows ( + session_id TEXT PRIMARY KEY, session_key TEXT NOT NULL, + updated_at INTEGER NOT NULL, transcript_updated_at INTEGER, + ended_at INTEGER, channel TEXT, chat_type TEXT, display_name TEXT + ); + CREATE TABLE transcript_events ( + session_id TEXT NOT NULL, seq INTEGER NOT NULL, + event_json TEXT NOT NULL, created_at INTEGER NOT NULL, + PRIMARY KEY(session_id, seq) + ); + CREATE TABLE session_nodes ( + current_session_id TEXT NOT NULL, entry_json TEXT NOT NULL, + label TEXT, display_name TEXT + ); + CREATE TABLE session_conversations ( + session_id TEXT, conversation_id TEXT, role TEXT, last_seen_at INTEGER + ); + CREATE TABLE conversations ( + conversation_id TEXT, channel TEXT, kind TEXT, peer_id TEXT, + delivery_target TEXT, label TEXT + ); + `); + db.prepare( + "INSERT INTO session_windows VALUES (?, ?, ?, ?, NULL, ?, NULL, NULL)", + ).run(uuid, `agent:${agentId}:main`, updatedAt, updatedAt, channel); + db.prepare("INSERT INTO transcript_events VALUES (?, 0, ?, ?)").run( + uuid, + JSON.stringify({ type: "session", id: uuid, cwd: `/work/${agentId}` }), + updatedAt, + ); + db.close(); + return true; + } catch { + return false; + } } function seed(): string { @@ -47,7 +107,12 @@ function seed(): string { lastChannel: "telegram", lastTo: "telegram:8674922496", chatType: "direct", - origin: { label: "Chetan (@chhhee10) id:8674922496", provider: "telegram", from: "telegram:8674922496", chatType: "direct" }, + origin: { + label: "Chetan (@chhhee10) id:8674922496", + provider: "telegram", + from: "telegram:8674922496", + chatType: "direct", + }, }, // A pure CLI/local session — no channel metadata. "agent:main:cli": { sessionId: UUID_CLI, lastInteractionAt: 2000 }, @@ -86,7 +151,11 @@ describe("getOpenClawSessions", () => { const sessions = await getOpenClawSessions(); // Sorted by mtime desc, ACROSS agents → weather-bot (9000), telegram // (5000), cli (2000). - expect(sessions.map((s) => s.sessionId)).toEqual([UUID_WEATHER, UUID_TG, UUID_CLI]); + expect(sessions.map((s) => s.sessionId)).toEqual([ + UUID_WEATHER, + UUID_TG, + UUID_CLI, + ]); const tg = sessions.find((s) => s.sessionId === UUID_TG)!; expect(tg.channel).toBe("telegram"); @@ -98,6 +167,60 @@ describe("getOpenClawSessions", () => { expect(cli.channel).toBe("local"); // no channel metadata → local expect(cli.label).toBeUndefined(); }); + + it("discovers SQLite-only agents, defaults their channel, and keeps profiles separate", async () => { + home = mkdtempSync(join(tmpdir(), "openclaw-sqlite-proj-")); + if (!(await writeSqliteSession(home, "main", UUID_SQLITE, 12_000))) return; + await writeSqliteSession( + home, + "research", + UUID_WEATHER, + 11_000, + "telegram", + ); + process.env.OPENCLAW_HOME = home; + + const sessions = await getOpenClawSessions(); + expect(sessions.map((s) => [s.agentId, s.sessionId, s.channel])).toEqual([ + ["main", UUID_SQLITE, "local"], + ["research", UUID_WEATHER, "telegram"], + ]); + + const projects = await getOpenClawProjects(); + expect(projects.map((p) => p.name)).toEqual([ + "openclaw-main-local", + "openclaw-research-telegram", + ]); + }); + + it("prefers a live SQLite row over an archived JSONL copy of the same session", async () => { + home = mkdtempSync(join(tmpdir(), "openclaw-sqlite-dedup-")); + const legacyDir = join(home, "agents", "main", "sessions"); + mkdirSync(legacyDir, { recursive: true }); + writeSession(legacyDir, UUID_SQLITE); + writeFileSync( + join(legacyDir, "sessions.json"), + JSON.stringify({ + old: { + sessionId: UUID_SQLITE, + lastInteractionAt: 1000, + lastChannel: "slack", + }, + }), + ); + if ( + !(await writeSqliteSession(home, "main", UUID_SQLITE, 15_000, "telegram")) + ) + return; + process.env.OPENCLAW_HOME = home; + + const sessions = await getOpenClawSessions(); + expect(sessions).toHaveLength(1); + expect(sessions[0]).toMatchObject({ channel: "telegram", mtimeMs: 15_000 }); + expect(sessions[0].transcriptPath).toBe( + `openclaw-sqlite://main/${UUID_SQLITE}`, + ); + }); }); describe("getOpenClawProjects / getOpenClawSessionsByEncodedName", () => { @@ -118,8 +241,12 @@ describe("getOpenClawProjects / getOpenClawSessionsByEncodedName", () => { // Regression: grouping by channel alone collapsed every agent on Telegram // into one row with one mixed session list. home = seed(); - const mainTg = await getOpenClawSessionsByEncodedName("openclaw-main-telegram"); - const weatherTg = await getOpenClawSessionsByEncodedName("openclaw-weather-bot-telegram"); + const mainTg = await getOpenClawSessionsByEncodedName( + "openclaw-main-telegram", + ); + const weatherTg = await getOpenClawSessionsByEncodedName( + "openclaw-weather-bot-telegram", + ); expect(mainTg.sessions.map((s) => s.sessionId)).toEqual([UUID_TG]); expect(weatherTg.sessions.map((s) => s.sessionId)).toEqual([UUID_WEATHER]); }); @@ -157,7 +284,11 @@ describe("getOpenClawProjects / getOpenClawSessionsByEncodedName", () => { writeFileSync( join(mainSessions, "sessions.json"), JSON.stringify({ - "agent:main:main": { sessionId: UUID_TG, lastInteractionAt: 5000, lastChannel: "bot-telegram" }, + "agent:main:main": { + sessionId: UUID_TG, + lastInteractionAt: 5000, + lastChannel: "bot-telegram", + }, }), ); process.env.OPENCLAW_HOME = home; @@ -168,7 +299,9 @@ describe("getOpenClawProjects / getOpenClawSessionsByEncodedName", () => { channel: "telegram", }); // Resolution still finds the real owner. - const resolved = await getOpenClawSessionsByEncodedName("openclaw-main-bot-telegram"); + const resolved = await getOpenClawSessionsByEncodedName( + "openclaw-main-bot-telegram", + ); expect(resolved.sessions.map((s) => s.sessionId)).toEqual([UUID_TG]); expect(resolved.cwd).toBe("openclaw:main:bot-telegram"); }); @@ -179,7 +312,9 @@ describe("getOpenClawProjects / getOpenClawSessionsByEncodedName", () => { home = seed(); const legacy = await getOpenClawSessionsByEncodedName("openclaw-telegram"); expect(legacy.cwd).toBe("openclaw:telegram"); - expect(legacy.sessions.map((s) => s.sessionId).sort()).toEqual([UUID_TG, UUID_WEATHER].sort()); + expect(legacy.sessions.map((s) => s.sessionId).sort()).toEqual( + [UUID_TG, UUID_WEATHER].sort(), + ); }); it("names sessions by origin.label and carries channel metadata; non-openclaw names return empty", async () => { diff --git a/__tests__/lib/openclaw-sessions.test.ts b/__tests__/lib/openclaw-sessions.test.ts index 3a3147f39..60ed6c4aa 100644 --- a/__tests__/lib/openclaw-sessions.test.ts +++ b/__tests__/lib/openclaw-sessions.test.ts @@ -15,16 +15,42 @@ import { findOpenClawTranscript, listOpenClawTranscripts, OPENCLAW_SESSION_ID_RE, + getOpenClawSessionLog, + openclawHome, } from "@/lib/openclaw-sessions"; import type { AssistantEntry } from "@/lib/log-entries"; const UUID = "f9e8516e-fed2-4e54-acbe-7a20aefc6cfa"; const LINES: Record[] = [ - { type: "session", version: 3, id: UUID, timestamp: "2026-07-14T05:01:34.420Z", cwd: "/home/node/.openclaw/workspace" }, - { type: "model_change", id: "m1", timestamp: "2026-07-14T05:01:34.431Z", provider: "groq", modelId: "x" }, - { type: "custom", customType: "model-snapshot", data: {}, id: "c1", timestamp: "2026-07-14T05:01:34.527Z" }, - { type: "message", id: "u1", parentId: null, timestamp: "2026-07-14T05:10:20.000Z", message: { role: "user", content: "run echo probe" } }, + { + type: "session", + version: 3, + id: UUID, + timestamp: "2026-07-14T05:01:34.420Z", + cwd: "/home/node/.openclaw/workspace", + }, + { + type: "model_change", + id: "m1", + timestamp: "2026-07-14T05:01:34.431Z", + provider: "groq", + modelId: "x", + }, + { + type: "custom", + customType: "model-snapshot", + data: {}, + id: "c1", + timestamp: "2026-07-14T05:01:34.527Z", + }, + { + type: "message", + id: "u1", + parentId: null, + timestamp: "2026-07-14T05:10:20.000Z", + message: { role: "user", content: "run echo probe" }, + }, { type: "message", id: "a1", @@ -35,7 +61,12 @@ const LINES: Record[] = [ model: "llama-4-scout", content: [ { type: "text", text: "Running it." }, - { type: "toolCall", id: "c5sf91qpf", name: "exec", arguments: { command: "echo probe-test-123" } }, + { + type: "toolCall", + id: "c5sf91qpf", + name: "exec", + arguments: { command: "echo probe-test-123" }, + }, ], }, }, @@ -66,7 +97,8 @@ describe("openclawLinesToLogEntries", () => { const entries = openclawLinesToLogEntries(LINES); const user = entries[0]; expect(user.type).toBe("user"); - if (user.type === "user") expect(user.message.content).toBe("run echo probe"); + if (user.type === "user") + expect(user.message.content).toBe("run echo probe"); }); it("parses assistant text + toolCall and pairs the toolResult by toolCallId", () => { @@ -102,7 +134,10 @@ describe("OpenClaw transcript resolution", () => { const home = mkdtempSync(join(tmpdir(), "openclaw-home-")); const sessions = join(home, "agents", "main", "sessions"); mkdirSync(sessions, { recursive: true }); - writeFileSync(join(sessions, `${UUID}.jsonl`), LINES.map((l) => JSON.stringify(l)).join("\n")); + writeFileSync( + join(sessions, `${UUID}.jsonl`), + LINES.map((l) => JSON.stringify(l)).join("\n"), + ); // Heavy OTel trace + pointer must be ignored. writeFileSync(join(sessions, `${UUID}.trajectory.jsonl`), "{}\n"); writeFileSync(join(sessions, `${UUID}.trajectory-path.json`), "{}\n"); @@ -112,7 +147,9 @@ describe("OpenClaw transcript resolution", () => { const found = listOpenClawTranscripts(); expect(found.map((t) => t.sessionId)).toEqual([UUID]); expect(found[0].agentId).toBe("main"); - expect(findOpenClawTranscript(UUID)).toBe(join(sessions, `${UUID}.jsonl`)); + expect(findOpenClawTranscript(UUID)).toBe( + join(sessions, `${UUID}.jsonl`), + ); // Traversal id never resolves. expect(findOpenClawTranscript("../../etc/passwd")).toBeNull(); } finally { @@ -121,4 +158,91 @@ describe("OpenClaw transcript resolution", () => { rmSync(home, { recursive: true, force: true }); } }); + + it("prefers OPENCLAW_STATE_DIR over OPENCLAW_HOME", () => { + const previousHome = process.env.OPENCLAW_HOME; + const previousState = process.env.OPENCLAW_STATE_DIR; + process.env.OPENCLAW_HOME = "/legacy"; + process.env.OPENCLAW_STATE_DIR = "/state"; + try { + expect(openclawHome()).toBe("/state"); + } finally { + if (previousHome === undefined) delete process.env.OPENCLAW_HOME; + else process.env.OPENCLAW_HOME = previousHome; + if (previousState === undefined) delete process.env.OPENCLAW_STATE_DIR; + else process.env.OPENCLAW_STATE_DIR = previousState; + } + }); + + it("loads and parses a SQLite transcript before an archived JSONL copy", async () => { + let DatabaseSync: new (path: string) => { + exec(sql: string): void; + prepare(sql: string): { run(...params: unknown[]): void }; + close(): void; + }; + try { + ({ DatabaseSync } = (await import("node:sqlite")) as unknown as { + DatabaseSync: typeof DatabaseSync; + }); + } catch { + return; + } + const home = mkdtempSync(join(tmpdir(), "openclaw-sqlite-log-")); + const agentDir = join(home, "agents", "main", "agent"); + const sessionsDir = join(home, "agents", "main", "sessions"); + mkdirSync(agentDir, { recursive: true }); + mkdirSync(sessionsDir, { recursive: true }); + writeFileSync( + join(sessionsDir, `${UUID}.jsonl`), + JSON.stringify({ + type: "message", + message: { role: "user", content: "archived" }, + }), + ); + const db = new DatabaseSync(join(agentDir, "openclaw-agent.sqlite")); + db.exec(`CREATE TABLE transcript_events ( + session_id TEXT, seq INTEGER, event_json TEXT, created_at INTEGER, + PRIMARY KEY(session_id, seq) + )`); + const insert = db.prepare( + "INSERT INTO transcript_events VALUES (?, ?, ?, ?)", + ); + insert.run( + UUID, + 0, + JSON.stringify({ type: "session", cwd: "/sqlite/work" }), + 1000, + ); + insert.run( + UUID, + 1, + JSON.stringify({ + type: "message", + timestamp: "2026-09-11T00:00:00Z", + message: { role: "user", content: "live sqlite" }, + }), + 2000, + ); + db.close(); + + const previous = process.env.OPENCLAW_HOME; + const previousState = process.env.OPENCLAW_STATE_DIR; + process.env.OPENCLAW_HOME = home; + delete process.env.OPENCLAW_STATE_DIR; + try { + const result = await getOpenClawSessionLog(UUID); + expect(result?.cwd).toBe("/sqlite/work"); + expect(result?.filePath).toBe(`openclaw-sqlite://main/${UUID}`); + expect(result?.entries[0].type).toBe("user"); + if (result?.entries[0].type === "user") { + expect(result.entries[0].message.content).toBe("live sqlite"); + } + } finally { + if (previous === undefined) delete process.env.OPENCLAW_HOME; + else process.env.OPENCLAW_HOME = previous; + if (previousState === undefined) delete process.env.OPENCLAW_STATE_DIR; + else process.env.OPENCLAW_STATE_DIR = previousState; + rmSync(home, { recursive: true, force: true }); + } + }); }); diff --git a/crates/failproofaid/src/main.rs b/crates/failproofaid/src/main.rs index 0f88e4965..56b52cc2a 100644 --- a/crates/failproofaid/src/main.rs +++ b/crates/failproofaid/src/main.rs @@ -1031,12 +1031,13 @@ fn collector_tasks() -> Vec { ); let openclaw_roots = openclaw::default_roots(); + let openclaw_extra = extras("openclaw", &openclaw_roots); file_source( &mut tasks, "openclaw", openclaw::FORMAT, openclaw_roots.clone(), - &extras("openclaw", &openclaw_roots), + &openclaw_extra, openclaw::DEFAULT_AGENT_ID, &spool, &cursors, @@ -1045,6 +1046,17 @@ fn collector_tasks() -> Vec { os_user.as_deref(), redact, ); + openclaw_sqlite_harness( + &mut tasks, + openclaw_roots, + &openclaw_extra, + &spool, + &cursors, + &env, + machine.as_deref(), + os_user.as_deref(), + redact, + ); let pi_roots = vec![pi::sessions_root()]; file_source( @@ -1410,6 +1422,94 @@ fn file_source_instance( })); } +/// Register OpenClaw's 2026.9.2+ SQLite transcript store alongside the legacy +/// JSONL tailer. Each configured root gets an independent cursor store because +/// the stores are atomically rewritten whole and must never have two writers. +#[allow(clippy::too_many_arguments)] +fn openclaw_sqlite_harness( + tasks: &mut Vec, + roots: Vec, + extra: &[fpai_collect::ExtraPath], + spool_dir: &std::path::Path, + cursor_root: &std::path::Path, + environment: &str, + machine_id: Option<&str>, + user: Option<&str>, + redact: fpai_collect::Redact, +) { + openclaw_sqlite_source( + tasks, + roots, + None, + cursor_root.join("openclaw-sqlite"), + None, + spool_dir, + environment, + machine_id, + user, + redact, + ); + for ep in extra { + openclaw_sqlite_source( + tasks, + vec![ep.path.clone()], + Some(ep.label.clone()), + cursor_root.join("openclaw-sqlite").join(&ep.label), + Some(format!("openclaw-sqlite:{}", ep.label)), + spool_dir, + environment, + machine_id, + user, + redact, + ); + } +} + +#[allow(clippy::too_many_arguments)] +fn openclaw_sqlite_source( + tasks: &mut Vec, + roots: Vec, + label: Option, + state_dir: std::path::PathBuf, + health_key: Option, + spool_dir: &std::path::Path, + environment: &str, + machine_id: Option<&str>, + user: Option<&str>, + redact: fpai_collect::Redact, +) { + let task_name = match &label { + Some(label) => format!("openclaw-sqlite:{label}"), + None => "openclaw-sqlite".to_string(), + }; + let spool_dir = spool_dir.to_path_buf(); + let environment = environment.to_string(); + let machine_id = machine_id.map(str::to_string); + let user = user.map(str::to_string); + tasks.push(fpai_collect::TaskSpec::new(task_name, move |sd| { + fpai_collect::sources::openclaw::sqlite::run( + fpai_collect::sources::openclaw::sqlite::Spec { + roots: roots.clone(), + spool_dir: spool_dir.clone(), + state_dir: state_dir.clone(), + poll_interval: std::time::Duration::from_secs(2), + health_key: health_key.clone(), + params: fpai_collect::sources::openclaw::sqlite::Params { + environment: environment.clone(), + redact, + machine_id: machine_id.clone(), + user: user.clone(), + label: label.clone(), + max_rows_per_session: 2_000, + max_batch_bytes: fpai_collect::spool::DEFAULT_MAX_BATCH_BYTES, + since_days: file_source_since_days(), + }, + }, + sd, + ) + })); +} + /// Register one SQLite-polling source: its default database, plus one further /// instance per configured extra path. /// diff --git a/crates/fpai-collect/src/cursor.rs b/crates/fpai-collect/src/cursor.rs index 4503a56ea..f5368c331 100644 --- a/crates/fpai-collect/src/cursor.rs +++ b/crates/fpai-collect/src/cursor.rs @@ -151,6 +151,19 @@ pub struct FileCursor { pub head_fingerprint: Option, #[serde(default)] pub state: TailState, + /// Last OpenClaw SQLite transcript sequence durably spooled. + /// + /// The SQLite store is ordered per session, so its adapter uses a + /// synthetic `(dev, inode)` key per `(database, session_id)` and carries + /// the real row position here. `None` means no row has been consumed yet; + /// OpenClaw sequences begin at zero. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sqlite_seq: Option, + /// OpenClaw rotates this token whenever it destructively rewrites a + /// transcript. A changed generation invalidates the sequence, byte offset + /// and transform state together and forces a deterministic re-read. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sqlite_generation: Option, } /// Bytes of a file's head that [`FileCursor::head_fingerprint`] covers. diff --git a/crates/fpai-collect/src/filetail.rs b/crates/fpai-collect/src/filetail.rs index d5c18c0a4..d9ec6801d 100644 --- a/crates/fpai-collect/src/filetail.rs +++ b/crates/fpai-collect/src/filetail.rs @@ -522,6 +522,8 @@ async fn new_cursor( first_seen_epoch_ms: mtime_epoch_ms(meta), head_fingerprint: Some(cursor::head_fingerprint(&head)), state, + sqlite_seq: None, + sqlite_generation: None, })) } diff --git a/crates/fpai-collect/src/sources/mod.rs b/crates/fpai-collect/src/sources/mod.rs index d70c0c543..790355a4d 100644 --- a/crates/fpai-collect/src/sources/mod.rs +++ b/crates/fpai-collect/src/sources/mod.rs @@ -4,11 +4,15 @@ //! //! Sources fall into three shapes: //! -//! * **File tailers** ([`crate::filetail`]) — claude, codex, copilot, openclaw, -//! pi, factory, antigravity. Each supplies a `Format` table of pure functions. +//! * **File tailers** ([`crate::filetail`]) — claude, codex, copilot, legacy +//! openclaw, pi, factory, antigravity. Each supplies a `Format` table of pure +//! functions. //! * **SQLite pollers** ([`crate::sqlitepoll`]) — goose, opencode, hermes, //! devin. Each supplies a `SqliteFormat` and declares how its database orders //! changes. +//! * **OpenClaw SQLite** — 2026.9.2+ uses one database per agent and per-session +//! sequence/rewrite cursors, so it has a specialised poller beside its legacy +//! file adapter. //! * **The hook stream** ([`hooks`]) — CLI-agnostic, and the one capability //! that comes from failproofai sitting in the hook path rather than reading //! somebody else's files. diff --git a/crates/fpai-collect/src/sources/openclaw/mod.rs b/crates/fpai-collect/src/sources/openclaw/mod.rs index 3f16c655a..bc9fe1d65 100644 --- a/crates/fpai-collect/src/sources/openclaw/mod.rs +++ b/crates/fpai-collect/src/sources/openclaw/mod.rs @@ -1,9 +1,13 @@ -//! OpenClaw session capture — a [`filetail`](crate::filetail) adapter. +//! OpenClaw session capture from both generations of its transcript storage. //! -//! OpenClaw writes live-appended JSONL transcripts at +//! OpenClaw through 2026.7 wrote live-appended JSONL transcripts at //! `/agents//sessions/.jsonl`, where `` is //! `$OPENCLAW_STATE_DIR`, else `$OPENCLAW_HOME`, else `~/.openclaw`. We open //! them read-only; OpenClaw's own files are never written, moved or deleted. +//! OpenClaw 2026.9.2 moved the live stream to +//! `/agents//agent/openclaw-agent.sqlite`; [`sqlite`] captures +//! that store while this module's [`FORMAT`] keeps old and archived JSONL +//! sessions working. //! //! # The sibling that must never be discovered //! @@ -30,10 +34,9 @@ //! nested below the sessions directory. Excluded by requiring the transcript //! to sit *directly* in `sessions/`. //! -//! `/state/openclaw.sqlite` (964 KB on the probe capture) and the -//! agent's `workspace/` git checkout are outside the root entirely — see -//! [`default_roots`], which points at `agents/` rather than the state directory -//! so neither is even walked. +//! `/state/openclaw.sqlite` (the unrelated global state database) and +//! the agent's `workspace/` git checkout are outside the root entirely — see +//! [`default_roots`], which points at `agents/` rather than the state directory. //! //! # Grouping is by agent, not by working directory //! @@ -54,6 +57,7 @@ //! model snapshots) advance the session clock but emit nothing — they describe //! the harness, not the conversation. +pub mod sqlite; pub mod transform; use std::path::{Path, PathBuf}; diff --git a/crates/fpai-collect/src/sources/openclaw/sqlite.rs b/crates/fpai-collect/src/sources/openclaw/sqlite.rs new file mode 100644 index 000000000..13f239f0f --- /dev/null +++ b/crates/fpai-collect/src/sources/openclaw/sqlite.rs @@ -0,0 +1,433 @@ +//! OpenClaw 2026.9.2+ transcript capture from per-agent SQLite databases. +//! +//! Live transcripts moved from `sessions/.jsonl` to +//! `agent/openclaw-agent.sqlite`. The `event_json` rows are the exact logical +//! JSONL records, so this adapter feeds them through the existing OpenClaw +//! transform and assigns the byte offsets they would have had in the archived +//! JSONL file. That keeps legacy-file and SQLite delivery dedup-compatible. +//! +//! A global rowid watermark is deliberately not used. `seq` is scoped to one +//! session, and OpenClaw can replace or rewrite prior rows. The +//! `transcript_rewrite_watermarks.generation` token is the authority for that +//! case: when it changes, all derived state for that session is reset and the +//! current generation is read again from sequence zero. + +use std::path::{Path, PathBuf}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use rusqlite::{Connection, OpenFlags, OptionalExtension}; + +use crate::config::Redact; +use crate::cursor::{CursorStore, FileCursor}; +use crate::filetail::Ctx; +use crate::spool::SpoolWriter; +use crate::supervisor::{Shutdown, TaskError}; + +use super::{DEFAULT_AGENT_ID, transform}; + +const DB_NAME: &str = "openclaw-agent.sqlite"; +const AGENT_DIR: &str = "agent"; +const BUSY_TIMEOUT: Duration = Duration::from_secs(5); +const HEADER_ROWS: i64 = 64; + +#[derive(Debug, Clone)] +pub struct Params { + pub environment: String, + pub redact: Redact, + pub machine_id: Option, + pub user: Option, + pub label: Option, + pub max_rows_per_session: u64, + pub max_batch_bytes: u64, + /// Skip sessions with no activity inside this window on first discovery. + pub since_days: Option, +} + +pub struct Spec { + /// Each root is OpenClaw's `agents/` directory. + pub roots: Vec, + pub spool_dir: PathBuf, + pub state_dir: PathBuf, + pub poll_interval: Duration, + pub params: Params, + pub health_key: Option, +} + +#[derive(Debug)] +struct SessionRow { + session_id: String, + generation: Option, + activity_ms: i64, + ended_at: Option, + max_seq: i64, +} + +#[derive(Debug)] +struct TranscriptRow { + seq: i64, + event_json: String, +} + +#[derive(Debug)] +struct TranscriptPoll { + generation: Option, + max_seq: i64, + header: Vec, + rows: Vec, +} + +#[derive(Debug)] +struct DbPoll { + sessions: Vec, +} + +/// Poll every discovered per-agent database until shutdown. +pub async fn run(spec: Spec, sd: Shutdown) -> Result<(), TaskError> { + let health_key = spec + .health_key + .clone() + .unwrap_or_else(|| "openclaw-sqlite".to_string()); + let mut cursors = CursorStore::load(spec.state_dir.clone()); + + loop { + let databases = discover_databases(&spec.roots); + let present = !databases.is_empty(); + let mut events = 0u64; + let mut first_error = None; + + for db_path in databases { + match process_database(&spec, &mut cursors, &db_path).await { + Ok(n) => events += n, + Err(err) => { + tracing::warn!(db = %db_path.display(), %err, "could not process OpenClaw database"); + if first_error.is_none() { + first_error = Some(format!("{}: {err}", db_path.display())); + } + } + } + } + + cursors.retain_existing(); + cursors.save().map_err(io_err)?; + crate::health::report_poll(&health_key, present, events, cursors.len() as u64); + if let Some(err) = first_error { + crate::health::report_error(&health_key, &err); + } + + if !sd.sleep(spec.poll_interval).await { + return Ok(()); + } + } +} + +/// Discover `//agent/openclaw-agent.sqlite` dynamically. +pub fn discover_databases(roots: &[PathBuf]) -> Vec { + let mut out = Vec::new(); + for root in roots { + if root.file_name().is_some_and(|name| name == DB_NAME) && root.is_file() { + out.push(root.clone()); + continue; + } + discover_under_agents(root, &mut out); + // Extra paths have historically accepted either the `agents/` folder + // or the OpenClaw state directory containing it. + discover_under_agents(&root.join("agents"), &mut out); + } + out.sort(); + out.dedup(); + out +} + +fn discover_under_agents(root: &Path, out: &mut Vec) { + let direct = root.join(AGENT_DIR).join(DB_NAME); + if direct.is_file() { + out.push(direct); + } + let Ok(entries) = std::fs::read_dir(root) else { + return; + }; + for entry in entries.flatten() { + let candidate = entry.path().join(AGENT_DIR).join(DB_NAME); + if candidate.is_file() { + out.push(candidate); + } + } +} + +async fn process_database( + spec: &Spec, + cursors: &mut CursorStore, + db_path: &Path, +) -> Result { + let path = db_path.to_path_buf(); + let db_poll = tokio::task::spawn_blocking(move || read_sessions(&path)) + .await + .map_err(|e| TaskError::from(format!("OpenClaw SQLite task failed: {e}")))? + .map_err(sql_err)?; + + let agent_id = agent_id_from_database(db_path).unwrap_or_else(|| DEFAULT_AGENT_ID.to_string()); + let db_key = stable_hash(db_path.to_string_lossy().as_bytes()); + let mut emitted = 0u64; + + for session in db_poll.sessions { + let session_key = stable_hash(session.session_id.as_bytes()); + let existing = cursors.resume(db_key, session_key, db_path).cloned(); + if existing.is_none() && !within_window(session.activity_ms, spec.params.since_days) { + continue; + } + + let mut cursor = existing.unwrap_or_else(|| FileCursor { + path: db_path.to_path_buf(), + dev: db_key, + inode: session_key, + session_id: Some(session.session_id.clone()), + agent_id: Some(agent_id.clone()), + sqlite_seq: None, + sqlite_generation: session.generation.clone(), + ..Default::default() + }); + + if cursor.sqlite_generation != session.generation { + cursor = FileCursor { + path: db_path.to_path_buf(), + dev: db_key, + inode: session_key, + session_id: Some(session.session_id.clone()), + agent_id: Some(agent_id.clone()), + sqlite_seq: None, + sqlite_generation: session.generation.clone(), + ..Default::default() + }; + } + + let has_new_rows = cursor.sqlite_seq.is_none_or(|seq| seq < session.max_seq); + let needs_end = !cursor.ended + && cursor.agent_start_emitted + && session.ended_at.is_some() + && !has_new_rows; + if !has_new_rows && cursor.agent_start_emitted && !needs_end { + continue; + } + + let path = db_path.to_path_buf(); + let session_id = session.session_id.clone(); + let after_seq = cursor.sqlite_seq.unwrap_or(-1); + let limit = spec.params.max_rows_per_session.max(1) as i64; + let transcript = tokio::task::spawn_blocking(move || { + read_transcript(&path, &session_id, after_seq, limit) + }) + .await + .map_err(|e| TaskError::from(format!("OpenClaw SQLite task failed: {e}")))? + .map_err(sql_err)?; + // A rewrite between the session-list query and this transcript + // snapshot would mix two generations. Leave the durable cursor alone; + // the next poll will see the new token and restart cleanly. + if transcript.generation != session.generation { + continue; + } + + let ctx = Ctx { + session_id: session.session_id.clone(), + agent_id: cursor.agent_id.clone().unwrap_or_else(|| agent_id.clone()), + environment: spec.params.environment.clone(), + file_epoch_ms: None, + }; + let mut writer = SpoolWriter::new( + spec.spool_dir.clone(), + spec.params.max_batch_bytes, + "openclaw", + &ctx.session_id, + ) + .with_label(spec.params.label.clone()) + .with_machine_id(spec.params.machine_id.clone()) + .with_user(spec.params.user.clone()) + .with_redact(spec.params.redact); + + if !cursor.agent_start_emitted + && let Some((event, ts)) = transform::agent_start(&transcript.header, &ctx, 0) + { + writer.push(event).await.map_err(io_err)?; + emitted += 1; + cursor.agent_start_emitted = true; + if cursor.last_ts.is_none() { + cursor.last_ts = ts; + } + } + + if cursor.ended + && transcript + .rows + .last() + .is_some_and(|r| Some(r.seq) > cursor.sqlite_seq) + { + cursor.ended = false; + } + + for row in transcript.rows { + let offset = cursor.offset; + let (ts, events) = + transform::transform_line(&row.event_json, &ctx, offset, &mut cursor.state); + if let Some(ts) = ts { + cursor.last_ts = Some(ts); + } + for event in events { + writer.push(event).await.map_err(io_err)?; + emitted += 1; + } + cursor.offset += row.event_json.len() as u64 + 1; + cursor.size_seen = cursor.offset; + cursor.sqlite_seq = Some(row.seq); + } + + if !cursor.ended + && cursor.agent_start_emitted + && session.ended_at.is_some() + && cursor + .sqlite_seq + .is_some_and(|seq| seq >= transcript.max_seq) + && let Some(last_ts) = cursor.last_ts.clone() + { + writer + .push(transform::agent_end(&ctx, &last_ts, cursor.offset)) + .await + .map_err(io_err)?; + emitted += 1; + cursor.ended = true; + } + + // Flush before advancing the durable cursor. A crash in between only + // re-ships deterministic events, which server-side dedup collapses. + writer.flush().await.map_err(io_err)?; + cursors.set(cursor); + } + + Ok(emitted) +} + +fn read_sessions(path: &Path) -> rusqlite::Result { + let conn = open_readonly(path)?; + let mut stmt = conn.prepare( + "SELECT w.session_id, r.generation, + MAX(COALESCE(w.transcript_updated_at, 0), + COALESCE(w.updated_at, 0), + COALESCE((SELECT MAX(e.created_at) FROM transcript_events e + WHERE e.session_id = w.session_id), 0)) AS activity_ms, + w.ended_at, + (SELECT MAX(e.seq) FROM transcript_events e + WHERE e.session_id = w.session_id) AS max_seq + FROM session_windows w + LEFT JOIN transcript_rewrite_watermarks r ON r.session_id = w.session_id + WHERE EXISTS (SELECT 1 FROM transcript_events e WHERE e.session_id = w.session_id) + ORDER BY activity_ms ASC, w.session_id ASC", + )?; + let sessions = stmt + .query_map([], |row| { + Ok(SessionRow { + session_id: row.get(0)?, + generation: row.get(1)?, + activity_ms: row.get(2)?, + ended_at: row.get(3)?, + max_seq: row.get(4)?, + }) + })? + .collect::>>()?; + Ok(DbPoll { sessions }) +} + +fn read_transcript( + path: &Path, + session_id: &str, + after_seq: i64, + limit: i64, +) -> rusqlite::Result { + let mut conn = open_readonly(path)?; + let tx = conn.transaction()?; + let generation = tx + .query_row( + "SELECT generation FROM transcript_rewrite_watermarks WHERE session_id = ?1", + [session_id], + |row| row.get(0), + ) + .optional()?; + let max_seq = tx.query_row( + "SELECT COALESCE(MAX(seq), -1) FROM transcript_events WHERE session_id = ?1", + [session_id], + |row| row.get(0), + )?; + let mut header_stmt = tx.prepare( + "SELECT event_json FROM transcript_events + WHERE session_id = ?1 ORDER BY seq ASC LIMIT ?2", + )?; + let header = header_stmt + .query_map((session_id, HEADER_ROWS), |row| row.get(0))? + .collect::>>()?; + + let mut rows_stmt = tx.prepare( + "SELECT seq, event_json FROM transcript_events + WHERE session_id = ?1 AND seq > ?2 ORDER BY seq ASC LIMIT ?3", + )?; + let rows = rows_stmt + .query_map((session_id, after_seq, limit), |row| { + Ok(TranscriptRow { + seq: row.get(0)?, + event_json: row.get(1)?, + }) + })? + .collect::>>()?; + drop(rows_stmt); + drop(header_stmt); + tx.commit()?; + Ok(TranscriptPoll { + generation, + max_seq, + header, + rows, + }) +} + +fn open_readonly(path: &Path) -> rusqlite::Result { + let conn = Connection::open_with_flags( + path, + OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, + )?; + conn.busy_timeout(BUSY_TIMEOUT)?; + Ok(conn) +} + +fn agent_id_from_database(path: &Path) -> Option { + let agent_dir = path.parent()?.parent()?; + if agent_dir.parent()?.file_name()? != "agents" { + return None; + } + let id = transform::sanitize_id_part(agent_dir.file_name()?.to_str()?); + (!id.is_empty()).then(|| format!("openclaw-{id}")) +} + +fn within_window(activity_ms: i64, days: Option) -> bool { + let Some(days) = days else { + return true; + }; + let now_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as i128) + .unwrap_or(0); + let window_ms = i128::from(days) * 24 * 60 * 60 * 1000; + i128::from(activity_ms) >= now_ms.saturating_sub(window_ms) +} + +fn stable_hash(bytes: &[u8]) -> u64 { + let mut hash = 0xcbf2_9ce4_8422_2325u64; + for byte in bytes { + hash ^= u64::from(*byte); + hash = hash.wrapping_mul(0x0000_0100_0000_01b3); + } + hash +} + +fn sql_err(err: rusqlite::Error) -> TaskError { + TaskError::from(err.to_string()) +} + +fn io_err(err: std::io::Error) -> TaskError { + TaskError::from(err.to_string()) +} diff --git a/crates/fpai-collect/tests/openclaw_source.rs b/crates/fpai-collect/tests/openclaw_source.rs index 39f51ec0e..ffddcfb4a 100644 --- a/crates/fpai-collect/tests/openclaw_source.rs +++ b/crates/fpai-collect/tests/openclaw_source.rs @@ -13,6 +13,7 @@ use fpai_collect::cursor::TailState; use fpai_collect::filetail::{self, Ctx, Params, RereadPolicy, Spec}; use fpai_collect::sources::openclaw::{self, transform}; use fpai_collect::supervisor::Shutdown; +use rusqlite::{Connection, params}; use serde_json::{Value, json}; const UUID: &str = "0c751d66-8f74-429d-a604-b29855d36c41"; @@ -618,6 +619,117 @@ async fn run_briefly(s: Spec, ms: u64) { let _ = tokio::time::timeout(Duration::from_millis(ms), filetail::run(s, sd)).await; } +fn sqlite_spec(base: PathBuf, spool: PathBuf, state: PathBuf) -> openclaw::sqlite::Spec { + openclaw::sqlite::Spec { + roots: vec![base.join("agents")], + spool_dir: spool, + state_dir: state, + poll_interval: Duration::from_millis(100), + health_key: Some("openclaw-sqlite-test".into()), + params: openclaw::sqlite::Params { + environment: "local".into(), + redact: fpai_collect::Redact::Minimal, + machine_id: None, + user: None, + label: None, + max_rows_per_session: 2_000, + max_batch_bytes: 8 * 1024 * 1024, + since_days: None, + }, + } +} + +async fn run_sqlite_briefly(s: openclaw::sqlite::Spec, ms: u64) { + let sd = Shutdown::for_test(Arc::new(AtomicBool::new(false))); + let _ = tokio::time::timeout(Duration::from_millis(ms), openclaw::sqlite::run(s, sd)).await; +} + +fn openclaw_db(base: &Path) -> PathBuf { + let dir = base.join("agents").join("main").join("agent"); + fs::create_dir_all(&dir).unwrap(); + dir.join("openclaw-agent.sqlite") +} + +fn create_openclaw_db(base: &Path) -> Connection { + let conn = Connection::open(openclaw_db(base)).unwrap(); + conn.execute_batch( + "PRAGMA journal_mode = WAL; + PRAGMA wal_autocheckpoint = 0; + CREATE TABLE session_windows ( + session_id TEXT PRIMARY KEY, + updated_at INTEGER NOT NULL, + transcript_updated_at INTEGER, + ended_at INTEGER + ); + CREATE TABLE transcript_events ( + session_id TEXT NOT NULL, + seq INTEGER NOT NULL, + event_json TEXT NOT NULL, + created_at INTEGER NOT NULL, + PRIMARY KEY (session_id, seq) + ); + CREATE TABLE transcript_rewrite_watermarks ( + session_id TEXT PRIMARY KEY, + generation TEXT NOT NULL, + updated_at INTEGER NOT NULL + );", + ) + .unwrap(); + conn +} + +fn replace_sqlite_session(conn: &Connection, generation: &str, lines: &[String], ended: bool) { + let activity = 1_788_251_542_907i64; + conn.execute( + "DELETE FROM transcript_events WHERE session_id = ?1", + [UUID], + ) + .unwrap(); + conn.execute( + "INSERT INTO session_windows(session_id, updated_at, transcript_updated_at, ended_at) + VALUES(?1, ?2, ?2, ?3) + ON CONFLICT(session_id) DO UPDATE SET + updated_at = excluded.updated_at, + transcript_updated_at = excluded.transcript_updated_at, + ended_at = excluded.ended_at", + params![UUID, activity, ended.then_some(activity)], + ) + .unwrap(); + conn.execute( + "INSERT INTO transcript_rewrite_watermarks(session_id, generation, updated_at) + VALUES(?1, ?2, ?3) + ON CONFLICT(session_id) DO UPDATE SET + generation = excluded.generation, updated_at = excluded.updated_at", + params![UUID, generation, activity], + ) + .unwrap(); + for (seq, line) in lines.iter().enumerate() { + conn.execute( + "INSERT INTO transcript_events(session_id, seq, event_json, created_at) + VALUES(?1, ?2, ?3, ?4)", + params![UUID, seq as i64, line, activity + seq as i64], + ) + .unwrap(); + } +} + +fn append_sqlite_event(conn: &Connection, seq: i64, line: &str) { + let activity = 1_788_251_600_000i64 + seq; + conn.execute( + "INSERT INTO transcript_events(session_id, seq, event_json, created_at) + VALUES(?1, ?2, ?3, ?4)", + params![UUID, seq, line, activity], + ) + .unwrap(); + conn.execute( + "UPDATE session_windows + SET updated_at = ?2, transcript_updated_at = ?2, ended_at = NULL + WHERE session_id = ?1", + params![UUID, activity], + ) + .unwrap(); +} + /// A tree laid out exactly like OpenClaw's: `/agents//sessions/`. fn sessions_dir(base: &Path) -> PathBuf { let d = base.join("agents").join("main").join("sessions"); @@ -859,3 +971,163 @@ async fn a_partially_written_final_line_is_held_back_then_picked_up() { fs::remove_dir_all(&spool).ok(); fs::remove_dir_all(&state).ok(); } + +// ── OpenClaw 2026.9.2+ SQLite store ───────────────────────────────────── + +#[test] +fn sqlite_databases_are_discovered_per_agent_and_nothing_else_is_claimed() { + let root = tmpdir("sqlite-discovery"); + let expected = openclaw_db(&root); + let other = root + .join("agents") + .join("research") + .join("agent") + .join("openclaw-agent.sqlite"); + fs::create_dir_all(other.parent().unwrap()).unwrap(); + fs::write(&expected, b"").unwrap(); + fs::write(&other, b"").unwrap(); + fs::write(root.join("agents").join("not-a-database.sqlite"), b"").unwrap(); + + assert_eq!( + openclaw::sqlite::discover_databases(&[root.join("agents")]), + vec![expected.clone(), other.clone()] + ); + assert_eq!( + openclaw::sqlite::discover_databases(std::slice::from_ref(&root)), + vec![expected, other], + "an extra path may name the OpenClaw state directory" + ); + fs::remove_dir_all(&root).ok(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn sqlite_and_legacy_jsonl_produce_dedup_identical_events() { + let root = tmpdir("sqlite-equivalent-root"); + let legacy_spool = tmpdir("sqlite-equivalent-legacy-spool"); + let legacy_state = tmpdir("sqlite-equivalent-legacy-state"); + let sqlite_spool = tmpdir("sqlite-equivalent-db-spool"); + let sqlite_state = tmpdir("sqlite-equivalent-db-state"); + let lines = full_session(); + write_session(&root, &lines); + + // Checkpoint the schema, then leave the transcript rows in the live WAL. + // A reader opened with `immutable=1` would miss these rows entirely. + let conn = create_openclaw_db(&root); + conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);") + .unwrap(); + replace_sqlite_session(&conn, "generation-a", &lines, true); + let wal = PathBuf::from(format!("{}-wal", openclaw_db(&root).display())); + assert!( + fs::metadata(&wal).unwrap().len() > 0, + "fixture must live in WAL" + ); + + run_briefly( + spec(root.clone(), legacy_spool.clone(), legacy_state.clone()), + 700, + ) + .await; + let mut db_spec = sqlite_spec(root.clone(), sqlite_spool.clone(), sqlite_state.clone()); + db_spec.params.max_rows_per_session = 2; + run_sqlite_briefly(db_spec, 900).await; + + let mut legacy: Vec = spooled(&legacy_spool) + .into_iter() + .map(|event| serde_json::to_string(&event).unwrap()) + .collect(); + let mut sqlite: Vec = spooled(&sqlite_spool) + .into_iter() + .map(|event| serde_json::to_string(&event).unwrap()) + .collect(); + legacy.sort(); + sqlite.sort(); + assert_eq!( + sqlite, legacy, + "SQLite rows must retain JSONL offsets and event identities" + ); + + drop(conn); + fs::remove_dir_all(&root).ok(); + fs::remove_dir_all(&legacy_spool).ok(); + fs::remove_dir_all(&legacy_state).ok(); + fs::remove_dir_all(&sqlite_spool).ok(); + fs::remove_dir_all(&sqlite_state).ok(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn sqlite_resume_only_ships_rows_appended_after_the_saved_sequence() { + let root = tmpdir("sqlite-resume-root"); + let spool = tmpdir("sqlite-resume-spool"); + let state = tmpdir("sqlite-resume-state"); + let conn = create_openclaw_db(&root); + let lines = full_session(); + replace_sqlite_session(&conn, "generation-a", &lines, false); + + run_sqlite_briefly(sqlite_spec(root.clone(), spool.clone(), state.clone()), 500).await; + assert!(!spooled(&spool).is_empty()); + clear(&spool); + + run_sqlite_briefly(sqlite_spec(root.clone(), spool.clone(), state.clone()), 350).await; + assert!(spooled(&spool).is_empty(), "a resumed poll must be idle"); + + append_sqlite_event( + &conn, + lines.len() as i64, + &assistant_text("2026-08-03T08:06:00.000Z", "new SQLite turn"), + ); + run_sqlite_briefly(sqlite_spec(root.clone(), spool.clone(), state.clone()), 500).await; + let events = spooled(&spool); + assert_eq!( + events.len(), + 1, + "old rows must not be re-shipped: {events:?}" + ); + assert_eq!(events[0]["type"], "model_response"); + assert_eq!(events[0]["content"], "new SQLite turn"); + + drop(conn); + fs::remove_dir_all(&root).ok(); + fs::remove_dir_all(&spool).ok(); + fs::remove_dir_all(&state).ok(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_rewrite_generation_change_resets_sequence_offset_and_transform_state() { + let root = tmpdir("sqlite-rewrite-root"); + let spool = tmpdir("sqlite-rewrite-spool"); + let state = tmpdir("sqlite-rewrite-state"); + let conn = create_openclaw_db(&root); + replace_sqlite_session(&conn, "generation-a", &full_session(), false); + + run_sqlite_briefly(sqlite_spec(root.clone(), spool.clone(), state.clone()), 500).await; + clear(&spool); + + let rewritten = vec![ + session_header("2026-08-03T09:00:00.000Z"), + model_change("2026-08-03T09:00:00.010Z", "rewritten-model"), + user_prompt("2026-08-03T09:00:00.020Z", "rewritten opening prompt"), + ]; + replace_sqlite_session(&conn, "generation-b", &rewritten, false); + run_sqlite_briefly(sqlite_spec(root.clone(), spool.clone(), state.clone()), 500).await; + + let events = spooled(&spool); + let start = events + .iter() + .find(|event| event["type"] == "agent_start") + .unwrap(); + assert_eq!(start["goal"], "rewritten opening prompt"); + let request = events + .iter() + .find(|event| event["type"] == "model_request") + .unwrap(); + assert_eq!(request["model"], "rewritten-model"); + assert_eq!( + request["openclaw_line_offset"], + rewritten[0].len() + 1 + rewritten[1].len() + 1 + ); + + drop(conn); + fs::remove_dir_all(&root).ok(); + fs::remove_dir_all(&spool).ok(); + fs::remove_dir_all(&state).ok(); +} diff --git a/lib/download-session.ts b/lib/download-session.ts index 1eea27f53..e9db577b4 100644 --- a/lib/download-session.ts +++ b/lib/download-session.ts @@ -2,8 +2,8 @@ * Per-CLI dispatcher for the dashboard's Download Logs endpoint. * * Returns either a real on-disk path (so the route can `createReadStream` and - * stream the bytes verbatim) or a synthesized JSONL body (used only by - * OpenCode, whose transcripts live in SQLite rather than on disk). + * stream the bytes verbatim) or a synthesized JSON/JSONL body for agents whose + * transcripts live in SQLite. * * Per-CLI session loaders / transcript finders already exist in their own * files; this module is the thin glue that picks the right one. Imports of @@ -34,7 +34,12 @@ export const GOOSE_SESSION_RE = /^\d{8}_\d+$/; export type DownloadSource = | { kind: "file"; path: string } - | { kind: "synthesized"; body: string; contentType: string; extension: string }; + | { + kind: "synthesized"; + body: string; + contentType: string; + extension: string; + }; /** Validate a session ID against the per-CLI shape. OpenCode uses `ses_*` * prefixes; everyone else is a UUID. */ @@ -98,7 +103,12 @@ export async function resolveDownloadSource( const result = await getOpenCodeSessionExport(sessionId); if (!result) return null; const body = JSON.stringify(result, null, 2) + "\n"; - return { kind: "synthesized", body, contentType: "application/json", extension: "json" }; + return { + kind: "synthesized", + body, + contentType: "application/json", + extension: "json", + }; } if (cli === "hermes") { @@ -107,13 +117,35 @@ export async function resolveDownloadSource( const { getHermesSessionLog } = await import("./hermes-sessions"); const result = await getHermesSessionLog(sessionId); if (!result) return null; - const body = result.rawLines.map((r) => JSON.stringify(r)).join("\n") + "\n"; - return { kind: "synthesized", body, contentType: "application/x-ndjson", extension: "jsonl" }; + const body = + result.rawLines.map((r) => JSON.stringify(r)).join("\n") + "\n"; + return { + kind: "synthesized", + body, + contentType: "application/x-ndjson", + extension: "jsonl", + }; } if (cli === "openclaw") { - // OpenClaw writes real JSONL transcripts on disk — stream the file verbatim. - const { findOpenClawTranscript } = await import("./openclaw-sessions"); + // New OpenClaw versions keep live transcripts in SQLite. Export the exact + // event_json rows as JSONL; fall back to streaming archived JSONL files. + const { findOpenClawTranscript, listOpenClawAgents, openclawHome } = + await import("./openclaw-sessions"); + const { readOpenClawSqliteTranscript } = await import("./openclaw-db"); + const sqlite = await readOpenClawSqliteTranscript( + openclawHome(), + listOpenClawAgents(), + sessionId, + ); + if (sqlite) { + return { + kind: "synthesized", + body: sqlite.eventJsonLines.join("\n") + "\n", + contentType: "application/x-ndjson", + extension: "jsonl", + }; + } const path = findOpenClawTranscript(sessionId); return path ? { kind: "file", path } : null; } @@ -131,13 +163,20 @@ export async function resolveDownloadSource( const { getDevinSessionLog } = await import("./devin-sessions"); const result = await getDevinSessionLog(sessionId); if (!result) return null; - const body = result.rawLines.map((r) => JSON.stringify(r)).join("\n") + "\n"; - return { kind: "synthesized", body, contentType: "application/x-ndjson", extension: "jsonl" }; + const body = + result.rawLines.map((r) => JSON.stringify(r)).join("\n") + "\n"; + return { + kind: "synthesized", + body, + contentType: "application/x-ndjson", + extension: "jsonl", + }; } if (cli === "antigravity") { // Antigravity (agy) writes real JSONL transcripts on disk — stream verbatim. - const { findAntigravityTranscript } = await import("./antigravity-sessions"); + const { findAntigravityTranscript } = + await import("./antigravity-sessions"); const path = findAntigravityTranscript(sessionId); return path ? { kind: "file", path } : null; } @@ -148,8 +187,14 @@ export async function resolveDownloadSource( const { getGooseSessionLog } = await import("./goose-sessions"); const result = await getGooseSessionLog(sessionId); if (!result) return null; - const body = result.rawLines.map((r) => JSON.stringify(r)).join("\n") + "\n"; - return { kind: "synthesized", body, contentType: "application/x-ndjson", extension: "jsonl" }; + const body = + result.rawLines.map((r) => JSON.stringify(r)).join("\n") + "\n"; + return { + kind: "synthesized", + body, + contentType: "application/x-ndjson", + extension: "jsonl", + }; } // Exhaustive — but TypeScript can't always see CliId is exhausted across the diff --git a/lib/openclaw-db.ts b/lib/openclaw-db.ts new file mode 100644 index 000000000..3615ccee4 --- /dev/null +++ b/lib/openclaw-db.ts @@ -0,0 +1,271 @@ +/** + * Read OpenClaw 2026.9.2+ transcripts from the per-agent SQLite databases at + * `agents//agent/openclaw-agent.sqlite`. + * + * `event_json` is the exact logical JSONL record, so the dashboard can feed it + * through the same parser and download format used by archived JSONL files. + */ +import { existsSync } from "node:fs"; +import { join } from "node:path"; +import { openSqliteReadonly, type SqliteReader } from "./sqlite-reader"; + +const DB_NAME = "openclaw-agent.sqlite"; + +export interface OpenClawSqliteSession { + sessionId: string; + agentId: string; + dbPath: string; + transcriptPath: string; + mtimeMs: number; + sizeBytes: number; + channel: string; + label?: string; + chatId?: string; + chatType?: string; +} + +export interface OpenClawSqliteTranscript { + agentId: string; + dbPath: string; + filePath: string; + rawLines: Record[]; + eventJsonLines: string[]; +} + +interface WindowRow { + session_id: string; + session_key: string; + updated_at: number; + transcript_updated_at: number | null; + ended_at: number | null; + channel: string | null; + chat_type: string | null; + display_name: string | null; + event_updated_at: number; + size_bytes: number; +} + +interface NodeRow { + current_session_id: string; + entry_json: string; + label: string | null; + display_name: string | null; +} + +interface ConversationRow { + session_id: string; + channel: string; + kind: string; + peer_id: string; + delivery_target: string; + label: string | null; +} + +interface EventRow { + event_json: string; +} + +interface SessionMeta { + channel?: string; + label?: string; + chatId?: string; + chatType?: string; +} + +function str(value: unknown): string | undefined { + return typeof value === "string" && value.length > 0 ? value : undefined; +} + +function parseObject(value: string): Record | undefined { + try { + const parsed: unknown = JSON.parse(value); + return parsed && typeof parsed === "object" && !Array.isArray(parsed) + ? (parsed as Record) + : undefined; + } catch { + return undefined; + } +} + +function syntheticPath(agentId: string, sessionId: string): string { + return `openclaw-sqlite://${encodeURIComponent(agentId)}/${sessionId}`; +} + +/** Existing per-agent SQLite databases, derived from the configured agents. */ +export function listOpenClawDatabases( + home: string, + agentIds: string[], +): Array<{ agentId: string; dbPath: string }> { + const out: Array<{ agentId: string; dbPath: string }> = []; + for (const agentId of agentIds) { + const dbPath = join(home, "agents", agentId, "agent", DB_NAME); + if (existsSync(dbPath)) out.push({ agentId, dbPath }); + } + return out; +} + +function readNodeMetadata(db: SqliteReader): Map { + const out = new Map(); + let rows: NodeRow[]; + try { + rows = db.query( + `SELECT current_session_id, entry_json, label, display_name + FROM session_nodes`, + ); + } catch { + return out; + } + for (const row of rows) { + const entry = parseObject(row.entry_json) ?? {}; + const origin = + entry.origin && + typeof entry.origin === "object" && + !Array.isArray(entry.origin) + ? (entry.origin as Record) + : {}; + out.set(row.current_session_id, { + channel: + str(entry.lastChannel) ?? str(origin.provider) ?? str(origin.surface), + label: + str(row.label) ?? + str(row.display_name) ?? + str(entry.displayName) ?? + str(origin.label), + chatId: str(entry.lastTo) ?? str(origin.from), + chatType: str(entry.chatType) ?? str(origin.chatType), + }); + } + return out; +} + +function readConversationMetadata(db: SqliteReader): Map { + const out = new Map(); + let rows: ConversationRow[]; + try { + rows = db.query( + `SELECT sc.session_id, c.channel, c.kind, c.peer_id, + c.delivery_target, c.label + FROM session_conversations sc + JOIN conversations c ON c.conversation_id = sc.conversation_id + ORDER BY CASE sc.role WHEN 'primary' THEN 0 ELSE 1 END, sc.last_seen_at DESC`, + ); + } catch { + return out; + } + for (const row of rows) { + if (out.has(row.session_id)) continue; + out.set(row.session_id, { + channel: str(row.channel), + label: str(row.label), + chatId: str(row.delivery_target) ?? str(row.peer_id), + chatType: str(row.kind), + }); + } + return out; +} + +/** List every transcript-bearing session in all per-agent databases. */ +export async function listOpenClawSqliteSessions( + home: string, + agentIds: string[], +): Promise { + const sessions: OpenClawSqliteSession[] = []; + for (const { agentId, dbPath } of listOpenClawDatabases(home, agentIds)) { + const db = await openSqliteReadonly(dbPath); + if (!db) continue; + try { + let rows: WindowRow[]; + try { + rows = db.query( + `SELECT w.session_id, w.session_key, w.updated_at, + w.transcript_updated_at, w.ended_at, w.channel, + w.chat_type, w.display_name, + COALESCE((SELECT MAX(e.created_at) + FROM transcript_events e + WHERE e.session_id = w.session_id), 0) AS event_updated_at, + COALESCE((SELECT SUM(length(e.event_json) + 1) + FROM transcript_events e + WHERE e.session_id = w.session_id), 0) AS size_bytes + FROM session_windows w + WHERE EXISTS (SELECT 1 FROM transcript_events e + WHERE e.session_id = w.session_id)`, + ); + } catch { + continue; + } + + const nodes = readNodeMetadata(db); + const conversations = readConversationMetadata(db); + for (const row of rows) { + const node = nodes.get(row.session_id); + const conversation = conversations.get(row.session_id); + sessions.push({ + sessionId: row.session_id, + agentId, + dbPath, + transcriptPath: syntheticPath(agentId, row.session_id), + mtimeMs: Math.max( + Number(row.updated_at) || 0, + Number(row.transcript_updated_at) || 0, + Number(row.ended_at) || 0, + Number(row.event_updated_at) || 0, + ), + sizeBytes: Number(row.size_bytes) || 0, + channel: + str(row.channel) ?? + conversation?.channel ?? + node?.channel ?? + "local", + label: str(row.display_name) ?? conversation?.label ?? node?.label, + chatId: conversation?.chatId ?? node?.chatId, + chatType: + str(row.chat_type) ?? conversation?.chatType ?? node?.chatType, + }); + } + } finally { + db.close(); + } + } + return sessions; +} + +/** Read one SQLite transcript by UUID. The first matching agent wins. */ +export async function readOpenClawSqliteTranscript( + home: string, + agentIds: string[], + sessionId: string, +): Promise { + for (const { agentId, dbPath } of listOpenClawDatabases(home, agentIds)) { + const db = await openSqliteReadonly(dbPath); + if (!db) continue; + try { + let rows: EventRow[]; + try { + rows = db.query( + `SELECT event_json + FROM transcript_events + WHERE session_id = ? + ORDER BY seq ASC`, + [sessionId], + ); + } catch { + continue; + } + if (rows.length === 0) continue; + const eventJsonLines = rows.map((row) => row.event_json); + const rawLines = eventJsonLines + .map(parseObject) + .filter((line): line is Record => line !== undefined); + return { + agentId, + dbPath, + filePath: syntheticPath(agentId, sessionId), + rawLines, + eventJsonLines, + }; + } finally { + db.close(); + } + } + return null; +} diff --git a/lib/openclaw-projects.ts b/lib/openclaw-projects.ts index a6f4976ba..d6dab53a3 100644 --- a/lib/openclaw-projects.ts +++ b/lib/openclaw-projects.ts @@ -1,18 +1,21 @@ /** * OpenClaw (openclaw gateway) session enumeration — AUDIT-ONLY. * - * Surfaces the on-disk transcripts (agents//sessions/.jsonl) as - * synthetic dashboard "projects" grouped by (agentId, channel). The per-agent - * `sessions.json` index maps sessionKey → {sessionId, timestamps}; we read it to - * recover the sessionKey (which encodes the channel for gateway sessions) and a - * reliable last-activity time. Verified live against openclaw v2026.7.1. + * Surfaces both legacy JSONL transcripts and OpenClaw 2026.9.2+ per-agent + * SQLite transcripts as synthetic dashboard "projects" grouped by + * (agentId, channel). */ import { readFileSync } from "node:fs"; import { join } from "node:path"; import { runtimeCache } from "./runtime-cache"; -import { listOpenClawAgents, listOpenClawTranscripts, openclawHome } from "./openclaw-sessions"; +import { + listOpenClawAgents, + listOpenClawTranscripts, + openclawHome, +} from "./openclaw-sessions"; import type { ProjectFolder, SessionFile } from "./projects"; import { formatDate } from "./format-date"; +import { listOpenClawSqliteSessions } from "./openclaw-db"; export interface OpenClawSessionRef { sessionId: string; @@ -51,7 +54,13 @@ function str(v: unknown): string | undefined { * than in the key — verified live against v2026.7.1. */ function readSessionsIndex(agentId: string): Map { const out = new Map(); - const indexPath = join(openclawHome(), "agents", agentId, "sessions", "sessions.json"); + const indexPath = join( + openclawHome(), + "agents", + agentId, + "sessions", + "sessions.json", + ); let raw: unknown; try { raw = JSON.parse(readFileSync(indexPath, "utf-8")); @@ -64,7 +73,9 @@ function readSessionsIndex(agentId: string): Map { const e = v as Record; const sessionId = str(e.sessionId); if (!sessionId) continue; - const origin = (e.origin && typeof e.origin === "object" ? e.origin : {}) as Record; + const origin = ( + e.origin && typeof e.origin === "object" ? e.origin : {} + ) as Record; const lastMs = typeof e.lastInteractionAt === "number" ? e.lastInteractionAt @@ -73,7 +84,8 @@ function readSessionsIndex(agentId: string): Map { : undefined; out.set(sessionId, { lastMs, - channel: str(e.lastChannel) ?? str(origin.provider) ?? str(origin.surface), + channel: + str(e.lastChannel) ?? str(origin.provider) ?? str(origin.surface), chatType: str(e.chatType) ?? str(origin.chatType), label: str(origin.label), chatId: str(e.lastTo) ?? str(origin.from), @@ -86,7 +98,7 @@ function readSessionsIndex(agentId: string): Map { export async function getOpenClawSessions(): Promise { const transcripts = listOpenClawTranscripts(); const indexByAgent = new Map>(); - const refs: OpenClawSessionRef[] = []; + const refs = new Map(); for (const t of transcripts) { let idx = indexByAgent.get(t.agentId); if (!idx) { @@ -94,7 +106,7 @@ export async function getOpenClawSessions(): Promise { indexByAgent.set(t.agentId, idx); } const meta = idx.get(t.sessionId); - refs.push({ + refs.set(`${t.agentId}\0${t.sessionId}`, { sessionId: t.sessionId, agentId: t.agentId, // Gateway sessions group by channel; CLI/local runs have none. @@ -107,8 +119,18 @@ export async function getOpenClawSessions(): Promise { sizeBytes: t.sizeBytes, }); } - refs.sort((a, b) => b.mtimeMs - a.mtimeMs); - return refs; + + // SQLite is the live source on OpenClaw 2026.9.2+. Insert it second so it + // replaces an archived JSONL copy of the same agent/session. + const sqliteSessions = await listOpenClawSqliteSessions( + openclawHome(), + listOpenClawAgents(), + ); + for (const s of sqliteSessions) { + refs.set(`${s.agentId}\0${s.sessionId}`, s); + } + + return [...refs.values()].sort((a, b) => b.mtimeMs - a.mtimeMs); } export const getCachedOpenClawSessions = runtimeCache(getOpenClawSessions, 2); @@ -145,7 +167,9 @@ export interface OpenClawNameSplit { * form (`agentId: null`), so links shared before agents became part of the name * keep resolving — to every agent on that channel, which is what they meant. */ -export function openClawProjectNameCandidates(name: string): OpenClawNameSplit[] { +export function openClawProjectNameCandidates( + name: string, +): OpenClawNameSplit[] { if (!name.startsWith("openclaw-")) return []; const rest = name.slice("openclaw-".length); if (!rest) return []; @@ -163,7 +187,9 @@ export function openClawProjectNameCandidates(name: string): OpenClawNameSplit[] } /** The best-guess split for a project name — `null` if it isn't an OpenClaw one. */ -export function parseOpenClawProjectName(name: string): OpenClawNameSplit | null { +export function parseOpenClawProjectName( + name: string, +): OpenClawNameSplit | null { return openClawProjectNameCandidates(name)[0] ?? null; } @@ -190,7 +216,12 @@ export async function getOpenClawProjects(): Promise { prev.latest = Math.max(prev.latest, s.mtimeMs); prev.count += 1; } else { - groups.set(key, { agentId: s.agentId, channel: s.channel, latest: s.mtimeMs, count: 1 }); + groups.set(key, { + agentId: s.agentId, + channel: s.channel, + latest: s.mtimeMs, + count: 1, + }); } } const out: ProjectFolder[] = []; @@ -228,7 +259,9 @@ export async function getOpenClawSessionsByEncodedName( const sessions = await getOpenClawSessions(); const matching = (split: OpenClawNameSplit) => sessions.filter( - (s) => s.channel === split.channel && (split.agentId === null || s.agentId === split.agentId), + (s) => + s.channel === split.channel && + (split.agentId === null || s.agentId === split.agentId), ); // Take the first split that actually owns sessions — length alone picks the @@ -250,7 +283,10 @@ export async function getOpenClawSessionsByEncodedName( return { // Legacy channel-only names keep their old cwd label, since they really do // span every agent on that channel. - cwd: agentId === null ? `openclaw:${channel}` : openClawProjectPath(agentId, channel), + cwd: + agentId === null + ? `openclaw:${channel}` + : openClawProjectPath(agentId, channel), sessions: matched.map((s) => { const lastModified = new Date(s.mtimeMs); return { diff --git a/lib/openclaw-sessions.ts b/lib/openclaw-sessions.ts index afc81b35c..875e80040 100644 --- a/lib/openclaw-sessions.ts +++ b/lib/openclaw-sessions.ts @@ -1,11 +1,9 @@ /** * OpenClaw (openclaw gateway) session transcript loader + parser. * - * AUDIT-ONLY (Pillar 2). OpenClaw writes one JSONL transcript per session at - * `~/.openclaw/agents//sessions/.jsonl` (sessionId is a - * UUID), alongside a much larger `.trajectory.jsonl` OTel trace we - * IGNORE, and a `sessions.json` index keyed by sessionKey. Verified live - * against openclaw v2026.7.1. + * Legacy OpenClaw writes one JSONL transcript per session under `sessions/`; + * 2026.9.2+ stores the same logical records as `event_json` rows in each + * agent's `agent/openclaw-agent.sqlite`. SQLite is preferred when both exist. * * The transcript is type-discriminated JSONL: * {type:"session", cwd, …} — header (carries cwd) @@ -18,8 +16,8 @@ * `toolResult` by `toolCallId` (mirrors lib/hermes-sessions.ts) and is PURE, so * it is unit-testable with plain line objects. * - * Home override: set `OPENCLAW_HOME` (used by tests / to point at a copied - * gateway config dir). + * Home override: `OPENCLAW_STATE_DIR`, then `OPENCLAW_HOME` (used by tests / + * to point at a copied gateway config dir). */ import { readFile } from "node:fs/promises"; import { readdirSync, statSync } from "node:fs"; @@ -39,13 +37,18 @@ import { type LogSource, } from "./log-entries"; import { formatDuration } from "./format-duration"; +import { readOpenClawSqliteTranscript } from "./openclaw-db"; /** OpenClaw sessions are stored under UUID filenames. */ export const OPENCLAW_SESSION_ID_RE = /^[0-9a-fA-F-]{36}$/; -/** Absolute path to OpenClaw's config home (override with OPENCLAW_HOME). */ +/** Absolute path to OpenClaw's state home. */ export function openclawHome(): string { - return process.env.OPENCLAW_HOME || join(homedir(), ".openclaw"); + return ( + process.env.OPENCLAW_STATE_DIR || + process.env.OPENCLAW_HOME || + join(homedir(), ".openclaw") + ); } // ── Parsing helpers ── @@ -60,7 +63,11 @@ function extractText(content: unknown): string { if (typeof content === "string") return content; if (Array.isArray(content)) { return content - .map((c) => (isPlainObject(c) && typeof c.text === "string" ? (c.text as string) : "")) + .map((c) => + isPlainObject(c) && typeof c.text === "string" + ? (c.text as string) + : "", + ) .filter(Boolean) .join("\n"); } @@ -134,7 +141,10 @@ export function openclawLinesToLogEntries( if (b.type === "text" && typeof b.text === "string") { blocks.push({ type: "text", text: b.text }); } else if (b.type === "toolCall") { - const id = typeof b.id === "string" ? b.id : `${String(b.name ?? "tool")}-${blocks.length}`; + const id = + typeof b.id === "string" + ? b.id + : `${String(b.name ?? "tool")}-${blocks.length}`; const name = typeof b.name === "string" ? b.name : "tool"; const input = isPlainObject(b.arguments) ? b.arguments : {}; const block: ToolUseBlock = { type: "tool_use", id, name, input }; @@ -151,17 +161,23 @@ export function openclawLinesToLogEntries( entries.push({ type: "assistant", ...base, - message: { role: "assistant", content: blocks, model: typeof m.model === "string" ? m.model : undefined }, + message: { + role: "assistant", + content: blocks, + model: typeof m.model === "string" ? m.model : undefined, + }, } satisfies AssistantEntry); continue; } if (role === "toolResult") { - const callId = typeof m.toolCallId === "string" ? m.toolCallId : undefined; + const callId = + typeof m.toolCallId === "string" ? m.toolCallId : undefined; const block = callId ? toolUseById.get(callId) : undefined; if (block) { const details = isPlainObject(m.details) ? m.details : undefined; - const startMs = (callId && toolUseStartMs.get(callId)) || date.getTime(); + const startMs = + (callId && toolUseStartMs.get(callId)) || date.getTime(); const durationMs = details && typeof details.durationMs === "number" ? details.durationMs @@ -238,13 +254,20 @@ export function listOpenClawTranscripts(): OpenClawTranscriptFile[] { } for (const file of files) { // Only `.jsonl` — not `.trajectory.jsonl` / `.trajectory-path.json`. - if (!file.endsWith(".jsonl") || file.endsWith(".trajectory.jsonl")) continue; + if (!file.endsWith(".jsonl") || file.endsWith(".trajectory.jsonl")) + continue; const sessionId = file.slice(0, -".jsonl".length); if (!OPENCLAW_SESSION_ID_RE.test(sessionId)) continue; const transcriptPath = join(sessionsDir, file); try { const st = statSync(transcriptPath); - out.push({ agentId, sessionId, transcriptPath, mtimeMs: st.mtimeMs, sizeBytes: st.size }); + out.push({ + agentId, + sessionId, + transcriptPath, + mtimeMs: st.mtimeMs, + sizeBytes: st.size, + }); } catch { // skip unreadable } @@ -276,6 +299,36 @@ export interface OpenClawSessionLogData { export async function getOpenClawSessionLog( sessionId: string, ): Promise { + if (!OPENCLAW_SESSION_ID_RE.test(sessionId)) return null; + + // New OpenClaw versions keep live transcripts in SQLite. Prefer that copy + // when an archived JSONL with the same UUID also exists. + const sqlite = await readOpenClawSqliteTranscript( + openclawHome(), + listOpenClawAgents(), + sessionId, + ); + if (sqlite) { + const entries = openclawLinesToLogEntries(sqlite.rawLines, "session"); + let cwd: string | undefined; + for (const line of sqlite.rawLines) { + if ( + line.type === "session" && + typeof line.cwd === "string" && + line.cwd.length > 0 + ) { + cwd = line.cwd; + break; + } + } + return { + entries, + rawLines: sqlite.rawLines, + cwd, + filePath: sqlite.filePath, + }; + } + const filePath = findOpenClawTranscript(sessionId); if (!filePath) return null; let content: string; @@ -289,7 +342,12 @@ export async function getOpenClawSessionLog( // cwd lives on the `type:"session"` header line. let cwd: string | undefined; for (const line of rawLines) { - if (isPlainObject(line) && line.type === "session" && typeof line.cwd === "string" && line.cwd.length > 0) { + if ( + isPlainObject(line) && + line.type === "session" && + typeof line.cwd === "string" && + line.cwd.length > 0 + ) { cwd = line.cwd; break; } diff --git a/package.json b/package.json index baedca443..973f13176 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "failproofai", - "version": "1.0.4-beta.7", + "version": "1.0.4-beta.8", "description": "Observability and enforcement for AI agent harnesses. 39 built-in policies hooked into 12 of them — Claude Code, Codex, Cursor, Hermes, OpenClaw and more — blocking the tool call before it runs. Local dashboard included, no account needed.", "bin": { "failproofai": "./dist/cli.mjs", From f52f11bc3f168b701647ba5125cbea00bda8704d Mon Sep 17 00:00:00 2001 From: chhhee10 Date: Sat, 12 Sep 2026 01:47:19 +0530 Subject: [PATCH 2/3] chore: prepare 1.0.4 stable release --- CHANGELOG.md | 11 +++++++++++ Cargo.lock | 6 +++--- Cargo.toml | 2 +- package.json | 2 +- 4 files changed, 16 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9d655262a..1a22aaf4d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## 1.0.4 — 2026-09-12 + +The stable release of the OpenClaw 2026.9 compatibility work shipped through +`1.0.4-beta.8`. + +### Fixes + +- Restore OpenClaw 2026.9.2+ observability after live transcripts moved from per-session JSONL files into per-agent SQLite databases. `failproofaid` now discovers every agent profile and configured extra path, tails each SQLite transcript incrementally, handles transcript rewrites without duplicating delivery, and retains legacy JSONL compatibility. + +- Restore OpenClaw sessions in the local dashboard's Projects view and session viewer. SQLite and legacy sessions are merged per agent, live SQLite copies win over archived duplicates, missing channels group under `local`, and downloads export the original `event_json` records as JSONL. + ## 1.0.4-beta.8 — 2026-09-11 ### Fixes diff --git a/Cargo.lock b/Cargo.lock index 7e5d8da4c..5d987ae8d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -202,7 +202,7 @@ dependencies = [ [[package]] name = "failproofaid" -version = "1.0.4-beta.8" +version = "1.0.4" dependencies = [ "fpai-collect", "fpai-ipc", @@ -265,7 +265,7 @@ dependencies = [ [[package]] name = "fpai-collect" -version = "1.0.4-beta.8" +version = "1.0.4" dependencies = [ "notify", "reqwest", @@ -280,7 +280,7 @@ dependencies = [ [[package]] name = "fpai-ipc" -version = "1.0.4-beta.8" +version = "1.0.4" dependencies = [ "libc", "proptest", diff --git a/Cargo.toml b/Cargo.toml index 31df1bcb4..d23c078ec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "3" members = ["crates/*"] [workspace.package] -version = "1.0.4-beta.8" +version = "1.0.4" edition = "2024" license-file = "LICENSE" repository = "https://github.com/FailproofAI/failproofai" diff --git a/package.json b/package.json index 973f13176..77a3d6fb5 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "failproofai", - "version": "1.0.4-beta.8", + "version": "1.0.4", "description": "Observability and enforcement for AI agent harnesses. 39 built-in policies hooked into 12 of them — Claude Code, Codex, Cursor, Hermes, OpenClaw and more — blocking the tool call before it runs. Local dashboard included, no account needed.", "bin": { "failproofai": "./dist/cli.mjs", From f190591028c3ea866f5dd4d12d1c2a1c28205b44 Mon Sep 17 00:00:00 2001 From: chhhee10 Date: Sat, 12 Sep 2026 03:08:32 +0530 Subject: [PATCH 3/3] fix(openclaw): deliver pre-tool instructions --- CHANGELOG.md | 2 + CLAUDE.md | 8 +- .../openclaw-instruct-retry-gate.test.ts | 90 +++++++++++++++++++ .../hooks/openclaw-invoice-instruct.test.ts | 64 +++++++++++++ __tests__/hooks/policy-evaluator.test.ts | 8 +- openclaw-plugin/index.js | 16 ++-- openclaw-plugin/instruct-retry-gate.js | 89 ++++++++++++++++++ src/hooks/policy-evaluator.ts | 25 +++++- 8 files changed, 287 insertions(+), 15 deletions(-) create mode 100644 __tests__/hooks/openclaw-instruct-retry-gate.test.ts create mode 100644 __tests__/hooks/openclaw-invoice-instruct.test.ts create mode 100644 openclaw-plugin/instruct-retry-gate.js diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a22aaf4d..74bc34f2a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ The stable release of the OpenClaw 2026.9 compatibility work shipped through - Restore OpenClaw sessions in the local dashboard's Projects view and session viewer. SQLite and legacy sessions are merged per agent, live SQLite copies win over archived duplicates, missing channels group under `local`, and downloads export the original `event_json` records as JSONL. +- Deliver OpenClaw `PreToolUse` instructions to the agent through its model-visible tool rejection reason. The first matching `instruct()` temporarily interrupts the tool call, while a session-and-policy-scoped retry window lets the agent proceed after following the guidance. Policy source, tool canonicalization, and transcript ingestion remain unchanged. + ## 1.0.4-beta.8 — 2026-09-11 ### Fixes diff --git a/CLAUDE.md b/CLAUDE.md index caa646a94..1729326f6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -477,9 +477,11 @@ maps in `types.ts` (single source of truth). `stopHookActive`, ≈ Claude's Stop payload), so the 5 `require-*-before-stop` builtins **enforce** on OpenClaw — a deny becomes a `{action:"revise"}` that re-runs the turn (unlike Hermes, which has no Stop event at all). **Instruct** -degrades to allow + stderr note on non-Stop events (no additional-context -channel); on Stop it emits the MANDATORY-ACTION deny so the revise loop carries -the directive. **Omitted hooks:** `agent_end` (would double-fire Stop) and +on `PreToolUse` uses a model-visible `blockReason` to interrupt the first +matching tool attempt, then permits retries from that session/policy for five +minutes; other non-Stop events still degrade to allow + stderr note. On Stop it +emits the MANDATORY-ACTION deny so the revise loop carries the directive. +**Omitted hooks:** `agent_end` (would double-fire Stop) and `message_sending` (outbound-message cancel gate — an OpenClaw-only capability, deferred). diff --git a/__tests__/hooks/openclaw-instruct-retry-gate.test.ts b/__tests__/hooks/openclaw-instruct-retry-gate.test.ts new file mode 100644 index 000000000..7849816f6 --- /dev/null +++ b/__tests__/hooks/openclaw-instruct-retry-gate.test.ts @@ -0,0 +1,90 @@ +// @vitest-environment node +import { describe, expect, it } from "vitest"; +import { + createInstructRetryGate, + mapBeforeToolVerdict, +} from "../../openclaw-plugin/instruct-retry-gate.js"; + +describe("OpenClaw instruct retry gate", () => { + it("interrupts the first instruction and permits retries during the window", () => { + let now = 1_000; + const gate = createInstructRetryGate({ windowMs: 300_000, now: () => now }); + const verdict = { + permission: "instruct", + reason: "recover once", + policyName: "failproofai/warn-invoice-self-resolution", + }; + const ctx = { sessionKey: "invoice-session" }; + + expect(gate.shouldInterrupt(verdict, {}, ctx)).toBe(true); + expect(gate.shouldInterrupt(verdict, {}, ctx)).toBe(false); + + now += 300_001; + expect(gate.shouldInterrupt(verdict, {}, ctx)).toBe(true); + }); + + it("keeps sessions and policies independent", () => { + const gate = createInstructRetryGate(); + const invoice = { permission: "instruct", reason: "recover", policyName: "invoice" }; + const security = { permission: "instruct", reason: "review", policyName: "security" }; + + expect(gate.shouldInterrupt(invoice, {}, { sessionKey: "one" })).toBe(true); + expect(gate.shouldInterrupt(invoice, {}, { sessionKey: "one" })).toBe(false); + expect(gate.shouldInterrupt(security, {}, { sessionKey: "one" })).toBe(true); + expect(gate.shouldInterrupt(invoice, {}, { sessionKey: "two" })).toBe(true); + }); + + it("keeps separate OpenClaw runs independent within one session", () => { + const gate = createInstructRetryGate(); + const verdict = { permission: "instruct", reason: "recover", policyName: "invoice" }; + + expect(gate.shouldInterrupt(verdict, {}, { sessionKey: "one", runId: "run-a" })).toBe(true); + expect(gate.shouldInterrupt(verdict, {}, { sessionKey: "one", runId: "run-a" })).toBe(false); + expect(gate.shouldInterrupt(verdict, {}, { sessionKey: "one", runId: "run-b" })).toBe(true); + }); + + it("does not share a retry window between anonymous invocations", () => { + const gate = createInstructRetryGate(); + const verdict = { permission: "instruct", reason: "recover", policyName: "invoice" }; + + expect(gate.shouldInterrupt(verdict, {}, {})).toBe(true); + expect(gate.shouldInterrupt(verdict, {}, {})).toBe(true); + }); + + it("clears the retry window when the session ends", () => { + const gate = createInstructRetryGate(); + const verdict = { permission: "instruct", reason: "recover", policyName: "invoice" }; + const ctx = { sessionKey: "invoice-session", runId: "invoice-run" }; + + expect(gate.shouldInterrupt(verdict, {}, ctx)).toBe(true); + expect(gate.shouldInterrupt(verdict, {}, ctx)).toBe(false); + gate.clear({}, { sessionKey: "invoice-session" }); + expect(gate.shouldInterrupt(verdict, {}, ctx)).toBe(true); + }); + + it("maps deny permanently and instruct to a one-shot model-visible rejection", () => { + const gate = createInstructRetryGate(); + const ctx = { sessionKey: "invoice-session" }; + const deny = { permission: "deny", reason: "never send this" }; + const instruct = { + permission: "instruct", + reason: "perform one more recovery pass", + policyName: "invoice", + }; + + expect(mapBeforeToolVerdict(deny, {}, ctx, gate)).toEqual({ + block: true, + blockReason: "never send this", + }); + expect(mapBeforeToolVerdict(deny, {}, ctx, gate)).toEqual({ + block: true, + blockReason: "never send this", + }); + expect(mapBeforeToolVerdict(instruct, {}, ctx, gate)).toEqual({ + block: true, + blockReason: "perform one more recovery pass", + }); + expect(mapBeforeToolVerdict(instruct, {}, ctx, gate)).toBeUndefined(); + expect(mapBeforeToolVerdict({ permission: "allow" }, {}, ctx, gate)).toBeUndefined(); + }); +}); diff --git a/__tests__/hooks/openclaw-invoice-instruct.test.ts b/__tests__/hooks/openclaw-invoice-instruct.test.ts new file mode 100644 index 000000000..74bf683b5 --- /dev/null +++ b/__tests__/hooks/openclaw-invoice-instruct.test.ts @@ -0,0 +1,64 @@ +// @vitest-environment node +import { beforeEach, describe, expect, it } from "vitest"; +import { evaluatePolicies } from "../../src/hooks/policy-evaluator"; +import { clearPolicies, registerPolicy } from "../../src/hooks/policy-registry"; + +const INVOICE_WORKSPACE_RE = /(?:^|\/)invoice(?:\/|$)/; +const NEEDS_HUMAN_MAPPING_RE = /\bneeds_human_mapping\b/; +const FINANCE_SEND_RE = /\bopenclaw\s+message\s+send\b[\s\S]*?(?:--channel\s+slack|-c\s+slack)[\s\S]*?(?:-t\s+C0AVC7K2XDM|--target\s+C0AVC7K2XDM)/i; +const ESCALATION_STATUS_RE = /(?:needs\s+review|couldn['’]?t\s+file)/i; + +describe("Zaum OpenClaw invoice instruct policy", () => { + beforeEach(() => { + clearPolicies(); + registerPolicy("warn-invoice-self-resolution", "invoice recovery", (ctx) => { + if (ctx.cli !== "openclaw") return { decision: "allow" }; + if (!INVOICE_WORKSPACE_RE.test(String(ctx.session?.cwd ?? ""))) return { decision: "allow" }; + + const toolInput = JSON.stringify(ctx.toolInput ?? {}); + const isPreQuestionCheckpoint = NEEDS_HUMAN_MAPPING_RE.test(toolInput); + const isFinanceEscalation = + ctx.toolName === "Bash" && + FINANCE_SEND_RE.test(toolInput) && + ESCALATION_STATUS_RE.test(toolInput); + + if (!isPreQuestionCheckpoint && !isFinanceEscalation) return { decision: "allow" }; + return { decision: "instruct", reason: "make one more evidence-backed recovery pass" }; + }, { events: ["PreToolUse"] }); + }); + + it("emits an OpenClaw instruct verdict for Zaum's observed Finance escalation shape", async () => { + const result = await evaluatePolicies("PreToolUse", { + tool_name: "Bash", + tool_input: { + command: "openclaw message send --channel slack -t C0AVC7K2XDM --message 'Needs review: couldn’t file invoice'", + }, + cwd: "/Users/atlas/.openclaw/workspace/invoice", + }, { + cli: "openclaw", + cwd: "/Users/atlas/.openclaw/workspace/invoice", + }); + + expect(result.decision).toBe("instruct"); + expect(JSON.parse(result.stdout)).toMatchObject({ + permission: "instruct", + policyName: "failproofai/warn-invoice-self-resolution", + }); + }); + + it("does not affect unrelated OpenClaw Slack sends", async () => { + const result = await evaluatePolicies("PreToolUse", { + tool_name: "Bash", + tool_input: { + command: "openclaw message send --channel slack -t C0AVC7K2XDM --message 'Invoice filed successfully'", + }, + cwd: "/Users/atlas/.openclaw/workspace/invoice", + }, { + cli: "openclaw", + cwd: "/Users/atlas/.openclaw/workspace/invoice", + }); + + expect(result.decision).toBe("allow"); + expect(result.stdout).toBe(""); + }); +}); diff --git a/__tests__/hooks/policy-evaluator.test.ts b/__tests__/hooks/policy-evaluator.test.ts index ffc457710..f5a2586a3 100644 --- a/__tests__/hooks/policy-evaluator.test.ts +++ b/__tests__/hooks/policy-evaluator.test.ts @@ -356,7 +356,7 @@ describe("hooks/policy-evaluator", () => { expect(String(parsed.reason)).toContain("commit first"); }); - it("OpenClaw instruct on Stop emits MANDATORY-ACTION deny (revise); on tool events degrades to allow + note", async () => { + it("OpenClaw instruct uses revise on Stop and a one-shot shim verdict on PreToolUse", async () => { registerPolicy("advise-stop", "desc", () => ({ decision: "instruct", reason: "run tests" }), { events: ["Stop"], }); @@ -373,9 +373,11 @@ describe("hooks/policy-evaluator", () => { const pre = await evaluatePolicies("PreToolUse", { tool_name: "Bash" }, { cli: "openclaw" }); expect(pre.decision).toBe("instruct"); const preParsed = JSON.parse(pre.stdout) as Record; - expect(preParsed.permission).toBe("allow"); // does NOT block — no context channel on tool events + expect(preParsed.permission).toBe("instruct"); expect(preParsed.reason).toContain("prefer git mv"); - expect(pre.stderr).toContain("prefer git mv"); + expect(preParsed.policyName).toBe("failproofai/advise-tool"); + expect(preParsed.policyNames).toEqual(["failproofai/advise-tool"]); + expect(pre.stderr).toBe(""); }); it("Cursor SubagentStop + instruct emits {followup_message} JSON (parity with Stop branch)", async () => { diff --git a/openclaw-plugin/index.js b/openclaw-plugin/index.js index 869d8a24c..461753bec 100644 --- a/openclaw-plugin/index.js +++ b/openclaw-plugin/index.js @@ -7,7 +7,9 @@ * file-based "internal hooks" are observation-only — and forwards each to the * failproofai binary as `failproofai --hook --cli openclaw`. failproofai * prints a flat `{permission, reason}` verdict on stdout; this shim maps it to - * each hook's native return shape. + * each hook's native return shape. On before_tool_call, an `instruct` verdict + * uses a one-shot blockReason as OpenClaw's model-visible instruction channel; + * retries from the same session/policy are allowed for a short window. * * Marker comment for failproofai's installer detection (do not remove): * __failproofai_hook__: true @@ -34,10 +36,12 @@ import { spawn } from "node:child_process"; import { existsSync } from "node:fs"; import { resolve, dirname } from "node:path"; import { fileURLToPath } from "node:url"; +import { createInstructRetryGate, mapBeforeToolVerdict } from "./instruct-retry-gate.js"; const HERE = dirname(fileURLToPath(import.meta.url)); const DIST_BIN = resolve(HERE, "..", "dist", "cli.mjs"); const SRC_BIN = resolve(HERE, "..", "bin", "failproofai.mjs"); +const instructRetryGate = createInstructRetryGate(); function resolveSpawn() { if (process.env.FAILPROOFAI_BINARY_OVERRIDE) { @@ -136,7 +140,9 @@ export default definePluginEntry({ name: "failproofai", description: "Real-time policy enforcement for OpenClaw by failproofai", register(api) { - // before_tool_call → PreToolUse. Full deny: return {block:true, blockReason}. + // before_tool_call → PreToolUse. Deny always blocks. Instruct blocks the + // first matching attempt only, using blockReason to give the model the + // instruction, then permits retries from that session/policy for 5 minutes. api.on( "before_tool_call", async (payload, ctx) => { @@ -147,10 +153,7 @@ export default definePluginEntry({ tool_input: p.params, hook_event_name: "before_tool_call", }); - if (verdict.permission === "deny") { - return { block: true, blockReason: verdict.reason || "Blocked by failproofai" }; - } - return undefined; + return mapBeforeToolVerdict(verdict, payload, ctx, instructRetryGate); }, { priority: 100, timeoutMs: 60_000 }, ); @@ -214,6 +217,7 @@ export default definePluginEntry({ reason: p.reason, hook_event_name: ev, }); + if (ev === "session_end") instructRetryGate.clear(payload, ctx); return undefined; }, { priority: 100, timeoutMs: 60_000 }, diff --git a/openclaw-plugin/instruct-retry-gate.js b/openclaw-plugin/instruct-retry-gate.js new file mode 100644 index 000000000..753f9a243 --- /dev/null +++ b/openclaw-plugin/instruct-retry-gate.js @@ -0,0 +1,89 @@ +/** + * OpenClaw has no non-blocking context channel on before_tool_call. To deliver + * an instruct verdict to the model, the shim rejects the first matching call + * with the instruction as blockReason. Calls from the same session and policy + * are then allowed for a short window so an advisory policy cannot trap the + * agent in an endless reject/retry loop. + */ + +export const DEFAULT_INSTRUCT_RETRY_WINDOW_MS = 5 * 60 * 1000; + +function policyKey(verdict) { + if (Array.isArray(verdict?.policyNames) && verdict.policyNames.length > 0) { + return [...verdict.policyNames].map(String).sort().join(","); + } + if (verdict?.policyName) return String(verdict.policyName); + return String(verdict?.reason ?? "instruct"); +} + +function identity(payload, ctx) { + const p = payload || {}; + const c = ctx || {}; + const session = c.sessionKey ?? p.sessionKey ?? c.sessionId ?? p.sessionId; + const run = c.runId ?? p.runId; + if (session !== undefined && session !== null && session !== "") { + return { + scope: `session:${String(session)}${run ? `:run:${String(run)}` : ""}`, + sessionPrefix: `session:${String(session)}`, + }; + } + if (run !== undefined && run !== null && run !== "") { + return { scope: `run:${String(run)}`, sessionPrefix: null }; + } + return { scope: null, sessionPrefix: null }; +} + +export function createInstructRetryGate({ + windowMs = DEFAULT_INSTRUCT_RETRY_WINDOW_MS, + now = () => Date.now(), +} = {}) { + const instructedUntil = new Map(); + + function prune(at) { + for (const [key, expiresAt] of instructedUntil) { + if (expiresAt <= at) instructedUntil.delete(key); + } + } + + return { + /** True means interrupt this attempt and show the instruction to the model. */ + shouldInterrupt(verdict, payload, ctx) { + const at = now(); + prune(at); + const { scope } = identity(payload, ctx); + // Never let one anonymous hook invocation suppress another unrelated + // session. OpenClaw normally supplies sessionKey/runId; if it does not, + // fail safe by delivering the instruction on every matching attempt. + if (!scope) return true; + const key = `${scope}\0${policyKey(verdict)}`; + const expiresAt = instructedUntil.get(key); + if (expiresAt !== undefined && expiresAt > at) return false; + instructedUntil.set(key, at + windowMs); + return true; + }, + + clear(payload, ctx) { + const { sessionPrefix, scope } = identity(payload, ctx); + const prefix = sessionPrefix ?? scope; + if (!prefix) return; + for (const key of instructedUntil.keys()) { + if (key.startsWith(`${prefix}\0`) || key.startsWith(`${prefix}:run:`)) { + instructedUntil.delete(key); + } + } + }, + }; +} + +/** Map failproofai's flat verdict to OpenClaw's before_tool_call result. */ +export function mapBeforeToolVerdict(verdict, payload, ctx, retryGate) { + if (verdict?.permission === "deny") { + return { block: true, blockReason: verdict.reason || "Blocked by failproofai" }; + } + if (verdict?.permission !== "instruct") return undefined; + if (!retryGate.shouldInterrupt(verdict, payload, ctx)) return undefined; + return { + block: true, + blockReason: verdict.reason || "Instruction from failproofai: reconsider this action before retrying", + }; +} diff --git a/src/hooks/policy-evaluator.ts b/src/hooks/policy-evaluator.ts index 42a979707..fc104c8be 100644 --- a/src/hooks/policy-evaluator.ts +++ b/src/hooks/policy-evaluator.ts @@ -798,9 +798,12 @@ export async function evaluatePolicies( // OpenClaw: Stop (before_agent_finalize) can force a revise, so we emit the // MANDATORY ACTION wording as a flat deny — the shim maps it to - // {action:"revise", reason}. Every other event lacks an additional-context - // channel (before_tool_call's return is {params,block,blockReason} only), so - // instruct degrades to allow + stderr note, like Hermes. + // {action:"revise", reason}. PreToolUse has no non-blocking context channel, + // but a rejected tool's blockReason is model-visible. Preserve `instruct` + // in the wire verdict so the shim can interrupt the first matching attempt, + // deliver the instruction through blockReason, and allow a retry. Other + // events still degrade to allow + stderr because their return channels + // cannot carry an instruction to the model. if (session?.cli === "openclaw") { if (eventType === "Stop") { const policyAttribution = policyNames.length === 1 @@ -817,6 +820,22 @@ export async function evaluatePolicies( decision: "instruct", }; } + if (eventType === "PreToolUse") { + return { + exitCode: 0, + stdout: JSON.stringify({ + permission: "instruct", + reason: `Instruction from failproofai: ${combined}`, + policyName: policyNames[0], + policyNames, + }), + stderr: "", + policyName: policyNames[0], + policyNames, + reason: combined, + decision: "instruct", + }; + } const stderrMsg = instructEntries .map((e) => `[failproofai] ${e.policyName}: ${e.reason}`) .join("\n");