diff --git a/harnesses/openclaw/src/index.ts b/harnesses/openclaw/src/index.ts index 91de87eb..7ac220ff 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"; import { redactSecrets } from "../../../src/hooks/shared/redact.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 @@ -1538,10 +1539,22 @@ 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, { + // 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, + 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/harnesses/pi/extension-source/hivemind.ts b/harnesses/pi/extension-source/hivemind.ts index 82af0f47..46520575 100644 --- a/harnesses/pi/extension-source/hivemind.ts +++ b/harnesses/pi/extension-source/hivemind.ts @@ -961,10 +961,21 @@ 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. + // 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) ` + - `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)}', '${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++) { try { diff --git a/src/hooks/capture.ts b/src/hooks/capture.ts index 8388d83d..d189993e 100644 --- a/src/hooks/capture.ts +++ b/src/hooks/capture.ts @@ -12,7 +12,6 @@ import { type Config } from "../config.js"; import { resolveCaptureConfig } from "./shared/dir-gate.js"; import { redactSecrets } from "./shared/redact.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"; @@ -30,6 +29,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"; @@ -168,10 +168,22 @@ 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, { + // 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, + 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 f6b33512..e5870abc 100644 --- a/src/hooks/codex/capture.ts +++ b/src/hooks/codex/capture.ts @@ -17,13 +17,13 @@ import { type Config } from "../../config.js"; import { resolveCaptureConfig } from "../shared/dir-gate.js"; import { redactSecrets } from "../shared/redact.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"; @@ -137,10 +137,22 @@ 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, { + // 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, + 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 a8526be6..e88694e2 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,22 @@ 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, { + // 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, + 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 dbdd1529..e3214aa3 100644 --- a/src/hooks/cursor/capture.ts +++ b/src/hooks/cursor/capture.ts @@ -17,13 +17,13 @@ import { readStdin } from "../../utils/stdin.js"; import { resolveCaptureConfig } from "../shared/dir-gate.js"; import { redactSecrets } from "../shared/redact.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"; @@ -163,10 +163,22 @@ 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, { + // 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, + 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 f79a5ef0..0448dff2 100644 --- a/src/hooks/hermes/capture.ts +++ b/src/hooks/hermes/capture.ts @@ -18,13 +18,13 @@ import { readStdin } from "../../utils/stdin.js"; import { resolveCaptureConfig } from "../shared/dir-gate.js"; import { redactSecrets } from "../shared/redact.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"; @@ -146,10 +146,22 @@ 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, { + // 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, + 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/session-queue.ts b/src/hooks/session-queue.ts index 455cc211..f2e17ffc 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/src/hooks/shared/session-insert-sql.ts b/src/hooks/shared/session-insert-sql.ts new file mode 100644 index 00000000..fea472de --- /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 a724b473..2f87fcdc 100644 --- a/tests/claude-code/plugin-version-resolution.test.ts +++ b/tests/claude-code/plugin-version-resolution.test.ts @@ -74,11 +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. Regex matches the actual concatenated INSERT line + // 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 sessionsInsert = /INSERT INTO\s+"\$\{sessionsTable\}"[^`]*?plugin_version[^`]*?VALUES/; - expect(src).toMatch(sessionsInsert); + 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-insert-sql.test.ts b/tests/claude-code/session-insert-sql.test.ts new file mode 100644 index 00000000..d58b1e53 --- /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/claude-code/session-queue.test.ts b/tests/claude-code/session-queue.test.ts index 068f41b0..92834712 100644 --- a/tests/claude-code/session-queue.test.ts +++ b/tests/claude-code/session-queue.test.ts @@ -109,6 +109,23 @@ 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)]); + 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.toContain(`${cols}) VALUES (`); + }); + it("wraps malformed messages in a valid JSON object before casting to jsonb", () => { const row = makeRow("session-sql-fallback", 1, { message: "{not-json", diff --git a/tests/openclaw/openclaw-embed-bundle.test.ts b/tests/openclaw/openclaw-embed-bundle.test.ts index a670cba3..b2c9a69f 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/pi/pi-extension-source.test.ts b/tests/pi/pi-extension-source.test.ts index 37552f36..e1fe3749 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 diff --git a/tests/shared/agent-sessions-insert-schema.test.ts b/tests/shared/agent-sessions-insert-schema.test.ts index 4e1ea33c..b0b47193 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", ];