FEA-1444: Codex hook ingestion + dedup + opt-in toggle

apps/desktop/package.json

@@ -1,6 +1,6 @@
 {
   "name": "desktop",
-  "version": "0.15.99",
+  "version": "0.15.100",
   "description": "ClosedLoop Desktop",
   "author": "ClosedLoop AI <support@closedloop.ai>",
   "private": true,

apps/desktop/scripts/agent-monitor-codex/codex-hook-handler.js

@@ -0,0 +1,80 @@
+#!/usr/bin/env node
+
+/**
+ * @file codex-hook-handler.js
+ * @description Codex CLI hook handler. Mirrors the upstream Claude
+ * `hook-handler.js` (provider-agnostic, POSTs to `/api/hooks/event` on the
+ * fixed agent-monitor sidecar port 4820) but injects `__provider: "codex"`
+ * into the forwarded payload so the sidecar can stamp the session row with
+ * `harness='codex'` via the existing `setSessionHarness` statement.
+ *
+ * Zero-dep, plain JS, fail-silent — same constraints as the upstream Claude
+ * handler so a hook never blocks a Codex turn. Codex calls this once per
+ * lifecycle event (SessionStart, UserPromptSubmit, PreToolUse, PostToolUse,
+ * Stop) with the event name as the single argv arg.
+ *
+ * Part of FEA-1444 (opt-in Codex hook ingestion).
+ */
+
+const http = require("http");
+
+const hookType = process.argv[2] || "unknown";
+const port = parseInt(process.env.CLAUDE_DASHBOARD_PORT || "4820", 10);
+
+let input = "";
+
+process.stdin.setEncoding("utf8");
+process.stdin.on("data", (chunk) => (input += chunk));
+process.stdin.on("end", () => {
+  let parsedData;
+  try {
+    parsedData = JSON.parse(input);
+  } catch {
+    parsedData = { raw: input };
+  }
+
+  // Mark the payload as Codex-sourced so the sidecar's hooks-route patch
+  // (build-agent-monitor.mjs `patchHooksRouteCodexHarness`) can stamp the
+  // session's `harness` column. Field name is dunder-prefixed to make it
+  // obvious this is a transport hint, not a Codex-native field.
+  const enrichedData =
+    parsedData && typeof parsedData === "object" && !Array.isArray(parsedData)
+      ? { ...parsedData, __provider: "codex" }
+      : { raw: parsedData, __provider: "codex" };
+
+  const payload = JSON.stringify({
+    hook_type: hookType,
+    data: enrichedData,
+  });
+
+  const req = http.request(
+    {
+      hostname: "127.0.0.1",
+      port,
+      path: "/api/hooks/event",
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        "Content-Length": Buffer.byteLength(payload),
+      },
+      timeout: 3000,
+    },
+    (res) => {
+      res.resume();
+      process.exit(0);
+    },
+  );
+
+  req.on("error", () => process.exit(0));
+  req.on("timeout", () => {
+    req.destroy();
+    process.exit(0);
+  });
+
+  req.write(payload);
+  req.end();
+});
+
+// Safety net timeout — Codex's default hook timeout is around 5s; never let
+// this process linger longer than that.
+setTimeout(() => process.exit(0), 5000);

apps/desktop/scripts/agent-monitor-codex/codex-import.js

