fix(capture): make session-event INSERTs idempotent to stop duplicate rows (C1)

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);

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 {

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<void> {
     : 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);

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<void> {
     : 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);

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<void> {
         : 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");

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<void> {
     : 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);

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<void> {
     : 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);

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)`
   );
 }
 

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 <table> WHERE id = <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}')`
+  );
+}