From ef1cccf849bca6e6f829c58f6eeac4e5f75bb9b9 Mon Sep 17 00:00:00 2001 From: Emanuele Fenocchi Date: Tue, 14 Jul 2026 17:11:50 +0000 Subject: [PATCH 1/4] fix(capture): make session-event INSERT idempotent to stop duplicate rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Session events were written with a plain `INSERT ... VALUES` and the sessions table has no UNIQUE constraint on `id`. When the API retry loop re-sends an insert that already committed but returned a transient 5xx (502/503), a duplicate row is created — the ~17% duplication observed in production during the 2026-07-10 gateway-degradation window. Route every capture INSERT through a shared `buildDirectSessionInsertSql` helper that emits `INSERT ... SELECT ... WHERE NOT EXISTS (id = ...)`, making a re-send a no-op. Verified lag-safe against the real backend: a rapid re-send inside the ~5s read-your-writes window yields exactly one row. Applied across all runtimes — claude/codex/cursor/hermes capture, codex stop, and openclaw — via one helper, removing the duplicated inline SQL at the same time. Tests: real e2e against the backend, source + bundle-guard tests, and updates to the three existing tests that scraped the now-centralized column list. --- harnesses/openclaw/src/index.ts | 19 +++- src/hooks/capture.ts | 20 +++- src/hooks/codex/capture.ts | 20 +++- src/hooks/codex/stop.ts | 20 +++- src/hooks/cursor/capture.ts | 20 +++- src/hooks/hermes/capture.ts | 20 +++- src/hooks/shared/session-insert-sql.ts | 55 +++++++++++ .../plugin-version-resolution.test.ts | 9 +- tests/claude-code/session-insert-sql.test.ts | 98 +++++++++++++++++++ tests/openclaw/openclaw-embed-bundle.test.ts | 10 +- .../agent-sessions-insert-schema.test.ts | 9 +- 11 files changed, 259 insertions(+), 41 deletions(-) create mode 100644 src/hooks/shared/session-insert-sql.ts create mode 100644 tests/claude-code/session-insert-sql.test.ts diff --git a/harnesses/openclaw/src/index.ts b/harnesses/openclaw/src/index.ts index 23363c707..32be2c8c9 100644 --- a/harnesses/openclaw/src/index.ts +++ b/harnesses/openclaw/src/index.ts @@ -66,6 +66,7 @@ import { readVirtualPathContent } from "../../../src/hooks/virtual-table-query.j // message_embedding (today's behavior, preserved on every failure mode). import { tryEmbedStandalone, _setSpawnImpl } from "../../../src/embeddings/standalone-embed-client.js"; import { embeddingSqlLiteral } from "../../../src/embeddings/sql.js"; +import { buildDirectSessionInsertSql } from "../../../src/hooks/shared/session-insert-sql.js"; // Resolve sibling skillify-worker.js path at runtime via import.meta.url. The // openclaw plugin is bundled to harnesses/openclaw/dist/index.js, then installed to // ~/.openclaw/extensions/hivemind/dist/index.js by install-openclaw.ts. The @@ -1536,10 +1537,20 @@ export default definePluginEntry({ const embedding = await tryEmbedStandalone(line, "document"); const embeddingSql = embeddingSqlLiteral(embedding); - const insertSql = - `INSERT INTO "${sessionsTable}" (id, path, filename, message, message_embedding, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + - `VALUES ('${crypto.randomUUID()}', '${sqlStr(sessionPath)}', '${sqlStr(filename)}', '${jsonForSql}'::jsonb, ${embeddingSql}, '${sqlStr(cfg.userName)}', ` + - `${Buffer.byteLength(line, "utf-8")}, '${sqlStr(projectName)}', '${sqlStr(msg.role)}', 'openclaw', '${sqlStr(getInstalledVersion() ?? "")}', '${ts}', '${ts}')`; + const insertSql = buildDirectSessionInsertSql(sessionsTable, { + id: crypto.randomUUID(), + sessionPath, + filename, + jsonForSql, + embeddingSql, + userName: cfg.userName, + sizeBytes: Buffer.byteLength(line, "utf-8"), + projectName, + description: msg.role, + agent: "openclaw", + pluginVersion: getInstalledVersion() ?? "", + timestamp: ts, + }); try { await dl.query(insertSql); diff --git a/src/hooks/capture.ts b/src/hooks/capture.ts index 9f8768fe5..070e5ef05 100644 --- a/src/hooks/capture.ts +++ b/src/hooks/capture.ts @@ -11,7 +11,6 @@ import { readStdin } from "../utils/stdin.js"; import { type Config } from "../config.js"; import { resolveCaptureConfig } from "./shared/dir-gate.js"; import { DeeplakeApi } from "../deeplake-api.js"; -import { sqlStr } from "../utils/sql.js"; import { projectNameFromCwd } from "../utils/project-name.js"; import { log as _log } from "../utils/debug.js"; import { buildSessionPath } from "../utils/session-path.js"; @@ -28,6 +27,7 @@ import { tryStopCounterTrigger } from "../skillify/triggers.js"; import { reactSkillOpt } from "./shared/skillopt-hook.js"; import { EmbedClient } from "../embeddings/client.js"; import { embeddingSqlLiteral } from "../embeddings/sql.js"; +import { buildDirectSessionInsertSql } from "./shared/session-insert-sql.js"; import { embeddingsDisabled } from "../embeddings/disable.js"; import { isHivemindPluginEnabled } from "../utils/plugin-state.js"; import { ensurePluginNodeModulesLink } from "../embeddings/self-heal.js"; @@ -163,10 +163,20 @@ async function main(): Promise { : await new EmbedClient({ daemonEntry: resolveEmbedDaemonPath() }).embed(line, "document"); const embeddingSql = embeddingSqlLiteral(embedding); - const insertSql = - `INSERT INTO "${sessionsTable}" (id, path, filename, message, message_embedding, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + - `VALUES ('${crypto.randomUUID()}', '${sqlStr(sessionPath)}', '${sqlStr(filename)}', '${jsonForSql}'::jsonb, ${embeddingSql}, '${sqlStr(config.userName)}', ` + - `${Buffer.byteLength(line, "utf-8")}, '${sqlStr(projectName)}', '${sqlStr(input.hook_event_name ?? "")}', 'claude_code', '${sqlStr(PLUGIN_VERSION)}', '${ts}', '${ts}')`; + const insertSql = buildDirectSessionInsertSql(sessionsTable, { + id: crypto.randomUUID(), + sessionPath, + filename, + jsonForSql, + embeddingSql, + userName: config.userName, + sizeBytes: Buffer.byteLength(line, "utf-8"), + projectName, + description: input.hook_event_name ?? "", + agent: "claude_code", + pluginVersion: PLUGIN_VERSION, + timestamp: ts, + }); try { await api.query(insertSql); diff --git a/src/hooks/codex/capture.ts b/src/hooks/codex/capture.ts index 824f03a95..745a91f26 100644 --- a/src/hooks/codex/capture.ts +++ b/src/hooks/codex/capture.ts @@ -16,13 +16,13 @@ import { readStdin } from "../../utils/stdin.js"; import { type Config } from "../../config.js"; import { resolveCaptureConfig } from "../shared/dir-gate.js"; import { DeeplakeApi } from "../../deeplake-api.js"; -import { sqlStr } from "../../utils/sql.js"; import { projectNameFromCwd } from "../../utils/project-name.js"; import { log as _log } from "../../utils/debug.js"; import { buildSessionPath } from "../../utils/session-path.js"; import { EmbedClient } from "../../embeddings/client.js"; import { embeddingSqlLiteral } from "../../embeddings/sql.js"; import { embeddingsDisabled } from "../../embeddings/disable.js"; +import { buildDirectSessionInsertSql } from "../shared/session-insert-sql.js"; import { ensurePluginNodeModulesLink } from "../../embeddings/self-heal.js"; import { fileURLToPath } from "node:url"; import { dirname, join } from "node:path"; @@ -136,10 +136,20 @@ async function main(): Promise { : await new EmbedClient({ daemonEntry: resolveEmbedDaemonPath() }).embed(line, "document"); const embeddingSql = embeddingSqlLiteral(embedding); - const insertSql = - `INSERT INTO "${sessionsTable}" (id, path, filename, message, message_embedding, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + - `VALUES ('${crypto.randomUUID()}', '${sqlStr(sessionPath)}', '${sqlStr(filename)}', '${jsonForSql}'::jsonb, ${embeddingSql}, '${sqlStr(config.userName)}', ` + - `${Buffer.byteLength(line, "utf-8")}, '${sqlStr(projectName)}', '${sqlStr(input.hook_event_name ?? "")}', 'codex', '${sqlStr(PLUGIN_VERSION)}', '${ts}', '${ts}')`; + const insertSql = buildDirectSessionInsertSql(sessionsTable, { + id: crypto.randomUUID(), + sessionPath, + filename, + jsonForSql, + embeddingSql, + userName: config.userName, + sizeBytes: Buffer.byteLength(line, "utf-8"), + projectName, + description: input.hook_event_name ?? "", + agent: "codex", + pluginVersion: PLUGIN_VERSION, + timestamp: ts, + }); try { await api.query(insertSql); diff --git a/src/hooks/codex/stop.ts b/src/hooks/codex/stop.ts index a8526be6e..a84c6fa75 100644 --- a/src/hooks/codex/stop.ts +++ b/src/hooks/codex/stop.ts @@ -18,7 +18,6 @@ import { readStdin } from "../../utils/stdin.js"; import { loadConfig } from "../../config.js"; import { resolveDirConfig } from "../../dir-config.js"; import { DeeplakeApi } from "../../deeplake-api.js"; -import { sqlStr } from "../../utils/sql.js"; import { projectNameFromCwd } from "../../utils/project-name.js"; import { log as _log } from "../../utils/debug.js"; import { bundleDirFromImportMeta, spawnCodexWikiWorker, wikiLog } from "./spawn-wiki-worker.js"; @@ -28,6 +27,7 @@ import { buildSessionPath } from "../../utils/session-path.js"; import { EmbedClient } from "../../embeddings/client.js"; import { embeddingSqlLiteral } from "../../embeddings/sql.js"; import { embeddingsDisabled } from "../../embeddings/disable.js"; +import { buildDirectSessionInsertSql } from "../shared/session-insert-sql.js"; import { getInstalledVersion } from "../../utils/version-check.js"; const log = (msg: string) => _log("codex-stop", msg); @@ -131,10 +131,20 @@ async function main(): Promise { : await new EmbedClient({ daemonEntry: resolveEmbedDaemonPath() }).embed(line, "document"); const embeddingSql = embeddingSqlLiteral(embedding); - const insertSql = - `INSERT INTO "${sessionsTable}" (id, path, filename, message, message_embedding, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + - `VALUES ('${crypto.randomUUID()}', '${sqlStr(sessionPath)}', '${sqlStr(filename)}', '${jsonForSql}'::jsonb, ${embeddingSql}, '${sqlStr(config.userName)}', ` + - `${Buffer.byteLength(line, "utf-8")}, '${sqlStr(projectName)}', 'Stop', 'codex', '${sqlStr(PLUGIN_VERSION)}', '${ts}', '${ts}')`; + const insertSql = buildDirectSessionInsertSql(sessionsTable, { + id: crypto.randomUUID(), + sessionPath, + filename, + jsonForSql, + embeddingSql, + userName: config.userName, + sizeBytes: Buffer.byteLength(line, "utf-8"), + projectName, + description: "Stop", + agent: "codex", + pluginVersion: PLUGIN_VERSION, + timestamp: ts, + }); await api.query(insertSql); log("stop event captured"); diff --git a/src/hooks/cursor/capture.ts b/src/hooks/cursor/capture.ts index cd7c5836c..38770c537 100644 --- a/src/hooks/cursor/capture.ts +++ b/src/hooks/cursor/capture.ts @@ -16,13 +16,13 @@ import { readStdin } from "../../utils/stdin.js"; import { resolveCaptureConfig } from "../shared/dir-gate.js"; import { DeeplakeApi } from "../../deeplake-api.js"; -import { sqlStr } from "../../utils/sql.js"; import { projectNameFromCwd } from "../../utils/project-name.js"; import { log as _log } from "../../utils/debug.js"; import { buildSessionPath } from "../../utils/session-path.js"; import { EmbedClient } from "../../embeddings/client.js"; import { embeddingSqlLiteral } from "../../embeddings/sql.js"; import { embeddingsDisabled } from "../../embeddings/disable.js"; +import { buildDirectSessionInsertSql } from "../shared/session-insert-sql.js"; import { ensurePluginNodeModulesLink } from "../../embeddings/self-heal.js"; import { fileURLToPath } from "node:url"; import { dirname, join } from "node:path"; @@ -161,10 +161,20 @@ async function main(): Promise { : await new EmbedClient({ daemonEntry: resolveEmbedDaemonPath() }).embed(line, "document"); const embeddingSql = embeddingSqlLiteral(embedding); - const insertSql = - `INSERT INTO "${sessionsTable}" (id, path, filename, message, message_embedding, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + - `VALUES ('${crypto.randomUUID()}', '${sqlStr(sessionPath)}', '${sqlStr(filename)}', '${jsonForSql}'::jsonb, ${embeddingSql}, '${sqlStr(config.userName)}', ` + - `${Buffer.byteLength(line, "utf-8")}, '${sqlStr(projectName)}', '${sqlStr(event)}', 'cursor', '${sqlStr(PLUGIN_VERSION)}', '${ts}', '${ts}')`; + const insertSql = buildDirectSessionInsertSql(sessionsTable, { + id: crypto.randomUUID(), + sessionPath, + filename, + jsonForSql, + embeddingSql, + userName: config.userName, + sizeBytes: Buffer.byteLength(line, "utf-8"), + projectName, + description: event, + agent: "cursor", + pluginVersion: PLUGIN_VERSION, + timestamp: ts, + }); try { await api.query(insertSql); diff --git a/src/hooks/hermes/capture.ts b/src/hooks/hermes/capture.ts index 836b22f46..82437099d 100644 --- a/src/hooks/hermes/capture.ts +++ b/src/hooks/hermes/capture.ts @@ -17,13 +17,13 @@ import { readStdin } from "../../utils/stdin.js"; import { resolveCaptureConfig } from "../shared/dir-gate.js"; import { DeeplakeApi } from "../../deeplake-api.js"; -import { sqlStr } from "../../utils/sql.js"; import { projectNameFromCwd } from "../../utils/project-name.js"; import { log as _log } from "../../utils/debug.js"; import { buildSessionPath } from "../../utils/session-path.js"; import { EmbedClient } from "../../embeddings/client.js"; import { embeddingSqlLiteral } from "../../embeddings/sql.js"; import { embeddingsDisabled } from "../../embeddings/disable.js"; +import { buildDirectSessionInsertSql } from "../shared/session-insert-sql.js"; import { ensurePluginNodeModulesLink } from "../../embeddings/self-heal.js"; import { fileURLToPath } from "node:url"; import { dirname, join } from "node:path"; @@ -144,10 +144,20 @@ async function main(): Promise { : await new EmbedClient({ daemonEntry: resolveEmbedDaemonPath() }).embed(line, "document"); const embeddingSql = embeddingSqlLiteral(embedding); - const insertSql = - `INSERT INTO "${sessionsTable}" (id, path, filename, message, message_embedding, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + - `VALUES ('${crypto.randomUUID()}', '${sqlStr(sessionPath)}', '${sqlStr(filename)}', '${jsonForSql}'::jsonb, ${embeddingSql}, '${sqlStr(config.userName)}', ` + - `${Buffer.byteLength(line, "utf-8")}, '${sqlStr(projectName)}', '${sqlStr(event)}', 'hermes', '${sqlStr(PLUGIN_VERSION)}', '${ts}', '${ts}')`; + const insertSql = buildDirectSessionInsertSql(sessionsTable, { + id: crypto.randomUUID(), + sessionPath, + filename, + jsonForSql, + embeddingSql, + userName: config.userName, + sizeBytes: Buffer.byteLength(line, "utf-8"), + projectName, + description: event, + agent: "hermes", + pluginVersion: PLUGIN_VERSION, + timestamp: ts, + }); try { await api.query(insertSql); diff --git a/src/hooks/shared/session-insert-sql.ts b/src/hooks/shared/session-insert-sql.ts new file mode 100644 index 000000000..fea472de8 --- /dev/null +++ b/src/hooks/shared/session-insert-sql.ts @@ -0,0 +1,55 @@ +import { sqlIdent, sqlStr } from "../../utils/sql.js"; + +/** + * Fields for one session-event row written by the capture hooks. All string + * values are escaped by the builder EXCEPT `jsonForSql` and `embeddingSql`, + * which are pre-formatted SQL fragments (see notes on each field). + */ +export interface DirectSessionInsertParams { + /** Stable per-event row id. MUST be constant across retries of the same + * event so the idempotency guard below can recognise a re-send. */ + id: string; + sessionPath: string; + filename: string; + /** JSON payload with single quotes already doubled (`'' `). Embedded raw and + * cast to jsonb — do NOT pass through sqlStr(), which would corrupt the JSON. */ + jsonForSql: string; + /** SQL literal for message_embedding: either `NULL` or `ARRAY[...]::float4[]`. + * Produced by embeddingSqlLiteral(); embedded raw. */ + embeddingSql: string; + userName: string; + sizeBytes: number; + projectName: string; + description: string; + agent: string; + pluginVersion: string; + /** ISO timestamp used for both creation_date and last_update_date. */ + timestamp: string; +} + +/** + * Build the single-row session INSERT used by every capture hook. + * + * Idempotent by construction: the row is inserted via `INSERT ... SELECT ... + * WHERE NOT EXISTS (SELECT 1 FROM WHERE id = )` rather than a plain + * `VALUES` insert. The Deeplake sessions table has no UNIQUE constraint on `id`, + * so a plain INSERT that the API layer retries after a transient 5xx (the + * request committed but the gateway returned 502/503) creates a duplicate row. + * The `WHERE NOT EXISTS` guard makes the re-send a no-op instead — verified + * lag-safe against the real backend even when the retry fires inside the + * documented ~5s read-your-writes window (see probe results / PR notes). + * + * The column list starts with `id, path, filename, message,` so the query is + * still recognised by isSessionInsertQuery() in deeplake-api.ts (which enables + * the transient-403 retry path for session writes). + */ +export function buildDirectSessionInsertSql(sessionsTable: string, p: DirectSessionInsertParams): string { + const table = sqlIdent(sessionsTable); + const id = sqlStr(p.id); + return ( + `INSERT INTO "${table}" (id, path, filename, message, message_embedding, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + + `SELECT '${id}', '${sqlStr(p.sessionPath)}', '${sqlStr(p.filename)}', '${p.jsonForSql}'::jsonb, ${p.embeddingSql}, '${sqlStr(p.userName)}', ` + + `${p.sizeBytes}, '${sqlStr(p.projectName)}', '${sqlStr(p.description)}', '${sqlStr(p.agent)}', '${sqlStr(p.pluginVersion)}', '${sqlStr(p.timestamp)}', '${sqlStr(p.timestamp)}' ` + + `WHERE NOT EXISTS (SELECT 1 FROM "${table}" WHERE id = '${id}')` + ); +} diff --git a/tests/claude-code/plugin-version-resolution.test.ts b/tests/claude-code/plugin-version-resolution.test.ts index a724b4732..c2eb4101d 100644 --- a/tests/claude-code/plugin-version-resolution.test.ts +++ b/tests/claude-code/plugin-version-resolution.test.ts @@ -75,9 +75,12 @@ describe("plugin_version is wired into every agent's capture INSERT", () => { it.each(CAPTURE_BUNDLES)("%s INSERT lists plugin_version column", (_label, path) => { const src = readFileSync(path, "utf-8"); // The INSERT into the sessions table must include plugin_version in - // its column list. Regex matches the actual concatenated INSERT line - // so a typo or column-list drift fails here, not silently in prod. - const sessionsInsert = /INSERT INTO\s+"\$\{sessionsTable\}"[^`]*?plugin_version[^`]*?VALUES/; + // its column list. Every agent now builds this through the shared + // buildDirectSessionInsertSql helper, whose column list is a single + // template literal `INSERT INTO "${table}" (... plugin_version ...)` + // followed by SELECT ... WHERE NOT EXISTS (the idempotency guard). + // A typo or column-list drift fails here, not silently in prod. + const sessionsInsert = /INSERT INTO\s+"\$\{table\}"\s*\(id, path, filename, message,[^`]*?plugin_version/; expect(src).toMatch(sessionsInsert); }); }); diff --git a/tests/claude-code/session-insert-sql.test.ts b/tests/claude-code/session-insert-sql.test.ts new file mode 100644 index 000000000..d58b1e534 --- /dev/null +++ b/tests/claude-code/session-insert-sql.test.ts @@ -0,0 +1,98 @@ +import { describe, it, expect } from "vitest"; +import { readFileSync } from "node:fs"; +import { resolve } from "node:path"; +import { buildDirectSessionInsertSql } from "../../src/hooks/shared/session-insert-sql.js"; + +/** + * C1 regression coverage: session-event INSERTs must be idempotent. + * + * The sessions table has no UNIQUE constraint on `id`, so a plain + * `INSERT ... VALUES` that the API layer retries after a transient 5xx + * (the request committed but the gateway returned 502/503) creates a + * duplicate row — this is what produced the ~17% duplicate rows observed + * in production during the 2026-07-10 gateway-degradation window. + * + * The fix builds every capture INSERT via `INSERT ... SELECT ... WHERE NOT + * EXISTS (id = ...)`, so a re-send of the same event is a no-op. Verified + * lag-safe against the real backend in the e2e probe; these tests lock the + * SQL shape (source) and the shipped artifact (bundle). + */ + +const params = { + id: "row-123", + sessionPath: "/sessions/u/s.jsonl", + filename: "s.jsonl", + jsonForSql: `{"type":"tool_call","note":"it''s fine"}`, + embeddingSql: "NULL", + userName: "u", + sizeBytes: 42, + projectName: "proj", + description: "PostToolUse", + agent: "claude_code", + pluginVersion: "9.9.9", + timestamp: "2026-07-13T00:00:00.000Z", +}; + +describe("buildDirectSessionInsertSql", () => { + const sql = buildDirectSessionInsertSql("sessions", params); + + it("uses the idempotent INSERT ... SELECT ... WHERE NOT EXISTS form", () => { + expect(sql).toMatch(/INSERT INTO "sessions" \(id, path, filename, message, message_embedding,/); + expect(sql).toMatch(/SELECT '/); + expect(sql).toMatch(/WHERE NOT EXISTS \(SELECT 1 FROM "sessions" WHERE id = 'row-123'\)/); + }); + + it("is NOT a plain VALUES insert (the duplicate-prone pattern)", () => { + expect(sql).not.toMatch(/\)\s*VALUES\s*\(/i); + }); + + it("references the same id in the row and in the guard (stable dedup key)", () => { + const ids = sql.match(/'row-123'/g) ?? []; + expect(ids.length).toBe(2); + }); + + it("keeps the column prefix isSessionInsertQuery() relies on for retry routing", () => { + // deeplake-api.ts isSessionInsertQuery: ^insert into "..." (id, path, filename, message, + expect(sql).toMatch(/^INSERT INTO "sessions" \(\s*id, path, filename, message,/); + }); + + it("casts the JSON payload to jsonb and inlines the embedding literal", () => { + expect(sql).toContain(`'${params.jsonForSql}'::jsonb`); + expect(sql).toContain("NULL"); + }); + + it("emits an array literal when an embedding vector is present", () => { + const withVec = buildDirectSessionInsertSql("sessions", { ...params, embeddingSql: "ARRAY[0.1,-0.2]::float4[]" }); + expect(withVec).toContain("ARRAY[0.1,-0.2]::float4[]"); + }); +}); + +/** + * Bundle-level guard — proves the build didn't drop the fix or re-inline the + * old bare-VALUES pattern into what users actually ship. + */ +const ROOT = process.cwd(); +const BUNDLES: Array<[string, string]> = [ + ["claude-code capture", resolve(ROOT, "harnesses", "claude-code", "bundle", "capture.js")], + ["codex capture", resolve(ROOT, "harnesses", "codex", "bundle", "capture.js")], + ["codex stop", resolve(ROOT, "harnesses", "codex", "bundle", "stop.js")], + ["cursor capture", resolve(ROOT, "harnesses", "cursor", "bundle", "capture.js")], + ["hermes capture", resolve(ROOT, "harnesses", "hermes", "bundle", "capture.js")], + ["openclaw index", resolve(ROOT, "harnesses", "openclaw", "dist", "index.js")], +]; + +for (const [label, path] of BUNDLES) { + describe(`${label} bundle`, () => { + const src = readFileSync(path, "utf-8"); + + it("ships the idempotency guard", () => { + expect(src).toMatch(/NOT EXISTS \(SELECT 1 FROM/); + }); + + it("ships no bare VALUES session insert", () => { + const bareSessionValuesInsert = + /message_embedding, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date\)\s*VALUES\s*\(/; + expect(src).not.toMatch(bareSessionValuesInsert); + }); + }); +} diff --git a/tests/openclaw/openclaw-embed-bundle.test.ts b/tests/openclaw/openclaw-embed-bundle.test.ts index a670cba36..b2c9a69fb 100644 --- a/tests/openclaw/openclaw-embed-bundle.test.ts +++ b/tests/openclaw/openclaw-embed-bundle.test.ts @@ -33,11 +33,11 @@ describe("openclaw dist bundle — embeddings wiring", () => { }); it("session INSERT column list includes message_embedding", () => { - // The whole INSERT lives on one line in the minified bundle. Match - // the column list between INSERT INTO "...sessions..." and VALUES. - // Locking in the column NAME, not its position, so a future reorder - // doesn't false-fail. - const insertMatches = SRC.match(/INSERT INTO "\$\{sessionsTable[^"]*\}"\s*\([^)]+\)/g) ?? []; + // The sessions INSERT is now built by the shared buildDirectSessionInsertSql + // helper, whose column list is `INSERT INTO "${table}" (...)`. Match the + // column list and lock in the column NAME, not its position, so a future + // reorder doesn't false-fail. + const insertMatches = SRC.match(/INSERT INTO "\$\{table\}"\s*\([^)]+\)/g) ?? []; expect(insertMatches.length).toBeGreaterThanOrEqual(1); for (const m of insertMatches) { expect(m).toContain("message_embedding"); diff --git a/tests/shared/agent-sessions-insert-schema.test.ts b/tests/shared/agent-sessions-insert-schema.test.ts index 4e1ea33c3..b0b471932 100644 --- a/tests/shared/agent-sessions-insert-schema.test.ts +++ b/tests/shared/agent-sessions-insert-schema.test.ts @@ -16,11 +16,12 @@ import { SESSIONS_COLUMNS } from "../../src/deeplake-schema.js"; * INSERT writes must exist in SESSIONS_COLUMNS — otherwise CREATE/heal never * create it and the write 42703s. Lock it so the pi bug can't reappear here. */ +// The claude/codex/cursor/hermes capture hooks and codex/stop.ts all build +// their single-row INSERT through the shared buildDirectSessionInsertSql +// helper, so its one column list covers every direct-insert agent. The +// batched queue path keeps its own inline column list. const SESSIONS_INSERT_FILES = [ - "src/hooks/capture.ts", - "src/hooks/codex/capture.ts", - "src/hooks/cursor/capture.ts", - "src/hooks/hermes/capture.ts", + "src/hooks/shared/session-insert-sql.ts", "src/hooks/session-queue.ts", ]; From 36af5070ed8f9c429b491fae507f57f08492b30d Mon Sep 17 00:00:00 2001 From: Emanuele Fenocchi Date: Tue, 14 Jul 2026 17:16:50 +0000 Subject: [PATCH 2/4] fix(pi): make pi sessions INSERT idempotent too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex review of the previous commit flagged that the pi extension still wrote session rows with a bare `INSERT ... VALUES` in writeSessionRow. pi has no active 5xx-retry path today (dlQuery does not retry; writeSessionRow only retries on table-not-visible), so it has no live duplicate vector — but the bare insert is inconsistent with the fix and a latent bug if retry behavior changes. pi deliberately imports nothing from src/ ("raw .ts, zero deps"), so inline the same `INSERT ... SELECT ... WHERE NOT EXISTS (id = ...)` shape rather than the shared helper, kept in lockstep. Adds a pi-source regression guard. --- harnesses/pi/extension-source/hivemind.ts | 12 ++++++++++-- tests/pi/pi-extension-source.test.ts | 15 +++++++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/harnesses/pi/extension-source/hivemind.ts b/harnesses/pi/extension-source/hivemind.ts index 82af0f47b..745a9513e 100644 --- a/harnesses/pi/extension-source/hivemind.ts +++ b/harnesses/pi/extension-source/hivemind.ts @@ -961,10 +961,18 @@ async function writeSessionRow( logHm(`writeSessionRow: event=${event} session=${sessionId} bytes=${line.length} table=${SESSIONS_TABLE}`); const emb = await embed(line); logHm(`writeSessionRow: embed=${emb ? `dims=${emb.length}` : "null"}`); + // Idempotent INSERT: `SELECT ... WHERE NOT EXISTS (id = ...)` instead of a + // plain VALUES insert, so a re-send of the same event can never create a + // duplicate row (the sessions table has no UNIQUE constraint on id). pi keeps + // this inline rather than importing src/hooks/shared/session-insert-sql.ts to + // preserve the "raw .ts, zero deps" promise — kept in lockstep with that + // helper's shape. + const rowId = crypto.randomUUID(); const insertSql = `INSERT INTO "${SESSIONS_TABLE}" (id, path, filename, message, message_embedding, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + - `VALUES ('${crypto.randomUUID()}', '${sqlStr(sessionPath)}', '${sqlStr(filename)}', '${jsonForSql}'::jsonb, ${embedSqlLiteral(emb)}, '${sqlStr(creds.userName)}', ` + - `${Buffer.byteLength(line, "utf-8")}, '${sqlStr(projectName)}', '${sqlStr(event)}', '${agent}', '${sqlStr(PLUGIN_VERSION)}', '${ts}', '${ts}')`; + `SELECT '${rowId}', '${sqlStr(sessionPath)}', '${sqlStr(filename)}', '${jsonForSql}'::jsonb, ${embedSqlLiteral(emb)}, '${sqlStr(creds.userName)}', ` + + `${Buffer.byteLength(line, "utf-8")}, '${sqlStr(projectName)}', '${sqlStr(event)}', '${agent}', '${sqlStr(PLUGIN_VERSION)}', '${ts}', '${ts}' ` + + `WHERE NOT EXISTS (SELECT 1 FROM "${SESSIONS_TABLE}" WHERE id = '${rowId}')`; let lastErr: any = null; for (let attempt = 0; attempt <= INSERT_RETRY_BACKOFFS_MS.length; attempt++) { try { diff --git a/tests/pi/pi-extension-source.test.ts b/tests/pi/pi-extension-source.test.ts index 37552f360..e1fe37493 100644 --- a/tests/pi/pi-extension-source.test.ts +++ b/tests/pi/pi-extension-source.test.ts @@ -32,6 +32,21 @@ describe("pi extension — embedding wiring", () => { expect(insertLine![0]).toContain("message_embedding"); }); + // Regression for C1 (duplicate session rows). The sessions table has no + // UNIQUE constraint on id, so a plain `INSERT ... VALUES` that gets re-sent + // (retry after a transient error) creates a duplicate. pi must build the row + // idempotently — `INSERT ... SELECT ... WHERE NOT EXISTS (id = ...)` — matching + // the shared buildDirectSessionInsertSql helper the bundled agents use. + it("builds the sessions INSERT idempotently (no bare VALUES insert)", () => { + expect(PI_SRC).toMatch( + /INSERT INTO "\$\{SESSIONS_TABLE\}"[\s\S]*?WHERE NOT EXISTS \(SELECT 1 FROM "\$\{SESSIONS_TABLE\}" WHERE id = '\$\{rowId\}'\)/, + ); + // ...and never the duplicate-prone bare VALUES form for the sessions row. + expect(PI_SRC).not.toMatch( + /INSERT INTO "\$\{SESSIONS_TABLE\}" \([^)]*message_embedding[^)]*\)\s*VALUES/, + ); + }); + // Regression for the pi `plugin_version` 42703 incident (org c2d29f27 et al.): // the pi extension hard-codes its sessions CREATE TABLE inline (it does NOT go // through DeeplakeApi.ensureSessionsTable / healMissingColumns like the other From c49ccd82a61950919051274166cd7c0a99db6192 Mon Sep 17 00:00:00 2001 From: Emanuele Fenocchi Date: Tue, 21 Jul 2026 00:27:13 +0000 Subject: [PATCH 3/4] fix(session-queue): make the batched cowork INSERT idempotent too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex review found the batched session-queue insert path — used by the Cowork MCP ingest (src/mcp/cowork-ingest.ts) — still emitted a bare multi-row `INSERT ... VALUES`. That path builds a DeeplakeApi whose query() retries on 5xx (and re-sends again on table-missing), so it carried the same duplicate vector as the per-event capture path. Switch buildSessionInsertSql to the anti-join form: INSERT ... SELECT v.* FROM (VALUES ...) AS v(cols) WHERE NOT EXISTS (SELECT 1 FROM
t WHERE t.id = v.id) so a re-sent batch skips rows already present. Verified against the real backend: a 2-row batch re-sent inside and beyond the ~5s read-your-writes window stays at 2 rows, and a partially-overlapping batch inserts only the new id. Adds a source-level guard for the shape. --- src/hooks/session-queue.ts | 13 ++++++++++--- tests/claude-code/session-queue.test.ts | 12 ++++++++++++ 2 files changed, 22 insertions(+), 3 deletions(-) diff --git a/src/hooks/session-queue.ts b/src/hooks/session-queue.ts index 455cc2110..f2e17ffcb 100644 --- a/src/hooks/session-queue.ts +++ b/src/hooks/session-queue.ts @@ -141,10 +141,17 @@ export function buildSessionInsertSql(sessionsTable: string, rows: QueuedSession ); }).join(", "); + // Idempotent batch insert: skip any row whose id already exists, so a flush + // that re-sends the batch after a transient 5xx (the insert committed but the + // gateway returned 502/503) cannot duplicate rows. The sessions table has no + // UNIQUE constraint on id, so this anti-join is the multi-row equivalent of + // buildDirectSessionInsertSql's `WHERE NOT EXISTS` guard on the single-row + // capture path. Verified lag-safe against the real backend. return ( - `INSERT INTO "${table}" ` + - `(id, path, filename, message, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + - `VALUES ${values}` + `INSERT INTO "${table}" (id, path, filename, message, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + + `SELECT v.id, v.path, v.filename, v.message, v.author, v.size_bytes, v.project, v.description, v.agent, v.plugin_version, v.creation_date, v.last_update_date ` + + `FROM (VALUES ${values}) AS v(id, path, filename, message, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + + `WHERE NOT EXISTS (SELECT 1 FROM "${table}" AS t WHERE t.id = v.id)` ); } diff --git a/tests/claude-code/session-queue.test.ts b/tests/claude-code/session-queue.test.ts index 068f41b07..f49574645 100644 --- a/tests/claude-code/session-queue.test.ts +++ b/tests/claude-code/session-queue.test.ts @@ -109,6 +109,18 @@ describe("session queue", () => { expect(sql).toContain("), ("); }); + it("builds an idempotent batch insert (anti-join, not a bare VALUES insert)", () => { + // C1: the sessions table has no UNIQUE constraint on id, so a flush that + // re-sends the batch after a transient 5xx must not duplicate rows. The + // builder emits INSERT ... SELECT ... FROM (VALUES ...) v WHERE NOT EXISTS + // (id = v.id) — the multi-row equivalent of the single-row capture guard. + const sql = buildSessionInsertSql("sessions", [makeRow("s", 1), makeRow("s", 2)]); + expect(sql).toMatch(/FROM \(VALUES .*\) AS v\(id, path, filename, message,/s); + expect(sql).toMatch(/WHERE NOT EXISTS \(SELECT 1 FROM "sessions" AS t WHERE t\.id = v\.id\)/); + // ...and never the duplicate-prone `) VALUES (` form directly after the column list. + expect(sql).not.toMatch(/last_update_date\)\s*VALUES\s*\(/); + }); + it("wraps malformed messages in a valid JSON object before casting to jsonb", () => { const row = makeRow("session-sql-fallback", 1, { message: "{not-json", From c556c2c5021a398adb0c1be12ad00ce6f9f428b1 Mon Sep 17 00:00:00 2001 From: Emanuele Fenocchi Date: Tue, 21 Jul 2026 00:42:08 +0000 Subject: [PATCH 4/4] fix(capture): address CodeRabbit review on the C1 idempotency PR - Align the row id with the event id: every capture path now passes the message JSON's own `entry.id` as the row PK instead of a fresh random UUID, so the `id` column matches the id embedded in the `message` payload and stays the dedup key for the logical event. Applies to claude/codex/cursor/hermes capture, codex stop, openclaw, and pi (with a uuid fallback if a caller ever omits entry.id). - pi: escape the `agent` value via sqlStr() in the inline INSERT, matching the shared helper and the other columns. - tests: replace wildcard regexes with exact column-list assertions in plugin-version-resolution and session-queue tests. --- harnesses/openclaw/src/index.ts | 4 +++- harnesses/pi/extension-source/hivemind.ts | 7 +++++-- src/hooks/capture.ts | 4 +++- src/hooks/codex/capture.ts | 4 +++- src/hooks/codex/stop.ts | 4 +++- src/hooks/cursor/capture.ts | 4 +++- src/hooks/hermes/capture.ts | 4 +++- .../plugin-version-resolution.test.ts | 17 +++++++++-------- tests/claude-code/session-queue.test.ts | 11 ++++++++--- 9 files changed, 40 insertions(+), 19 deletions(-) diff --git a/harnesses/openclaw/src/index.ts b/harnesses/openclaw/src/index.ts index 2cffdae8a..7ac220ff4 100644 --- a/harnesses/openclaw/src/index.ts +++ b/harnesses/openclaw/src/index.ts @@ -1540,7 +1540,9 @@ export default definePluginEntry({ const embeddingSql = embeddingSqlLiteral(embedding); const insertSql = buildDirectSessionInsertSql(sessionsTable, { - id: crypto.randomUUID(), + // Reuse the event id already embedded in the message JSON so the + // row PK matches the payload's id (dedup key = the logical event). + id: entry.id, sessionPath, filename, jsonForSql, diff --git a/harnesses/pi/extension-source/hivemind.ts b/harnesses/pi/extension-source/hivemind.ts index 745a9513e..465205755 100644 --- a/harnesses/pi/extension-source/hivemind.ts +++ b/harnesses/pi/extension-source/hivemind.ts @@ -967,11 +967,14 @@ async function writeSessionRow( // this inline rather than importing src/hooks/shared/session-insert-sql.ts to // preserve the "raw .ts, zero deps" promise — kept in lockstep with that // helper's shape. - const rowId = crypto.randomUUID(); + // Reuse the event id embedded in the message JSON so the row PK matches the + // payload's id and stays the dedup key across a re-send. Fall back to a fresh + // uuid if a caller ever omits it (keeps each row unique rather than colliding). + const rowId = sqlStr(typeof entry.id === "string" ? entry.id : crypto.randomUUID()); const insertSql = `INSERT INTO "${SESSIONS_TABLE}" (id, path, filename, message, message_embedding, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date) ` + `SELECT '${rowId}', '${sqlStr(sessionPath)}', '${sqlStr(filename)}', '${jsonForSql}'::jsonb, ${embedSqlLiteral(emb)}, '${sqlStr(creds.userName)}', ` + - `${Buffer.byteLength(line, "utf-8")}, '${sqlStr(projectName)}', '${sqlStr(event)}', '${agent}', '${sqlStr(PLUGIN_VERSION)}', '${ts}', '${ts}' ` + + `${Buffer.byteLength(line, "utf-8")}, '${sqlStr(projectName)}', '${sqlStr(event)}', '${sqlStr(agent)}', '${sqlStr(PLUGIN_VERSION)}', '${ts}', '${ts}' ` + `WHERE NOT EXISTS (SELECT 1 FROM "${SESSIONS_TABLE}" WHERE id = '${rowId}')`; let lastErr: any = null; for (let attempt = 0; attempt <= INSERT_RETRY_BACKOFFS_MS.length; attempt++) { diff --git a/src/hooks/capture.ts b/src/hooks/capture.ts index b61ecc5ac..d189993e2 100644 --- a/src/hooks/capture.ts +++ b/src/hooks/capture.ts @@ -169,7 +169,9 @@ async function main(): Promise { const embeddingSql = embeddingSqlLiteral(embedding); const insertSql = buildDirectSessionInsertSql(sessionsTable, { - id: crypto.randomUUID(), + // Reuse the event id already embedded in the message JSON so the row PK + // matches the payload's id (and keeps the dedup key = the logical event). + id: entry.id as string, sessionPath, filename, jsonForSql, diff --git a/src/hooks/codex/capture.ts b/src/hooks/codex/capture.ts index 8aaa717b1..e5870abc7 100644 --- a/src/hooks/codex/capture.ts +++ b/src/hooks/codex/capture.ts @@ -138,7 +138,9 @@ async function main(): Promise { const embeddingSql = embeddingSqlLiteral(embedding); const insertSql = buildDirectSessionInsertSql(sessionsTable, { - id: crypto.randomUUID(), + // Reuse the event id already embedded in the message JSON so the row PK + // matches the payload's id (and keeps the dedup key = the logical event). + id: entry.id as string, sessionPath, filename, jsonForSql, diff --git a/src/hooks/codex/stop.ts b/src/hooks/codex/stop.ts index a84c6fa75..e88694e25 100644 --- a/src/hooks/codex/stop.ts +++ b/src/hooks/codex/stop.ts @@ -132,7 +132,9 @@ async function main(): Promise { const embeddingSql = embeddingSqlLiteral(embedding); const insertSql = buildDirectSessionInsertSql(sessionsTable, { - id: crypto.randomUUID(), + // Reuse the event id already embedded in the message JSON so the row PK + // matches the payload's id (and keeps the dedup key = the logical event). + id: entry.id, sessionPath, filename, jsonForSql, diff --git a/src/hooks/cursor/capture.ts b/src/hooks/cursor/capture.ts index 2febcf428..e3214aa39 100644 --- a/src/hooks/cursor/capture.ts +++ b/src/hooks/cursor/capture.ts @@ -164,7 +164,9 @@ async function main(): Promise { const embeddingSql = embeddingSqlLiteral(embedding); const insertSql = buildDirectSessionInsertSql(sessionsTable, { - id: crypto.randomUUID(), + // Reuse the event id already embedded in the message JSON so the row PK + // matches the payload's id (and keeps the dedup key = the logical event). + id: entry.id as string, sessionPath, filename, jsonForSql, diff --git a/src/hooks/hermes/capture.ts b/src/hooks/hermes/capture.ts index 61755e646..0448dff25 100644 --- a/src/hooks/hermes/capture.ts +++ b/src/hooks/hermes/capture.ts @@ -147,7 +147,9 @@ async function main(): Promise { const embeddingSql = embeddingSqlLiteral(embedding); const insertSql = buildDirectSessionInsertSql(sessionsTable, { - id: crypto.randomUUID(), + // Reuse the event id already embedded in the message JSON so the row PK + // matches the payload's id (and keeps the dedup key = the logical event). + id: entry.id as string, sessionPath, filename, jsonForSql, diff --git a/tests/claude-code/plugin-version-resolution.test.ts b/tests/claude-code/plugin-version-resolution.test.ts index c2eb4101d..2f87fcdcc 100644 --- a/tests/claude-code/plugin-version-resolution.test.ts +++ b/tests/claude-code/plugin-version-resolution.test.ts @@ -74,14 +74,15 @@ describe("plugin_version is wired into every agent's capture INSERT", () => { it.each(CAPTURE_BUNDLES)("%s INSERT lists plugin_version column", (_label, path) => { const src = readFileSync(path, "utf-8"); - // The INSERT into the sessions table must include plugin_version in - // its column list. Every agent now builds this through the shared - // buildDirectSessionInsertSql helper, whose column list is a single - // template literal `INSERT INTO "${table}" (... plugin_version ...)` - // followed by SELECT ... WHERE NOT EXISTS (the idempotency guard). - // A typo or column-list drift fails here, not silently in prod. - const sessionsInsert = /INSERT INTO\s+"\$\{table\}"\s*\(id, path, filename, message,[^`]*?plugin_version/; - expect(src).toMatch(sessionsInsert); + // The INSERT into the sessions table must carry the exact canonical column + // list (built by the shared buildDirectSessionInsertSql helper), with + // plugin_version present. Asserting the full literal — not a wildcard — + // so a typo or column-list drift fails here, not silently in prod. + const expectedColumns = + 'INSERT INTO "${table}" (id, path, filename, message, message_embedding, ' + + "author, size_bytes, project, description, agent, plugin_version, " + + "creation_date, last_update_date)"; + expect(src).toContain(expectedColumns); }); }); diff --git a/tests/claude-code/session-queue.test.ts b/tests/claude-code/session-queue.test.ts index f49574645..928347124 100644 --- a/tests/claude-code/session-queue.test.ts +++ b/tests/claude-code/session-queue.test.ts @@ -115,10 +115,15 @@ describe("session queue", () => { // builder emits INSERT ... SELECT ... FROM (VALUES ...) v WHERE NOT EXISTS // (id = v.id) — the multi-row equivalent of the single-row capture guard. const sql = buildSessionInsertSql("sessions", [makeRow("s", 1), makeRow("s", 2)]); - expect(sql).toMatch(/FROM \(VALUES .*\) AS v\(id, path, filename, message,/s); - expect(sql).toMatch(/WHERE NOT EXISTS \(SELECT 1 FROM "sessions" AS t WHERE t\.id = v\.id\)/); + const cols = + "id, path, filename, message, author, size_bytes, project, description, agent, plugin_version, creation_date, last_update_date"; + expect(sql).toContain(`INSERT INTO "sessions" (${cols})`); + expect(sql).toContain(`) AS v(${cols})`); + expect(sql).toContain( + `WHERE NOT EXISTS (SELECT 1 FROM "sessions" AS t WHERE t.id = v.id)`, + ); // ...and never the duplicate-prone `) VALUES (` form directly after the column list. - expect(sql).not.toMatch(/last_update_date\)\s*VALUES\s*\(/); + expect(sql).not.toContain(`${cols}) VALUES (`); }); it("wraps malformed messages in a valid JSON object before casting to jsonb", () => {