@@ -30,6 +30,16 @@ const catchupCache = createCatchupCache({ persistPath: ingestCachePath("codex")
  * or { skipped: true } when the file has no usable content.
  */
 function importCodexSession(dbModule, session) {
+  // FEA-1444 dedup: if the user opted into Codex hooks, the same logical
+  // event will already be in `events` (inserted ~5s earlier by the hook
+  // handler). Without this filter the rollout-tail importer would create a
+  // duplicate row for every Codex event after the user opts in. Match on
+  // (session_id, event_type, tool_name, created_at-truncated-to-second) —
+  // hooks and rollout-tail timestamps usually agree within sub-second
+  // granularity for the same logical event. False negatives are tolerable
+  // (cosmetic duplicates); false positives would silently drop events, so
+  // the match is intentionally narrow.
+  filterEventsAlreadyCapturedByHooks(dbModule, session);
   const result = importSession(dbModule, session);
   // Stamp the harness regardless of skipped/backfilled — cheap, idempotent,
   // and self-heals rows imported before the `harness` column existed.
@@ -42,6 +52,73 @@ function importCodexSession(dbModule, session) {
   return { sessionId: session.sessionId, result, reactivated };
 }
 
+/**
+ * FEA-1444: filter session.events in place, removing any whose
+ * (session_id, event_type, COALESCE(tool_name, ''), created_at-rounded-to-second)
+ * already exists in the `events` table. The hook handler inserts in real time;
+ * rollout-tail catches up ~5s later — so the dedup query consistently sees
+ * hook-sourced rows first when both paths are active.
+ *
+ * No-op when session.events is empty or the events query fails (best-effort;
+ * never block the import on a dedup-time error).
+ */
+// FEA-1444 dedup-stmt cache: the filter runs per-session inside
+// importCodexSession, which itself runs inside the importBatch transaction
+// loop. Caching the prepared statement on the dbModule keeps us at O(1)
+// compilations per process instead of O(sessions) per batch. The cache is
+// keyed by dbModule so a fresh DatabaseSync in tests gets a fresh stmt.
+const dedupStmtCache = new WeakMap();
+function getDedupCheckStmt(dbModule) {
+  let stmt = dedupStmtCache.get(dbModule);
+  if (stmt) return stmt;
+  // Truncate created_at to seconds via `substr(?, 1, 19)` so sub-second
+  // timestamp drift between the hook payload and the rollout file doesn't
+  // produce a false negative.
+  stmt = dbModule.db.prepare(
+    "SELECT 1 FROM events " +
+      "WHERE session_id = ? AND event_type = ? " +
+      "AND COALESCE(tool_name, '') = COALESCE(?, '') " +
+      "AND substr(COALESCE(created_at, ''), 1, 19) = substr(COALESCE(?, ''), 1, 19) " +
+      "LIMIT 1",
+  );
+  dedupStmtCache.set(dbModule, stmt);
+  return stmt;
+}
+
+function filterEventsAlreadyCapturedByHooks(dbModule, session) {
+  if (!session || !Array.isArray(session.events) || session.events.length === 0) {
+    return;
+  }
+  let checkStmt;
+  try {
+    checkStmt = getDedupCheckStmt(dbModule);
+  } catch {
+    return; // schema mismatch or db locked — fall through, accept duplicates
+  }
+  const filtered = [];
+  for (const ev of session.events) {
+    if (!ev || typeof ev !== "object") {
+      filtered.push(ev);
+      continue;
+    }
+    try {
+      const hit = checkStmt.get(
+        session.sessionId,
+        ev.event_type ?? null,
+        ev.tool_name ?? null,
+        ev.created_at ?? null,
+      );
+      if (hit) {
+        continue; // already captured by the hook handler — drop the duplicate
+      }
+    } catch {
+      // best-effort — on any per-row failure, keep the event
+    }
+    filtered.push(ev);
+  }
+  session.events = filtered;
+}
+
 /**
  * Parse + import every discovered Codex rollout file. Designed to be cheap on
  * repeat runs: importSession skips already-imported sessions (or backfills
@@ -107,4 +184,9 @@ async function importAllCodexSessions(dbModule, opts = {}) {
   return { imported, skipped, errors };
 }
 
-module.exports = { importAllCodexSessions, importCodexSession };
+module.exports = {
+  importAllCodexSessions,
+  importCodexSession,
+  // Exposed for regression coverage of FEA-1444 dedup.
+  filterEventsAlreadyCapturedByHooks,
+};

apps/desktop/scripts/build-agent-monitor.mjs

@@ -133,6 +133,14 @@ const IS_SESSION_IN_SANDBOX_CJS = [
 // — relative requires resolve identically in the generated tree as they did
 // in the old vendored tree.
 const codexModulesDir = path.join(appDir, "scripts", "agent-monitor-codex");
+// FEA-1444: Codex hook handler wrapper ships in-repo (mirrors the upstream
+// Claude hook-handler.js placement under generated `scripts/`). Copied at
+// materialize time and surfaced by agent-monitor-hooks.ts via the same
+// `scriptsDir` resolver as the Claude handler.
+const codexHookHandlerSource = path.join(
+  codexModulesDir,
+  "codex-hook-handler.js",
+);
 const cursorModulesDir = path.join(appDir, "scripts", "agent-monitor-cursor");
 const copilotModulesDir = path.join(appDir, "scripts", "agent-monitor-copilot");
 const opencodeModulesDir = path.join(appDir, "scripts", "agent-monitor-opencode");
@@ -459,6 +467,9 @@ function currentStamp() {
     ...MULTI_HARNESS_SPECS.flatMap(({ modulesDir, modules }) =>
       modules.map((m) => path.join(modulesDir, `${m}.js`)),
     ),
+    // FEA-1444: invalidate the cached generated tree when the Codex hook
+    // wrapper source changes.
+    codexHookHandlerSource,
     ...SHARED_MODULES.map((m) => path.join(sharedModulesDir, `${m}.js`)),
     ...CLIENT_SNIPPET_FILES.map((file) => path.join(clientSnippetDir, file)),
     ...PLAN_MODULES.map((m) => path.join(planModulesDir, `${m}.js`)),
@@ -640,11 +651,24 @@ function materializeRuntimeTree() {
   patchHooksTranscriptOutsideTx(generatedHooksRoute);
   patchHooksWriteQueueAndWatchdog(generatedHooksRoute);
   patchHooksSandboxFilter(generatedHooksRoute);
+  // FEA-1444: stamp harness='codex' inside processEventCore when the inbound
+  // hook payload was forwarded by the Codex wrapper handler. Order: must run
+  // AFTER patchHooksTranscriptOutsideTx because it relies on the
+  // `function processEventCore(hookType, data, ...)` signature that patch
+  // installs.
+  patchHooksRouteCodexHarness(generatedHooksRoute);
   patchImportRoute(generatedImportRoute);
   patchPushFile(generatedPushLib);
   patchWebSocketFile(generatedWebSocketFile);
   patchCcDiscovery(generatedCcDiscovery);
   writeFileSync(generatedUninstallHooks, UNINSTALL_HOOKS_SOURCE, "utf8");
+  // FEA-1444: copy the Codex hook wrapper into the generated scripts/ tree
+  // so the agent-monitor-hooks installer can resolve it via the same
+  // `scriptsDir` it uses for the upstream Claude hook-handler.js.
+  cpSync(
+    codexHookHandlerSource,
+    path.join(generatedRootDir, "scripts", "codex-hook-handler.js"),
+  );
 }
 
 function patchServerIndex(file) {
@@ -1734,6 +1758,54 @@ function patchHooksRoute(file) {
   writeFileSync(file, source, "utf8");
 }
 
+// CLOSEDLOOP FEA-1444: stamp `harness='codex'` on the session row when the
+// inbound hook payload was forwarded by the in-repo `codex-hook-handler.js`
+// wrapper (which injects `__provider: "codex"`). The upstream hooks route is
+// provider-agnostic and already supports a `harness` column + setSessionHarness
+// statement (Codex Patch #4); this patch is the single place that wires the
+// hook stream to that stamp so the dashboard renders Codex-sourced hook events
+// alongside the existing rollout-tail-imported rows.
+function patchHooksRouteCodexHarness(file) {
+  let source = readFileSync(file, "utf8");
+  if (source.includes("FEA-1444 codex harness stamp")) return;
+
+  // Anchors on the processEventCore signature installed by
+  // patchHooksTranscriptOutsideTx (FEA-1363) + the immediate ensureSession
+  // call. Both must already be present; this patch runs after that one in
+  // the materialize sequence.
+  const needle = [
+    "function processEventCore(hookType, data, transcriptData) {",
+    "  const sessionId = data.session_id;",
+    "  if (!sessionId) return null;",
+    "",
+    "  const session = ensureSession(sessionId, data);",
+  ].join("\n");
+  if (!source.includes(needle)) {
+    throw new Error(
+      `Unable to patch ${file}: expected processEventCore + ensureSession anchor (FEA-1444).`,
+    );
+  }
+  source = source.replace(
+    needle,
+    [
+      needle,
+      "  // FEA-1444 codex harness stamp: the in-repo codex-hook-handler.js",
+      "  // wrapper injects `__provider: \"codex\"` into the forwarded hook",
+      "  // payload. When present, mark the session row as a Codex session so",
+      "  // the dashboard groups it with the rollout-tail-imported rows and",
+      "  // any harness-scoped UI affordances apply.",
+      "  if (session && data && data.__provider === \"codex\") {",
+      "    try {",
+      "      stmts.setSessionHarness.run(\"codex\", sessionId, \"codex\");",
+      "    } catch (_) {",
+      "      /* non-fatal: harness column/stmt guaranteed by Codex Patch #4 */",
+      "    }",
+      "  }",
+    ].join("\n"),
+  );
+  writeFileSync(file, source, "utf8");
+}
+
 // CLOSEDLOOP FEA-1363: Fix SQLite write contention under 22+ concurrent agents.
 // Three patches below address five compounding SQLite problems that cause the
 // agent monitor dashboard to corrupt when many agents fire hooks simultaneously.
@@ -3097,6 +3169,10 @@ function assertGeneratedTree() {
     generatedClientIndex,
     path.join(generatedRootDir, "scripts", "install-hooks.js"),
     path.join(generatedRootDir, "scripts", "hook-handler.js"),
+    // FEA-1444: Codex hook wrapper must materialize alongside the upstream
+    // Claude handler so agent-monitor-hooks.ts can resolve both from the same
+    // `scriptsDir`.
+    path.join(generatedRootDir, "scripts", "codex-hook-handler.js"),
     generatedUninstallHooks,
   ]) {
     if (!existsSync(required)) {
@@ -3322,6 +3398,15 @@ function assertGeneratedTree() {
       "Generated server/routes/hooks.js is missing the sandbox scoping filter (FEA-1407).",
     );
   }
+
+  // CLOSEDLOOP FEA-1444 hard-gates: Codex hook ingestion. A future upstream
+  // bump that drops the processEventCore anchor must fail the build rather
+  // than silently disabling Codex harness stamping on hook-sourced sessions.
+  if (!hooksRouteSource.includes("FEA-1444 codex harness stamp")) {
+    throw new Error(
+      "Generated server/routes/hooks.js is missing the Codex harness stamp (FEA-1444).",
+    );
+  }
   if (!importHistorySource.includes("FEA-1407 sandbox scoping")) {
     throw new Error(
       "Generated scripts/import-history.js is missing the sandbox scoping filter (FEA-1407).",