diff --git a/packages/cli/src/commands/claude-hook-ingest-spool.test.ts b/packages/cli/src/commands/claude-hook-ingest-spool.test.ts index 29e9b80e7..964a5485c 100644 --- a/packages/cli/src/commands/claude-hook-ingest-spool.test.ts +++ b/packages/cli/src/commands/claude-hook-ingest-spool.test.ts @@ -247,7 +247,8 @@ describe("claude-hook-ingest-spool", () => { writeFileSync(join(queueDir, "hook-array.json"), "[1,2,3]", "utf8"); writeFileSync(join(queueDir, "hook-string.json"), '"oops"', "utf8"); - await drainSpool(async () => true); + const result = await drainSpool(async () => true); + expect(result).toEqual({ processed: 0, failed: 2 }); const remaining = readdirSync(queueDir); // Both files quarantined under .bad-wrong-shape-, none left active. diff --git a/packages/cli/src/commands/claude-hook-ingest-spool.ts b/packages/cli/src/commands/claude-hook-ingest-spool.ts index 80cba341c..fa0718c64 100644 --- a/packages/cli/src/commands/claude-hook-ingest-spool.ts +++ b/packages/cli/src/commands/claude-hook-ingest-spool.ts @@ -1,465 +1,37 @@ /** - * Durability layer for `claude-hook-ingest`: file-based mutex to - * serialize concurrent invocations, on-disk spool that captures - * payloads when both HTTP and direct ingestion paths fail, and a - * recovery routine that promotes stale temp files back into the queue. - */ - -import { randomInt } from "node:crypto"; -import { - mkdirSync, - readdirSync, - readFileSync, - renameSync, - rmdirSync, - statSync, - unlinkSync, - writeFileSync, -} from "node:fs"; -import { homedir } from "node:os"; -import { dirname, join } from "node:path"; -import { logHookEvent } from "./claude-hook-plugin-log.js"; - -const DEFAULT_LOCK_TTL_S = 300; -const DEFAULT_LOCK_GRACE_S = 2; -const LOCK_ACQUIRE_ATTEMPTS = 100; -const LOCK_ACQUIRE_BACKOFF_MS = 50; - -type LockSnapshot = { - pid: string; - ts: number | null; - owner: string; -}; - -type LockConfig = { - lockDir: string; - ttlSeconds: number; - graceSeconds: number; -}; - -export class LockBusyError extends Error { - constructor() { - super("claude-hook-ingest lock busy"); - this.name = "LockBusyError"; - } -} - -function expandHome(value: string): string { - if (value === "~") return homedir(); - if (value.startsWith("~/")) return join(homedir(), value.slice(2)); - return value; -} - -function envInt(name: string, fallback: number): number { - const raw = process.env[name]; - if (raw === undefined) return fallback; - const parsed = Number.parseInt(raw, 10); - return Number.isFinite(parsed) ? parsed : fallback; -} - -function envTruthy(name: string, fallback: boolean): boolean { - const raw = process.env[name]; - if (raw === undefined) return fallback; - const normalized = raw.trim().toLowerCase(); - if (["1", "true", "yes", "on"].includes(normalized)) return true; - if (["0", "false", "no", "off"].includes(normalized)) return false; - return fallback; -} - -function lockConfig(): LockConfig { - const lockDir = expandHome( - process.env.CODEMEM_CLAUDE_HOOK_LOCK_DIR?.trim() || "~/.codemem/claude-hook-ingest.lock", - ); - return { - lockDir, - ttlSeconds: Math.max(1, envInt("CODEMEM_CLAUDE_HOOK_LOCK_TTL_S", DEFAULT_LOCK_TTL_S)), - graceSeconds: Math.max(1, envInt("CODEMEM_CLAUDE_HOOK_LOCK_GRACE_S", DEFAULT_LOCK_GRACE_S)), - }; -} - -export function spoolDir(): string { - return expandHome( - process.env.CODEMEM_CLAUDE_HOOK_SPOOL_DIR?.trim() || "~/.codemem/claude-hook-spool", - ); -} - -/** - * Cheap pre-check used by the unlocked HTTP-success path to decide - * whether it needs to acquire the ingest lock and drain queued - * payloads. Returns true when the spool directory contains at least - * one active entry (a `*.json` file that is neither an in-flight - * `.hook-tmp-*` nor a quarantined `.bad-*` file). Any I/O failure - * is treated as "no entries" so callers stay on the fast path. - */ -export function hasSpooledEntries(): boolean { - const dir = spoolDir(); - let entries: string[]; - try { - entries = readdirSync(dir); - } catch { - return false; - } - for (const name of entries) { - if (!name.endsWith(".json")) continue; - if (name.startsWith(".hook-tmp-") || name.startsWith(".bad-")) continue; - return true; - } - return false; -} - -function readFileTrimmedOrEmpty(path: string): string { - try { - return readFileSync(path, "utf8").trim(); - } catch { - return ""; - } -} - -function readLockMetadata(lockDir: string): LockSnapshot { - const pid = readFileTrimmedOrEmpty(join(lockDir, "pid")); - const owner = readFileTrimmedOrEmpty(join(lockDir, "owner")); - const tsRaw = readFileTrimmedOrEmpty(join(lockDir, "ts")); - const ts = tsRaw === "" ? null : Number.parseInt(tsRaw, 10); - return { - pid, - ts: ts === null || !Number.isFinite(ts) ? null : ts, - owner, - }; -} - -function isPidAlive(pidText: string): boolean { - const pid = Number.parseInt(pidText, 10); - if (!Number.isFinite(pid) || pid <= 0) return false; - try { - // Signal 0 performs the existence check without delivering a signal. - process.kill(pid, 0); - return true; - } catch { - return false; - } -} - -function lockIsStale(cfg: LockConfig): { stale: boolean; snapshot: LockSnapshot } { - const snapshot = readLockMetadata(cfg.lockDir); - const nowS = Math.floor(Date.now() / 1000); - - if (snapshot.pid) { - if (isPidAlive(snapshot.pid)) { - if (snapshot.ts === null) return { stale: false, snapshot }; - return { stale: nowS - snapshot.ts > cfg.ttlSeconds, snapshot }; - } - return { stale: true, snapshot }; - } - - if (snapshot.ts !== null) { - return { stale: nowS - snapshot.ts > cfg.graceSeconds, snapshot }; - } - - let mtimeS: number; - try { - mtimeS = Math.floor(statSync(cfg.lockDir).mtimeMs / 1000); - } catch { - return { stale: true, snapshot }; - } - return { stale: nowS - mtimeS > cfg.graceSeconds, snapshot }; -} - -function cleanupLockDir(lockDir: string): void { - for (const name of ["pid", "ts", "owner"]) { - try { - unlinkSync(join(lockDir, name)); - } catch { - // best-effort - } - } - try { - rmdirSync(lockDir); - } catch { - // best-effort - } -} - -function snapshotsEqual(a: LockSnapshot, b: LockSnapshot): boolean { - return a.pid === b.pid && a.ts === b.ts && a.owner === b.owner; -} - -function cleanupLockDirIfUnchanged(lockDir: string, snapshot: LockSnapshot): void { - const current = readLockMetadata(lockDir); - if (snapshotsEqual(current, snapshot)) { - cleanupLockDir(lockDir); - } -} - -function isErrnoException(err: unknown): err is NodeJS.ErrnoException { - return typeof err === "object" && err !== null && "code" in err; -} - -async function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - -/** - * Run `fn` while holding the claude-hook-ingest lock. Throws - * `LockBusyError` when the lock cannot be acquired within - * `LOCK_ACQUIRE_ATTEMPTS` attempts. - * - * The lock is a directory at `lockDir`, with three sentinel files - * (`pid`, `ts`, `owner`) recording who currently holds it. Stale locks - * are detected via PID liveness, TTL, and a grace window for the - * race between mkdir and writing pid/ts. - */ -export async function withClaudeHookIngestLock(fn: () => Promise | T): Promise { - const cfg = lockConfig(); - mkdirSync(dirname(cfg.lockDir), { recursive: true }); - const ownerToken = `${process.pid}-${Math.floor(Date.now() / 1000)}-${randomInt(1000, 10000)}`; - - let acquired = false; - for (let attempt = 0; attempt < LOCK_ACQUIRE_ATTEMPTS; attempt++) { - try { - mkdirSync(cfg.lockDir); - } catch (err) { - if (isErrnoException(err) && err.code === "EEXIST") { - const { stale, snapshot } = lockIsStale(cfg); - if (stale) { - cleanupLockDirIfUnchanged(cfg.lockDir, snapshot); - } - await sleep(LOCK_ACQUIRE_BACKOFF_MS); - continue; - } - await sleep(LOCK_ACQUIRE_BACKOFF_MS); - continue; - } - - try { - writeFileSync(join(cfg.lockDir, "ts"), String(Math.floor(Date.now() / 1000)), { - encoding: "utf8", - }); - writeFileSync(join(cfg.lockDir, "pid"), String(process.pid), { encoding: "utf8" }); - writeFileSync(join(cfg.lockDir, "owner"), ownerToken, { encoding: "utf8" }); - acquired = true; - break; - } catch { - cleanupLockDir(cfg.lockDir); - await sleep(LOCK_ACQUIRE_BACKOFF_MS); - } - } - - if (!acquired) { - throw new LockBusyError(); - } - - try { - return await fn(); - } finally { - const currentOwner = readFileTrimmedOrEmpty(join(cfg.lockDir, "owner")); - if (currentOwner === ownerToken) { - cleanupLockDir(cfg.lockDir); - } - } -} - -/** - * Persist a payload to the spool directory using a tmp+rename so that - * a partially-written file is never visible to the drainer. Returns - * true on success, false on any I/O failure. - */ -export function spoolPayload(payload: Record): boolean { - const dir = spoolDir(); - try { - mkdirSync(dir, { recursive: true }); - } catch { - logHookEvent("codemem claude-hook-ingest failed to create spool dir"); - return false; - } - - const payloadText = JSON.stringify(payload); - const tmpName = `.hook-tmp-${process.pid}-${Date.now()}-${randomInt(1000, 10000)}.json`; - const tmpPath = join(dir, tmpName); - try { - writeFileSync(tmpPath, payloadText, { encoding: "utf8" }); - } catch { - logHookEvent("codemem claude-hook-ingest failed to allocate spool temp file"); - return false; - } - - const finalName = `hook-${Math.floor(Date.now() / 1000)}-${process.pid}-${randomInt(1000, 10000)}.json`; - const finalPath = join(dir, finalName); - try { - renameSync(tmpPath, finalPath); - } catch { - try { - unlinkSync(tmpPath); - } catch { - // best-effort - } - logHookEvent("codemem claude-hook-ingest failed to spool payload"); - return false; - } - logHookEvent(`codemem claude-hook-ingest spooled payload: ${finalPath}`); - return true; -} - -/** - * Promote any `.hook-tmp-*.json` files older than `ttlSeconds` to a - * recovered name so they are picked up by the next drain. Caller is - * responsible for passing the same TTL used by lock acquisition so - * that an in-flight write inside an active locked region is never - * mistaken for a crashed-writer leftover. - */ -export function recoverStaleTmpSpool(ttlSeconds: number): void { - const dir = spoolDir(); - try { - mkdirSync(dir, { recursive: true }); - } catch { - return; - } - - let entries: string[]; - try { - entries = readdirSync(dir); - } catch { - return; - } - - const nowS = Date.now() / 1000; - for (const name of entries) { - if (!name.startsWith(".hook-tmp-") || !name.endsWith(".json")) continue; - const tmpPath = join(dir, name); - let mtimeS: number; - try { - mtimeS = statSync(tmpPath).mtimeMs / 1000; - } catch { - continue; - } - if (nowS - mtimeS <= ttlSeconds) continue; - - const recoveredName = `hook-recovered-${Math.floor(nowS)}-${process.pid}-${randomInt(1000, 10000)}.json`; - const recoveredPath = join(dir, recoveredName); - try { - renameSync(tmpPath, recoveredPath); - logHookEvent( - `codemem claude-hook-ingest recovered stale temp spool payload: ${recoveredPath}`, - ); - } catch { - // best-effort - } - } -} - -/** - * Move a permanently-broken spool entry out of the queue so that it - * stops being picked up by future drains. The entry is renamed in - * place with a `.bad--` prefix so an operator can inspect or - * delete it manually. - */ -function quarantineSpoolEntry(dir: string, name: string, reason: string): void { - const sourcePath = join(dir, name); - const quarantineName = `.bad-${reason}-${Date.now()}-${randomInt(1000, 10000)}-${name}`; - try { - renameSync(sourcePath, join(dir, quarantineName)); - logHookEvent( - `codemem claude-hook-ingest quarantined corrupt spool payload (${reason}): ${quarantineName}`, - ); - } catch { - // If rename fails, fall back to delete; either way the broken - // entry must not stay in the active queue. - try { - unlinkSync(sourcePath); - logHookEvent(`codemem claude-hook-ingest dropped corrupt spool payload (${reason}): ${name}`); - } catch { - // best-effort - } - } -} - -export type SpoolHandler = (payload: Record) => Promise | boolean; - -export type SpoolDrainResult = { - processed: number; - failed: number; -}; - -/** - * Process every queued payload in the spool directory in lexicographic - * order (which approximates oldest-first because filenames embed the - * second-precision creation timestamp). The handler returns true to - * indicate the payload has been durably accepted; only then is the - * spool entry deleted. Failed entries are left on disk for the next - * drain attempt. - */ -export async function drainSpool(handler: SpoolHandler): Promise { - const dir = spoolDir(); - try { - mkdirSync(dir, { recursive: true }); - } catch { - return { processed: 0, failed: 0 }; - } - - let entries: string[]; - try { - entries = readdirSync(dir); - } catch { - return { processed: 0, failed: 0 }; - } - - const queued = entries - .filter( - (name) => - name.endsWith(".json") && !name.startsWith(".hook-tmp-") && !name.startsWith(".bad-"), - ) - .sort(); - - const result: SpoolDrainResult = { processed: 0, failed: 0 }; - for (const name of queued) { - const path = join(dir, name); - let raw: string; - try { - raw = readFileSync(path, "utf8"); - } catch { - // Genuine I/O failure — leave the file alone so the next drain - // can retry, and surface the failure to the plugin log. - logHookEvent(`codemem claude-hook-ingest failed to read spooled payload: ${path}`); - result.failed++; - continue; - } - let parsed: unknown; - try { - parsed = JSON.parse(raw); - } catch { - // Permanently corrupt content — keeping the file around would - // loop forever every drain. Quarantine it under a `.bad-` prefix - // so an operator can inspect it without it being picked up again. - quarantineSpoolEntry(dir, name, "parse-error"); - result.failed++; - continue; - } - if (parsed == null || typeof parsed !== "object" || Array.isArray(parsed)) { - // Parseable but wrong shape — same problem, same fix. - quarantineSpoolEntry(dir, name, "wrong-shape"); - continue; - } - - let ok = false; - try { - ok = await handler(parsed as Record); - } catch { - ok = false; - } - - if (ok) { - try { - unlinkSync(path); - result.processed++; - } catch { - // best-effort - } - } else { - logHookEvent(`codemem claude-hook-ingest failed processing spooled payload: ${path}`); - result.failed++; - } - } - return result; -} + * Durability layer for `claude-hook-ingest`. The lock/spool/drain/ + * quarantine machinery is shared in `hook-ingest-spool.ts`; this file + * wires the claude-specific config (dirs, TTL 300s, 100 acquire + * attempts, error name/message) and keeps the claude flush predicate. + */ + +import { createHookIngestSpool, envTruthy } from "./hook-ingest-spool.js"; + +export type { SpoolDrainResult, SpoolHandler } from "./hook-ingest-spool.js"; + +const spool = createHookIngestSpool({ + logPrefix: "codemem claude-hook-ingest", + lockDirEnv: "CODEMEM_CLAUDE_HOOK_LOCK_DIR", + lockDirDefault: "~/.codemem/claude-hook-ingest.lock", + lockTtlEnv: "CODEMEM_CLAUDE_HOOK_LOCK_TTL_S", + lockTtlDefault: 300, + lockGraceEnv: "CODEMEM_CLAUDE_HOOK_LOCK_GRACE_S", + lockGraceDefault: 2, + lockAcquireAttempts: 100, + spoolDirEnv: "CODEMEM_CLAUDE_HOOK_SPOOL_DIR", + spoolDirDefault: "~/.codemem/claude-hook-spool", + lockBusyErrorName: "LockBusyError", + lockBusyErrorMessage: "claude-hook-ingest lock busy", +}); + +export const LockBusyError = spool.LockBusyError; +export const withClaudeHookIngestLock = spool.withLock; +export const spoolPayload = spool.spoolPayload; +export const drainSpool = spool.drainSpool; +export const hasSpooledEntries = spool.hasSpooledEntries; +export const recoverStaleTmpSpool = spool.recoverStaleTmpSpool; +export const lockTtlSeconds = spool.lockTtlSeconds; +export const spoolDir = spool.spoolDir; /** * Whether the boundary-flush write-through should run for this hook @@ -478,12 +50,3 @@ export function shouldForceBoundaryFlush(payload: Record): bool if (!envTruthy("CODEMEM_CLAUDE_HOOK_FLUSH", false)) return false; return envTruthy("CODEMEM_CLAUDE_HOOK_FLUSH_ON_STOP", false); } - -/** - * Returns the configured lock TTL so callers (`claude-hook-ingest`) - * can pass the same value to `recoverStaleTmpSpool` without re-reading - * the env. - */ -export function lockTtlSeconds(): number { - return lockConfig().ttlSeconds; -} diff --git a/packages/cli/src/commands/claude-hook-ingest.test.ts b/packages/cli/src/commands/claude-hook-ingest.test.ts index d11242ff1..1e44de7bc 100644 --- a/packages/cli/src/commands/claude-hook-ingest.test.ts +++ b/packages/cli/src/commands/claude-hook-ingest.test.ts @@ -439,6 +439,46 @@ describe("claude-hook-ingest command", () => { expect(boundaryFlushCalls[0]?.hook_event_name).toBe("SessionEnd"); }); + it("drains the backlog BEFORE the boundary flush on the HTTP-success path", async () => { + // A previously-spooled payload must be drained before the + // SessionEnd flush pass runs, so the flush sees the queued + // payloads of the session too. + mkdirSync(queueDir, { recursive: true }); + writeFileSync( + join(queueDir, "hook-0000000001-pid-1.json"), + JSON.stringify({ + hook_event_name: "Stop", + session_id: "queued-before-flush", + tag: "queued", + }), + "utf8", + ); + + const events: string[] = []; + const result = await ingestClaudeHookPayload( + { hook_event_name: "SessionEnd", session_id: "sess-end", tag: "fresh" }, + { host: "127.0.0.1", port: 38888 }, + { + httpIngest: async (payload) => { + events.push(`http:${String(payload.tag ?? "")}`); + return { ok: true, inserted: 0, skipped: 0 }; + }, + directIngest: (payload) => { + events.push(`direct:${String(payload.tag ?? "")}`); + return { inserted: 1, skipped: 0 }; + }, + boundaryFlush: (payload) => { + events.push(`flush:${String(payload.tag ?? "")}`); + }, + resolveDb: () => "/tmp/test.sqlite", + }, + ); + expect(result.via).toBe("http"); + // fresh HTTP → queued drain → boundary write-through → flush. + expect(events).toEqual(["http:fresh", "http:queued", "direct:fresh", "flush:fresh"]); + expect(readdirSync(queueDir)).toHaveLength(0); + }); + it("Stop flush truth table: only fires when BOTH flush envs are truthy", async () => { const directCalls: Array> = []; const boundaryFlushCalls: Array> = []; diff --git a/packages/cli/src/commands/claude-hook-ingest.ts b/packages/cli/src/commands/claude-hook-ingest.ts index 24ded30f3..51c6ca824 100644 --- a/packages/cli/src/commands/claude-hook-ingest.ts +++ b/packages/cli/src/commands/claude-hook-ingest.ts @@ -328,8 +328,11 @@ export async function ingestClaudeHookPayload( // 1. Unlocked HTTP attempt — fast path when the viewer is up. const httpResult = await httpIngest(httpPayload(payload), opts.host, port); if (httpResult.ok) { - await flushOnBoundaryIfRequested(); + // Drain any spooled backlog before the boundary flush so the + // flush pass sees every queued payload of the session, not just + // this event. await drainBacklogIfPresent(); + await flushOnBoundaryIfRequested(); return { inserted: httpResult.inserted, skipped: httpResult.skipped, via: "http" }; } diff --git a/packages/cli/src/commands/codex-hook-ingest-spool.test.ts b/packages/cli/src/commands/codex-hook-ingest-spool.test.ts new file mode 100644 index 000000000..10d5080e5 --- /dev/null +++ b/packages/cli/src/commands/codex-hook-ingest-spool.test.ts @@ -0,0 +1,67 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { + lockTtlSeconds as claudeLockTtlSeconds, + spoolDir as claudeSpoolDir, + LockBusyError, +} from "./claude-hook-ingest-spool.js"; +import { + CodexHookLockBusyError, + codexHookLockTtlSeconds, + codexHookSpoolDir, +} from "./codex-hook-ingest-spool.js"; + +describe("codex-hook-ingest-spool factory config", () => { + const savedEnv: Record = {}; + const keys = [ + "CODEMEM_CLAUDE_HOOK_LOCK_TTL_S", + "CODEMEM_CLAUDE_HOOK_SPOOL_DIR", + "CODEMEM_CODEX_HOOK_LOCK_TTL_S", + "CODEMEM_CODEX_HOOK_SPOOL_DIR", + ]; + + function setEnv(key: string, value: string | undefined): void { + if (value === undefined) delete process.env[key]; + else process.env[key] = value; + } + + beforeEach(() => { + for (const key of keys) savedEnv[key] = process.env[key]; + }); + + afterEach(() => { + for (const [key, value] of Object.entries(savedEnv)) setEnv(key, value); + }); + + it("defaults Codex TTL to 120s and honors CODEMEM_CODEX_HOOK_LOCK_TTL_S", () => { + setEnv("CODEMEM_CLAUDE_HOOK_LOCK_TTL_S", undefined); + setEnv("CODEMEM_CODEX_HOOK_LOCK_TTL_S", undefined); + expect(codexHookLockTtlSeconds()).toBe(120); + expect(claudeLockTtlSeconds()).toBe(300); + + setEnv("CODEMEM_CODEX_HOOK_LOCK_TTL_S", "45"); + expect(codexHookLockTtlSeconds()).toBe(45); + expect(claudeLockTtlSeconds()).toBe(300); + }); + + it("keeps Claude and Codex on separate spool directories", () => { + setEnv("CODEMEM_CLAUDE_HOOK_SPOOL_DIR", undefined); + setEnv("CODEMEM_CODEX_HOOK_SPOOL_DIR", undefined); + expect(claudeSpoolDir()).toMatch(/claude-hook-spool$/); + expect(codexHookSpoolDir()).toMatch(/codex-hook-spool$/); + expect(claudeSpoolDir()).not.toBe(codexHookSpoolDir()); + + setEnv("CODEMEM_CLAUDE_HOOK_SPOOL_DIR", "/tmp/claude-spool"); + setEnv("CODEMEM_CODEX_HOOK_SPOOL_DIR", "/tmp/codex-spool"); + expect(claudeSpoolDir()).toBe("/tmp/claude-spool"); + expect(codexHookSpoolDir()).toBe("/tmp/codex-spool"); + }); + + it("uses a distinct lock-busy error identity from Claude", () => { + const codexErr = new CodexHookLockBusyError(); + const claudeErr = new LockBusyError(); + expect(codexErr.name).toBe("CodexHookLockBusyError"); + expect(codexErr.message).toBe("codex-hook-ingest lock busy"); + expect(claudeErr.name).toBe("LockBusyError"); + expect(codexErr.name).not.toBe(claudeErr.name); + }); +}); diff --git a/packages/cli/src/commands/codex-hook-ingest-spool.ts b/packages/cli/src/commands/codex-hook-ingest-spool.ts index 372e4eb19..630de935b 100644 --- a/packages/cli/src/commands/codex-hook-ingest-spool.ts +++ b/packages/cli/src/commands/codex-hook-ingest-spool.ts @@ -1,342 +1,34 @@ -import { randomInt } from "node:crypto"; -import { - mkdirSync, - readdirSync, - readFileSync, - renameSync, - rmdirSync, - statSync, - unlinkSync, - writeFileSync, -} from "node:fs"; -import { homedir } from "node:os"; -import { dirname, join } from "node:path"; -import { logHookEvent } from "./claude-hook-plugin-log.js"; - -const DEFAULT_LOCK_TTL_S = 120; -const DEFAULT_LOCK_GRACE_S = 2; -const LOCK_ACQUIRE_ATTEMPTS = 20; -const LOCK_ACQUIRE_BACKOFF_MS = 50; - -type LockSnapshot = { pid: string; ts: number | null; owner: string }; -type LockConfig = { lockDir: string; ttlSeconds: number; graceSeconds: number }; - -export class CodexHookLockBusyError extends Error { - constructor() { - super("codex-hook-ingest lock busy"); - this.name = "CodexHookLockBusyError"; - } -} - -function expandHome(value: string): string { - if (value === "~") return homedir(); - if (value.startsWith("~/")) return join(homedir(), value.slice(2)); - return value; -} - -function envInt(name: string, fallback: number): number { - const parsed = Number.parseInt(process.env[name] ?? "", 10); - return Number.isFinite(parsed) ? parsed : fallback; -} - -function lockConfig(): LockConfig { - return { - lockDir: expandHome( - process.env.CODEMEM_CODEX_HOOK_LOCK_DIR?.trim() || "~/.codemem/codex-hook-ingest.lock", - ), - ttlSeconds: Math.max(1, envInt("CODEMEM_CODEX_HOOK_LOCK_TTL_S", DEFAULT_LOCK_TTL_S)), - graceSeconds: Math.max(1, envInt("CODEMEM_CODEX_HOOK_LOCK_GRACE_S", DEFAULT_LOCK_GRACE_S)), - }; -} - -export function codexHookSpoolDir(): string { - return expandHome( - process.env.CODEMEM_CODEX_HOOK_SPOOL_DIR?.trim() || "~/.codemem/codex-hook-spool", - ); -} - -export function codexHookLockTtlSeconds(): number { - return lockConfig().ttlSeconds; -} - -export function hasCodexHookSpooledEntries(): boolean { - let entries: string[]; - try { - entries = readdirSync(codexHookSpoolDir()); - } catch { - return false; - } - return entries.some( - (name) => name.endsWith(".json") && !name.startsWith(".hook-tmp-") && !name.startsWith(".bad-"), - ); -} - -function readTrimmed(path: string): string { - try { - return readFileSync(path, "utf8").trim(); - } catch { - return ""; - } -} - -function readLockMetadata(lockDir: string): LockSnapshot { - const rawTs = readTrimmed(join(lockDir, "ts")); - const ts = rawTs === "" ? null : Number.parseInt(rawTs, 10); - return { - pid: readTrimmed(join(lockDir, "pid")), - ts: ts === null || !Number.isFinite(ts) ? null : ts, - owner: readTrimmed(join(lockDir, "owner")), - }; -} - -function isPidAlive(pidText: string): boolean { - const pid = Number.parseInt(pidText, 10); - if (!Number.isFinite(pid) || pid <= 0) return false; - try { - process.kill(pid, 0); - return true; - } catch { - return false; - } -} - -function lockIsStale(cfg: LockConfig): { stale: boolean; snapshot: LockSnapshot } { - const snapshot = readLockMetadata(cfg.lockDir); - const nowS = Math.floor(Date.now() / 1000); - if (snapshot.pid) { - if (isPidAlive(snapshot.pid)) { - return { stale: snapshot.ts !== null && nowS - snapshot.ts > cfg.ttlSeconds, snapshot }; - } - return { stale: true, snapshot }; - } - if (snapshot.ts !== null) return { stale: nowS - snapshot.ts > cfg.graceSeconds, snapshot }; - try { - return { - stale: nowS - Math.floor(statSync(cfg.lockDir).mtimeMs / 1000) > cfg.graceSeconds, - snapshot, - }; - } catch { - return { stale: true, snapshot }; - } -} - -function cleanupLockDir(lockDir: string): void { - for (const name of ["pid", "ts", "owner"]) { - try { - unlinkSync(join(lockDir, name)); - } catch { - // best-effort - } - } - try { - rmdirSync(lockDir); - } catch { - // best-effort - } -} - -function cleanupLockDirIfUnchanged(lockDir: string, snapshot: LockSnapshot): void { - const current = readLockMetadata(lockDir); - if ( - current.pid === snapshot.pid && - current.ts === snapshot.ts && - current.owner === snapshot.owner - ) { - cleanupLockDir(lockDir); - } -} - -function isErrnoException(err: unknown): err is NodeJS.ErrnoException { - return typeof err === "object" && err !== null && "code" in err; -} - -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - -export async function withCodexHookIngestLock(fn: () => Promise | T): Promise { - const cfg = lockConfig(); - mkdirSync(dirname(cfg.lockDir), { recursive: true }); - const ownerToken = `${process.pid}-${Math.floor(Date.now() / 1000)}-${randomInt(1000, 10000)}`; - - let acquired = false; - for (let attempt = 0; attempt < LOCK_ACQUIRE_ATTEMPTS; attempt++) { - try { - mkdirSync(cfg.lockDir); - } catch (err) { - if (isErrnoException(err) && err.code === "EEXIST") { - const { stale, snapshot } = lockIsStale(cfg); - if (stale) cleanupLockDirIfUnchanged(cfg.lockDir, snapshot); - } - await sleep(LOCK_ACQUIRE_BACKOFF_MS); - continue; - } - - try { - writeFileSync(join(cfg.lockDir, "ts"), String(Math.floor(Date.now() / 1000)), "utf8"); - writeFileSync(join(cfg.lockDir, "pid"), String(process.pid), "utf8"); - writeFileSync(join(cfg.lockDir, "owner"), ownerToken, "utf8"); - acquired = true; - break; - } catch { - cleanupLockDir(cfg.lockDir); - await sleep(LOCK_ACQUIRE_BACKOFF_MS); - } - } - - if (!acquired) throw new CodexHookLockBusyError(); - try { - return await fn(); - } finally { - if (readTrimmed(join(cfg.lockDir, "owner")) === ownerToken) cleanupLockDir(cfg.lockDir); - } -} - -export function spoolCodexHookPayload(payload: Record): boolean { - const dir = codexHookSpoolDir(); - try { - mkdirSync(dir, { recursive: true }); - } catch { - logHookEvent("codemem codex-hook-ingest failed to create spool dir"); - return false; - } - - const tmpPath = join( - dir, - `.hook-tmp-${process.pid}-${Date.now()}-${randomInt(1000, 10000)}.json`, - ); - try { - writeFileSync(tmpPath, JSON.stringify(payload), "utf8"); - } catch { - logHookEvent("codemem codex-hook-ingest failed to allocate spool temp file"); - return false; - } - - const finalPath = join( - dir, - `hook-${Math.floor(Date.now() / 1000)}-${process.pid}-${randomInt(1000, 10000)}.json`, - ); - try { - renameSync(tmpPath, finalPath); - } catch { - try { - unlinkSync(tmpPath); - } catch { - // best-effort - } - logHookEvent("codemem codex-hook-ingest failed to spool payload"); - return false; - } - logHookEvent(`codemem codex-hook-ingest spooled payload: ${finalPath}`); - return true; -} - -export function recoverStaleCodexHookTmpSpool(ttlSeconds: number): void { - const dir = codexHookSpoolDir(); - try { - mkdirSync(dir, { recursive: true }); - } catch { - return; - } - - let entries: string[]; - try { - entries = readdirSync(dir); - } catch { - return; - } - - const nowS = Date.now() / 1000; - for (const name of entries) { - if (!name.startsWith(".hook-tmp-") || !name.endsWith(".json")) continue; - const tmpPath = join(dir, name); - try { - if (nowS - statSync(tmpPath).mtimeMs / 1000 <= ttlSeconds) continue; - renameSync( - tmpPath, - join( - dir, - `hook-recovered-${Math.floor(nowS)}-${process.pid}-${randomInt(1000, 10000)}.json`, - ), - ); - } catch { - // best-effort - } - } -} - -function quarantineSpoolEntry(dir: string, name: string, reason: string): void { - try { - renameSync( - join(dir, name), - join(dir, `.bad-${reason}-${Date.now()}-${randomInt(1000, 10000)}-${name}`), - ); - } catch { - try { - unlinkSync(join(dir, name)); - } catch { - // best-effort - } - } -} - -export type CodexSpoolHandler = (payload: Record) => Promise | boolean; - -export async function drainCodexHookSpool( - handler: CodexSpoolHandler, -): Promise<{ processed: number; failed: number }> { - const dir = codexHookSpoolDir(); - try { - mkdirSync(dir, { recursive: true }); - } catch { - return { processed: 0, failed: 0 }; - } - - let entries: string[]; - try { - entries = readdirSync(dir); - } catch { - return { processed: 0, failed: 0 }; - } - - const result = { processed: 0, failed: 0 }; - for (const name of entries - .filter( - (entry) => - entry.endsWith(".json") && !entry.startsWith(".hook-tmp-") && !entry.startsWith(".bad-"), - ) - .sort()) { - const path = join(dir, name); - let parsed: unknown; - try { - parsed = JSON.parse(readFileSync(path, "utf8")); - } catch { - quarantineSpoolEntry(dir, name, "parse-error"); - result.failed++; - continue; - } - if (parsed == null || typeof parsed !== "object" || Array.isArray(parsed)) { - quarantineSpoolEntry(dir, name, "wrong-shape"); - result.failed++; - continue; - } - - let ok = false; - try { - ok = await handler(parsed as Record); - } catch { - ok = false; - } - if (!ok) { - result.failed++; - continue; - } - try { - unlinkSync(path); - result.processed++; - } catch { - // best-effort - } - } - return result; -} +/** + * Durability layer for `codex-hook-ingest`. The lock/spool/drain/ + * quarantine machinery is shared in `hook-ingest-spool.ts`; this file + * wires the codex-specific config (dirs, TTL 120s, 20 acquire attempts, + * error name/message). + */ + +import { createHookIngestSpool } from "./hook-ingest-spool.js"; + +export type { SpoolDrainResult } from "./hook-ingest-spool.js"; + +const spool = createHookIngestSpool({ + logPrefix: "codemem codex-hook-ingest", + lockDirEnv: "CODEMEM_CODEX_HOOK_LOCK_DIR", + lockDirDefault: "~/.codemem/codex-hook-ingest.lock", + lockTtlEnv: "CODEMEM_CODEX_HOOK_LOCK_TTL_S", + lockTtlDefault: 120, + lockGraceEnv: "CODEMEM_CODEX_HOOK_LOCK_GRACE_S", + lockGraceDefault: 2, + lockAcquireAttempts: 20, + spoolDirEnv: "CODEMEM_CODEX_HOOK_SPOOL_DIR", + spoolDirDefault: "~/.codemem/codex-hook-spool", + lockBusyErrorName: "CodexHookLockBusyError", + lockBusyErrorMessage: "codex-hook-ingest lock busy", +}); + +export const CodexHookLockBusyError = spool.LockBusyError; +export const withCodexHookIngestLock = spool.withLock; +export const spoolCodexHookPayload = spool.spoolPayload; +export const drainCodexHookSpool = spool.drainSpool; +export const hasCodexHookSpooledEntries = spool.hasSpooledEntries; +export const recoverStaleCodexHookTmpSpool = spool.recoverStaleTmpSpool; +export const codexHookLockTtlSeconds = spool.lockTtlSeconds; +export const codexHookSpoolDir = spool.spoolDir; diff --git a/packages/cli/src/commands/hook-ingest-spool.ts b/packages/cli/src/commands/hook-ingest-spool.ts new file mode 100644 index 000000000..68a619bad --- /dev/null +++ b/packages/cli/src/commands/hook-ingest-spool.ts @@ -0,0 +1,507 @@ +/** + * Shared durability layer for the hook ingest commands (claude, codex): + * a file-based mutex that serializes concurrent invocations, an on-disk + * spool that captures payloads when both HTTP and direct ingestion fail, + * and a recovery routine that promotes stale temp files back into the + * queue. Each client instantiates its own copy via `createHookIngestSpool` + * with its own lock/spool dirs, TTL, retry budget, log prefix, and + * client-specific `LockBusyError` name/message. Per-client flush + * predicates stay in the client files. + */ + +import { randomInt } from "node:crypto"; +import { + mkdirSync, + readdirSync, + readFileSync, + renameSync, + rmdirSync, + statSync, + unlinkSync, + writeFileSync, +} from "node:fs"; +import { homedir } from "node:os"; +import { dirname, join } from "node:path"; +import { logHookEvent } from "./claude-hook-plugin-log.js"; + +const LOCK_ACQUIRE_BACKOFF_MS = 50; + +type LockSnapshot = { + pid: string; + ts: number | null; + owner: string; +}; + +type LockConfig = { + lockDir: string; + ttlSeconds: number; + graceSeconds: number; +}; + +export type SpoolHandler = (payload: Record) => Promise | boolean; + +export type SpoolDrainResult = { + processed: number; + failed: number; +}; + +export type HookIngestSpoolConfig = { + /** Log prefix, e.g. "codemem claude-hook-ingest". */ + logPrefix: string; + lockDirEnv: string; + lockDirDefault: string; + lockTtlEnv: string; + lockTtlDefault: number; + lockGraceEnv: string; + lockGraceDefault: number; + lockAcquireAttempts: number; + spoolDirEnv: string; + spoolDirDefault: string; + lockBusyErrorName: string; + lockBusyErrorMessage: string; +}; + +export type HookIngestSpool = { + LockBusyError: new () => Error; + spoolDir: () => string; + lockTtlSeconds: () => number; + hasSpooledEntries: () => boolean; + withLock: (fn: () => Promise | T) => Promise; + spoolPayload: (payload: Record) => boolean; + recoverStaleTmpSpool: (ttlSeconds: number) => void; + drainSpool: (handler: SpoolHandler) => Promise; +}; + +/** + * Boolean-shaped env toggle used by per-client flush predicates + * (claude's `shouldForceBoundaryFlush`). + */ +export function envTruthy(name: string, fallback: boolean): boolean { + const raw = process.env[name]; + if (raw === undefined) return fallback; + const normalized = raw.trim().toLowerCase(); + if (["1", "true", "yes", "on"].includes(normalized)) return true; + if (["0", "false", "no", "off"].includes(normalized)) return false; + return fallback; +} + +function expandHome(value: string): string { + if (value === "~") return homedir(); + if (value.startsWith("~/")) return join(homedir(), value.slice(2)); + return value; +} + +function envInt(name: string, fallback: number): number { + const raw = process.env[name]; + if (raw === undefined) return fallback; + const parsed = Number.parseInt(raw, 10); + return Number.isFinite(parsed) ? parsed : fallback; +} + +export function createHookIngestSpool(cfg: HookIngestSpoolConfig): HookIngestSpool { + class LockBusyError extends Error { + constructor() { + super(cfg.lockBusyErrorMessage); + this.name = cfg.lockBusyErrorName; + } + } + + function lockConfig(): LockConfig { + return { + lockDir: expandHome(process.env[cfg.lockDirEnv]?.trim() || cfg.lockDirDefault), + ttlSeconds: Math.max(1, envInt(cfg.lockTtlEnv, cfg.lockTtlDefault)), + graceSeconds: Math.max(1, envInt(cfg.lockGraceEnv, cfg.lockGraceDefault)), + }; + } + + function spoolDir(): string { + return expandHome(process.env[cfg.spoolDirEnv]?.trim() || cfg.spoolDirDefault); + } + + /** + * Returns the configured lock TTL so callers (hook-ingest commands) + * can pass the same value to `recoverStaleTmpSpool` without re-reading + * the env. + */ + function lockTtlSeconds(): number { + return lockConfig().ttlSeconds; + } + + /** + * Cheap pre-check used by the unlocked HTTP-success path to decide + * whether it needs to acquire the ingest lock and drain queued + * payloads. Returns true when the spool directory contains at least + * one active entry (a `*.json` file that is neither an in-flight + * `.hook-tmp-*` nor a quarantined `.bad-*` file). Any I/O failure + * is treated as "no entries" so callers stay on the fast path. + */ + function hasSpooledEntries(): boolean { + let entries: string[]; + try { + entries = readdirSync(spoolDir()); + } catch { + return false; + } + for (const name of entries) { + if (!name.endsWith(".json")) continue; + if (name.startsWith(".hook-tmp-") || name.startsWith(".bad-")) continue; + return true; + } + return false; + } + + function readFileTrimmedOrEmpty(path: string): string { + try { + return readFileSync(path, "utf8").trim(); + } catch { + return ""; + } + } + + function readLockMetadata(lockDir: string): LockSnapshot { + const pid = readFileTrimmedOrEmpty(join(lockDir, "pid")); + const owner = readFileTrimmedOrEmpty(join(lockDir, "owner")); + const tsRaw = readFileTrimmedOrEmpty(join(lockDir, "ts")); + const ts = tsRaw === "" ? null : Number.parseInt(tsRaw, 10); + return { + pid, + ts: ts === null || !Number.isFinite(ts) ? null : ts, + owner, + }; + } + + function isPidAlive(pidText: string): boolean { + const pid = Number.parseInt(pidText, 10); + if (!Number.isFinite(pid) || pid <= 0) return false; + try { + // Signal 0 performs the existence check without delivering a signal. + process.kill(pid, 0); + return true; + } catch { + return false; + } + } + + function lockIsStale(cfg: LockConfig): { stale: boolean; snapshot: LockSnapshot } { + const snapshot = readLockMetadata(cfg.lockDir); + const nowS = Math.floor(Date.now() / 1000); + + if (snapshot.pid) { + if (isPidAlive(snapshot.pid)) { + if (snapshot.ts === null) return { stale: false, snapshot }; + return { stale: nowS - snapshot.ts > cfg.ttlSeconds, snapshot }; + } + return { stale: true, snapshot }; + } + + if (snapshot.ts !== null) { + return { stale: nowS - snapshot.ts > cfg.graceSeconds, snapshot }; + } + + let mtimeS: number; + try { + mtimeS = Math.floor(statSync(cfg.lockDir).mtimeMs / 1000); + } catch { + return { stale: true, snapshot }; + } + return { stale: nowS - mtimeS > cfg.graceSeconds, snapshot }; + } + + function cleanupLockDir(lockDir: string): void { + for (const name of ["pid", "ts", "owner"]) { + try { + unlinkSync(join(lockDir, name)); + } catch { + // best-effort + } + } + try { + rmdirSync(lockDir); + } catch { + // best-effort + } + } + + function snapshotsEqual(a: LockSnapshot, b: LockSnapshot): boolean { + return a.pid === b.pid && a.ts === b.ts && a.owner === b.owner; + } + + function cleanupLockDirIfUnchanged(lockDir: string, snapshot: LockSnapshot): void { + const current = readLockMetadata(lockDir); + if (snapshotsEqual(current, snapshot)) { + cleanupLockDir(lockDir); + } + } + + function isErrnoException(err: unknown): err is NodeJS.ErrnoException { + return typeof err === "object" && err !== null && "code" in err; + } + + function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); + } + + /** + * Run `fn` while holding the client's ingest lock. Throws the + * client-specific `LockBusyError` when the lock cannot be acquired + * within `cfg.lockAcquireAttempts` attempts. + * + * The lock is a directory at `lockDir`, with three sentinel files + * (`pid`, `ts`, `owner`) recording who currently holds it. Stale locks + * are detected via PID liveness, TTL, and a grace window for the + * race between mkdir and writing pid/ts. + */ + async function withLock(fn: () => Promise | T): Promise { + const lock = lockConfig(); + mkdirSync(dirname(lock.lockDir), { recursive: true }); + const ownerToken = `${process.pid}-${Math.floor(Date.now() / 1000)}-${randomInt(1000, 10000)}`; + + let acquired = false; + for (let attempt = 0; attempt < cfg.lockAcquireAttempts; attempt++) { + try { + mkdirSync(lock.lockDir); + } catch (err) { + if (isErrnoException(err) && err.code === "EEXIST") { + const { stale, snapshot } = lockIsStale(lock); + if (stale) { + cleanupLockDirIfUnchanged(lock.lockDir, snapshot); + } + } + await sleep(LOCK_ACQUIRE_BACKOFF_MS); + continue; + } + + try { + writeFileSync(join(lock.lockDir, "ts"), String(Math.floor(Date.now() / 1000)), { + encoding: "utf8", + }); + writeFileSync(join(lock.lockDir, "pid"), String(process.pid), { encoding: "utf8" }); + writeFileSync(join(lock.lockDir, "owner"), ownerToken, { encoding: "utf8" }); + acquired = true; + break; + } catch { + cleanupLockDir(lock.lockDir); + await sleep(LOCK_ACQUIRE_BACKOFF_MS); + } + } + + if (!acquired) { + throw new LockBusyError(); + } + + try { + return await fn(); + } finally { + const currentOwner = readFileTrimmedOrEmpty(join(lock.lockDir, "owner")); + if (currentOwner === ownerToken) { + cleanupLockDir(lock.lockDir); + } + } + } + + /** + * Persist a payload to the spool directory using a tmp+rename so that + * a partially-written file is never visible to the drainer. Returns + * true on success, false on any I/O failure. + */ + function spoolPayload(payload: Record): boolean { + const dir = spoolDir(); + try { + mkdirSync(dir, { recursive: true }); + } catch { + logHookEvent(`${cfg.logPrefix} failed to create spool dir`); + return false; + } + + const payloadText = JSON.stringify(payload); + const tmpName = `.hook-tmp-${process.pid}-${Date.now()}-${randomInt(1000, 10000)}.json`; + const tmpPath = join(dir, tmpName); + try { + writeFileSync(tmpPath, payloadText, { encoding: "utf8" }); + } catch { + logHookEvent(`${cfg.logPrefix} failed to allocate spool temp file`); + return false; + } + + const finalName = `hook-${Math.floor(Date.now() / 1000)}-${process.pid}-${randomInt(1000, 10000)}.json`; + const finalPath = join(dir, finalName); + try { + renameSync(tmpPath, finalPath); + } catch { + try { + unlinkSync(tmpPath); + } catch { + // best-effort + } + logHookEvent(`${cfg.logPrefix} failed to spool payload`); + return false; + } + logHookEvent(`${cfg.logPrefix} spooled payload: ${finalPath}`); + return true; + } + + /** + * Promote any `.hook-tmp-*.json` files older than `ttlSeconds` to a + * recovered name so they are picked up by the next drain. Caller is + * responsible for passing the same TTL used by lock acquisition so + * that an in-flight write inside an active locked region is never + * mistaken for a crashed-writer leftover. + */ + function recoverStaleTmpSpool(ttlSeconds: number): void { + const dir = spoolDir(); + try { + mkdirSync(dir, { recursive: true }); + } catch { + return; + } + + let entries: string[]; + try { + entries = readdirSync(dir); + } catch { + return; + } + + const nowS = Date.now() / 1000; + for (const name of entries) { + if (!name.startsWith(".hook-tmp-") || !name.endsWith(".json")) continue; + const tmpPath = join(dir, name); + let mtimeS: number; + try { + mtimeS = statSync(tmpPath).mtimeMs / 1000; + } catch { + continue; + } + if (nowS - mtimeS <= ttlSeconds) continue; + + const recoveredName = `hook-recovered-${Math.floor(nowS)}-${process.pid}-${randomInt(1000, 10000)}.json`; + const recoveredPath = join(dir, recoveredName); + try { + renameSync(tmpPath, recoveredPath); + logHookEvent(`${cfg.logPrefix} recovered stale temp spool payload: ${recoveredPath}`); + } catch { + // best-effort + } + } + } + + /** + * Move a permanently-broken spool entry out of the queue so that it + * stops being picked up by future drains. The entry is renamed in + * place with a `.bad--` prefix so an operator can inspect or + * delete it manually. + */ + function quarantineSpoolEntry(dir: string, name: string, reason: string): void { + const sourcePath = join(dir, name); + const quarantineName = `.bad-${reason}-${Date.now()}-${randomInt(1000, 10000)}-${name}`; + try { + renameSync(sourcePath, join(dir, quarantineName)); + logHookEvent( + `${cfg.logPrefix} quarantined corrupt spool payload (${reason}): ${quarantineName}`, + ); + } catch { + // If rename fails, fall back to delete; either way the broken + // entry must not stay in the active queue. + try { + unlinkSync(sourcePath); + logHookEvent(`${cfg.logPrefix} dropped corrupt spool payload (${reason}): ${name}`); + } catch { + // best-effort + } + } + } + + /** + * Process every queued payload in the spool directory in lexicographic + * order (which approximates oldest-first because filenames embed the + * second-precision creation timestamp). The handler returns true to + * indicate the payload has been durably accepted; only then is the + * spool entry deleted. Failed entries are left on disk for the next + * drain attempt. + */ + async function drainSpool(handler: SpoolHandler): Promise { + const dir = spoolDir(); + try { + mkdirSync(dir, { recursive: true }); + } catch { + return { processed: 0, failed: 0 }; + } + + let entries: string[]; + try { + entries = readdirSync(dir); + } catch { + return { processed: 0, failed: 0 }; + } + + const queued = entries + .filter( + (name) => + name.endsWith(".json") && !name.startsWith(".hook-tmp-") && !name.startsWith(".bad-"), + ) + .sort(); + + const result: SpoolDrainResult = { processed: 0, failed: 0 }; + for (const name of queued) { + const path = join(dir, name); + let raw: string; + try { + raw = readFileSync(path, "utf8"); + } catch { + // Genuine I/O failure — leave the file alone so the next drain + // can retry, and surface the failure to the plugin log. + logHookEvent(`${cfg.logPrefix} failed to read spooled payload: ${path}`); + result.failed++; + continue; + } + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + // Permanently corrupt content — keeping the file around would + // loop forever every drain. Quarantine it under a `.bad-` prefix + // so an operator can inspect it without it being picked up again. + quarantineSpoolEntry(dir, name, "parse-error"); + result.failed++; + continue; + } + if (parsed == null || typeof parsed !== "object" || Array.isArray(parsed)) { + // Parseable but wrong shape — same problem, same fix. + quarantineSpoolEntry(dir, name, "wrong-shape"); + result.failed++; + continue; + } + + let ok = false; + try { + ok = await handler(parsed as Record); + } catch { + ok = false; + } + + if (ok) { + try { + unlinkSync(path); + result.processed++; + } catch { + // best-effort + } + } else { + logHookEvent(`${cfg.logPrefix} failed processing spooled payload: ${path}`); + result.failed++; + } + } + return result; + } + + return { + LockBusyError, + spoolDir, + lockTtlSeconds, + hasSpooledEntries, + withLock, + spoolPayload, + recoverStaleTmpSpool, + drainSpool, + }; +} diff --git a/packages/core/src/filters.ts b/packages/core/src/filters.ts index f9c234e37..09330a4b5 100644 --- a/packages/core/src/filters.ts +++ b/packages/core/src/filters.ts @@ -248,8 +248,10 @@ export function buildFilterClausesWithContext( addScopeVisibilityFilter(clauses, params, ownership); if (!filters) return result; - // Single kind filter - if (filters.kind) { + // Single kind filter — guard against non-string kinds from untyped + // JSON request bodies (a number/bool would otherwise become a broken + // bound parameter). + if (typeof filters.kind === "string" && filters.kind) { clauses.push("memory_items.kind = ?"); params.push(filters.kind); } diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 1ae657e49..ed144d6fa 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -533,6 +533,7 @@ export { startMaintenanceJob, updateMaintenanceJob, } from "./maintenance-jobs.js"; +export * from "./memory-kinds.js"; export type { DerivedMemoryRole, DerivedMemoryRoleResult, @@ -620,6 +621,15 @@ export { buildMemoryPackWithTraceAsync, estimateTokens, } from "./pack.js"; +export type { PiFlushSignal, PiHookAdapterEvent, PiHookRawEventEnvelope } from "./pi-hooks.js"; +export { + buildIngestPayloadFromPiEvent, + buildPiFlushSignalFromEvent, + buildRawEventEnvelopeFromPiEvent, + MAPPABLE_PI_EVENTS, + mapPiEventPayload, + PI_FLUSH_ONLY_EVENTS, +} from "./pi-hooks.js"; export type { BlockedPolicyTeamDeviceEligibilityResult, DerivePolicyTeamDeviceEligibilityInput, diff --git a/packages/core/src/ingest-pipeline.ts b/packages/core/src/ingest-pipeline.ts index 45c81d411..2d20f5cfc 100644 --- a/packages/core/src/ingest-pipeline.ts +++ b/packages/core/src/ingest-pipeline.ts @@ -65,6 +65,7 @@ import { shouldPreferRepairedObserverResponse, shouldRepairObserverResponse, } from "./ingest-xml-parser.js"; +import { REMEMBER_MEMORY_KINDS } from "./memory-kinds.js"; import { type ObserverClient, ObserverClient as ObserverClientImpl } from "./observer-client.js"; import { resolveProject } from "./project.js"; import * as schema from "./schema.js"; @@ -78,15 +79,7 @@ import { storeVectors } from "./vectors.js"; // Allowed memory kinds (matches Python) // --------------------------------------------------------------------------- -const ALLOWED_KINDS = new Set([ - "bugfix", - "feature", - "refactor", - "change", - "discovery", - "decision", - "exploration", -]); +const ALLOWED_KINDS = new Set(REMEMBER_MEMORY_KINDS); // --------------------------------------------------------------------------- // Path normalization diff --git a/packages/core/src/ingest-xml-parser.ts b/packages/core/src/ingest-xml-parser.ts index 203003ed4..69fe31fa1 100644 --- a/packages/core/src/ingest-xml-parser.ts +++ b/packages/core/src/ingest-xml-parser.ts @@ -10,6 +10,7 @@ import { isLowSignalObservation } from "./ingest-filters.js"; import type { ParsedObservation, ParsedOutput, ParsedSummary } from "./ingest-types.js"; +import { REMEMBER_MEMORY_KINDS } from "./memory-kinds.js"; // --------------------------------------------------------------------------- // Regex patterns @@ -43,15 +44,7 @@ const OBSERVATION_CONCEPTS = new Set([ "trade-off", ]); -export const SUPPORTED_OBSERVATION_KINDS = new Set([ - "bugfix", - "feature", - "refactor", - "change", - "discovery", - "decision", - "exploration", -]); +export const SUPPORTED_OBSERVATION_KINDS = new Set(REMEMBER_MEMORY_KINDS); export interface ObserverResponseStructuralDiagnostics { recognizedOutput: boolean; diff --git a/packages/core/src/memory-kinds.test.ts b/packages/core/src/memory-kinds.test.ts new file mode 100644 index 000000000..bf60cc053 --- /dev/null +++ b/packages/core/src/memory-kinds.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from "vitest"; +import { + ALLOWED_MEMORY_KINDS, + MEMORY_KIND_DESCRIPTIONS, + REMEMBER_MEMORY_KINDS, + validateMemoryKind, +} from "./memory-kinds.js"; + +describe("memory kind catalog", () => { + it("defines exactly the seven remember kinds", () => { + expect(Object.keys(MEMORY_KIND_DESCRIPTIONS)).toEqual([ + "discovery", + "change", + "feature", + "bugfix", + "refactor", + "decision", + "exploration", + ]); + expect(REMEMBER_MEMORY_KINDS).toEqual(Object.keys(MEMORY_KIND_DESCRIPTIONS)); + }); + + it("keeps session_summary out of the remember kinds but in the allowed set", () => { + expect(REMEMBER_MEMORY_KINDS).not.toContain("session_summary"); + expect(Object.keys(MEMORY_KIND_DESCRIPTIONS)).not.toContain("session_summary"); + expect(ALLOWED_MEMORY_KINDS.has("session_summary")).toBe(true); + expect(ALLOWED_MEMORY_KINDS.size).toBe(REMEMBER_MEMORY_KINDS.length + 1); + }); + + it("validateMemoryKind accepts session_summary and the remember kinds", () => { + expect(validateMemoryKind("session_summary")).toBe("session_summary"); + expect(validateMemoryKind(" Decision ")).toBe("decision"); + }); + + it("validateMemoryKind rejects unknown kinds", () => { + expect(() => validateMemoryKind("not-a-kind")).toThrow(/Invalid memory kind/); + }); +}); diff --git a/packages/core/src/memory-kinds.ts b/packages/core/src/memory-kinds.ts new file mode 100644 index 000000000..1bb4cce62 --- /dev/null +++ b/packages/core/src/memory-kinds.ts @@ -0,0 +1,39 @@ +/** + * Canonical memory-kind catalog. The seven kinds below are the ones the + * MCP remember tools accept; the store additionally allows + * `session_summary` (written by the observer pipeline). Every surface + * (MCP schema, store validation, viewer routes) imports this catalog so + * the kinds never drift. + */ + +export const MEMORY_KIND_DESCRIPTIONS = { + discovery: "Something learned about the codebase, architecture, or tools", + change: "A code change that was made", + feature: "A new feature that was implemented", + bugfix: "A bug that was found and fixed", + refactor: "Code that was refactored or restructured", + decision: "A design or architecture decision", + exploration: "An experiment or investigation (may not have shipped)", +} as const satisfies Record; + +export type RememberMemoryKind = keyof typeof MEMORY_KIND_DESCRIPTIONS; + +/** The seven kinds the MCP remember tools accept. */ +export const REMEMBER_MEMORY_KINDS = Object.keys(MEMORY_KIND_DESCRIPTIONS) as [ + RememberMemoryKind, + ...RememberMemoryKind[], +]; + +/** Kinds accepted by the store: remembers plus the observer's session_summary. */ +export const ALLOWED_MEMORY_KINDS = new Set([...REMEMBER_MEMORY_KINDS, "session_summary"]); + +/** Normalize and validate a memory kind. Throws on invalid kinds. */ +export function validateMemoryKind(kind: string): string { + const normalized = kind.trim().toLowerCase(); + if (!ALLOWED_MEMORY_KINDS.has(normalized)) { + throw new Error( + `Invalid memory kind "${kind}". Allowed: ${[...ALLOWED_MEMORY_KINDS].join(", ")}`, + ); + } + return normalized; +} diff --git a/packages/core/src/pi-hooks.test.ts b/packages/core/src/pi-hooks.test.ts new file mode 100644 index 000000000..89f7b7d44 --- /dev/null +++ b/packages/core/src/pi-hooks.test.ts @@ -0,0 +1,776 @@ +/** + * Tests for pi-hooks.ts — AdapterEvent v1 mapping for pi extension events. + * + * Covers: + * - mapPiEventPayload: all mappable event types, skip cases, deterministic ids, + * tool_result isError, fork → new stream identity + * - buildPiFlushSignalFromEvent: session_before_compact flush-only contract + * - buildRawEventEnvelopeFromPiEvent: envelope shape + source "pi" + * - buildIngestPayloadFromPiEvent: session context fields + * - store attribution: pi-ingested rows carry source = "pi" (no opencode rows) + */ + +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { connect } from "./db.js"; +import { + buildIngestPayloadFromPiEvent, + buildPiFlushSignalFromEvent, + buildRawEventEnvelopeFromPiEvent, + MAPPABLE_PI_EVENTS, + mapPiEventPayload, +} from "./pi-hooks.js"; +import { ingestRawEvents } from "./raw-event-ingest.js"; +import { MemoryStore } from "./store.js"; +import { initTestSchema } from "./test-utils.js"; + +function requireEnvelope(envelope: ReturnType) { + if (envelope === null) { + throw new Error("expected pi envelope"); + } + return envelope; +} + +// --------------------------------------------------------------------------- +// mapPiEventPayload — event type mapping +// --------------------------------------------------------------------------- + +describe("mapPiEventPayload", () => { + describe("session_start → session_start", () => { + it("maps session start with deterministic id", () => { + const event = mapPiEventPayload({ + piEvent: "session_start", + sessionId: "pi-sess-1", + cwd: "/tmp/repo", + ts: "2026-06-01T12:00:00Z", + }); + + expect(event).not.toBeNull(); + expect(event?.schema_version).toBe("1.0"); + expect(event?.source).toBe("pi"); + expect(event?.event_type).toBe("session_start"); + expect(event?.session_id).toBe("pi-sess-1"); + expect(event?.event_id).toBe("pi:pi-sess-1:session_start"); + expect(event?.cwd).toBe("/tmp/repo"); + expect(event?.ts).toBe("2026-06-01T12:00:00Z"); + }); + + it("prefers entryId for the event id suffix when present", () => { + const event = mapPiEventPayload({ + piEvent: "session_start", + sessionId: "pi-sess-1", + entryId: "entry-start-1", + ts: "2026-06-01T12:00:00Z", + }); + expect(event?.event_id).toBe("pi:pi-sess-1:entry-start-1"); + }); + }); + + describe("session_shutdown → session_end", () => { + it("maps reason field", () => { + const event = mapPiEventPayload({ + piEvent: "session_shutdown", + sessionId: "pi-sess-end", + reason: "user_exit", + ts: "2026-06-01T13:00:00Z", + }); + + expect(event).not.toBeNull(); + expect(event?.event_type).toBe("session_end"); + expect(event?.payload.reason).toBe("user_exit"); + expect(event?.event_id).toBe("pi:pi-sess-end:session_end"); + expect(event?.source).toBe("pi"); + }); + }); + + describe("message_end role=user → prompt", () => { + it("maps prompt text and meta", () => { + const event = mapPiEventPayload({ + piEvent: "message_end", + sessionId: "pi-sess-msg", + entryId: "entry-u1", + role: "user", + text: "Run tests", + cwd: "/tmp/repo", + custom_field: "keep-me", + ts: "2026-06-01T12:01:00Z", + }); + + expect(event).not.toBeNull(); + expect(event?.source).toBe("pi"); + expect(event?.event_type).toBe("prompt"); + expect(event?.payload.text).toBe("Run tests"); + expect(event?.event_id).toBe("pi:pi-sess-msg:entry-u1"); + expect(event?.meta.pi_event).toBe("message_end"); + expect(event?.meta.entry_id).toBe("entry-u1"); + expect((event?.meta.pi_fields as Record).custom_field).toBe("keep-me"); + }); + + it("returns null for empty text", () => { + expect( + mapPiEventPayload({ + piEvent: "message_end", + sessionId: "pi-sess-msg", + entryId: "entry-u1", + role: "user", + text: " ", + }), + ).toBeNull(); + }); + + it("returns null without entryId", () => { + expect( + mapPiEventPayload({ + piEvent: "message_end", + sessionId: "pi-sess-msg", + role: "user", + text: "hello", + }), + ).toBeNull(); + }); + }); + + describe("message_end role=assistant → assistant", () => { + it("maps assistant text", () => { + const event = mapPiEventPayload({ + piEvent: "message_end", + sessionId: "pi-sess-msg", + entryId: "entry-a1", + role: "assistant", + text: "All done", + ts: "2026-06-01T12:02:00Z", + }); + + expect(event).not.toBeNull(); + expect(event?.event_type).toBe("assistant"); + expect(event?.payload.text).toBe("All done"); + expect(event?.event_id).toBe("pi:pi-sess-msg:entry-a1"); + }); + }); + + describe("turn_end → assistant", () => { + it("maps turn_end text", () => { + const event = mapPiEventPayload({ + piEvent: "turn_end", + sessionId: "pi-sess-turn", + entryId: "entry-t1", + text: "Turn complete", + ts: "2026-06-01T12:03:00Z", + }); + expect(event?.event_type).toBe("assistant"); + expect(event?.payload.text).toBe("Turn complete"); + expect(event?.event_id).toBe("pi:pi-sess-turn:entry-t1"); + }); + + it("does not map agent_end (D2: extension emits message_end only)", () => { + expect(MAPPABLE_PI_EVENTS.has("agent_end")).toBe(false); + expect( + mapPiEventPayload({ + piEvent: "agent_end", + sessionId: "pi-sess-agent", + entryId: "entry-ag1", + text: "Agent finished", + ts: "2026-06-01T12:04:00Z", + }), + ).toBeNull(); + }); + }); + + describe("tool_call → tool_call", () => { + it("maps tool name, input, and toolCallId", () => { + const event = mapPiEventPayload({ + piEvent: "tool_call", + sessionId: "pi-sess-tool", + toolCallId: "tc-1", + toolName: "bash", + toolInput: { command: "pnpm test" }, + ts: "2026-06-01T12:05:00Z", + }); + + expect(event).not.toBeNull(); + expect(event?.event_type).toBe("tool_call"); + expect(event?.payload.tool_name).toBe("bash"); + expect(event?.payload.tool_input).toEqual({ command: "pnpm test" }); + expect(event?.event_id).toBe("pi:pi-sess-tool:tc-1"); + expect(event?.meta.tool_call_id).toBe("tc-1"); + }); + + it("defaults tool_input to {} when missing", () => { + const event = mapPiEventPayload({ + piEvent: "tool_call", + sessionId: "pi-sess-tool", + toolCallId: "tc-2", + toolName: "read", + ts: "2026-06-01T12:05:00Z", + }); + expect(event?.payload.tool_input).toEqual({}); + }); + + it("returns null for missing toolName", () => { + expect( + mapPiEventPayload({ + piEvent: "tool_call", + sessionId: "pi-sess-tool", + toolCallId: "tc-3", + }), + ).toBeNull(); + }); + + it("returns null without toolCallId or entryId", () => { + expect( + mapPiEventPayload({ + piEvent: "tool_call", + sessionId: "pi-sess-tool", + toolName: "bash", + }), + ).toBeNull(); + }); + }); + + describe("tool_result → tool_result (isError)", () => { + it("maps ok result", () => { + const event = mapPiEventPayload({ + piEvent: "tool_result", + sessionId: "pi-sess-tool", + toolCallId: "tc-1", + toolName: "bash", + toolOutput: { exit_code: 0 }, + isError: false, + ts: "2026-06-01T12:06:00Z", + }); + + expect(event).not.toBeNull(); + expect(event?.event_type).toBe("tool_result"); + expect(event?.payload.status).toBe("ok"); + expect(event?.payload.tool_output).toEqual({ exit_code: 0 }); + expect(event?.payload.tool_error).toBeNull(); + expect(event?.event_id).toBe("pi:pi-sess-tool:tc-1:result"); + }); + + it("maps isError result", () => { + const event = mapPiEventPayload({ + piEvent: "tool_result", + sessionId: "pi-sess-tool", + toolCallId: "tc-err", + toolName: "bash", + isError: true, + error: { message: "1 failed" }, + ts: "2026-06-01T12:07:00Z", + }); + + expect(event).not.toBeNull(); + expect(event?.event_type).toBe("tool_result"); + expect(event?.payload.status).toBe("error"); + expect(event?.payload.tool_output).toBeNull(); + expect(event?.payload.error).toEqual({ message: "1 failed" }); + expect(event?.payload.tool_error).toEqual({ message: "1 failed" }); + expect(event?.event_id).toBe("pi:pi-sess-tool:tc-err:result"); + }); + }); + + describe("skip cases", () => { + it("returns null for unsupported event type", () => { + expect( + mapPiEventPayload({ + piEvent: "before_agent_start", + sessionId: "pi-sess-1", + }), + ).toBeNull(); + }); + + it("returns null for missing sessionId", () => { + expect( + mapPiEventPayload({ + piEvent: "session_start", + }), + ).toBeNull(); + }); + + it("returns null for empty sessionId", () => { + expect( + mapPiEventPayload({ + piEvent: "session_start", + sessionId: " ", + }), + ).toBeNull(); + }); + + it("returns null for session_before_compact (flush-only, not transcript)", () => { + expect( + mapPiEventPayload({ + piEvent: "session_before_compact", + sessionId: "pi-sess-1", + ts: "2026-06-01T12:00:00Z", + }), + ).toBeNull(); + }); + + it("returns null for unknown role on message_end", () => { + expect( + mapPiEventPayload({ + piEvent: "message_end", + sessionId: "pi-sess-1", + entryId: "e1", + role: "system", + text: "nope", + }), + ).toBeNull(); + }); + }); + + describe("deterministic event ids", () => { + it("produces identical ids for identical payloads", () => { + const payload = { + piEvent: "message_end", + sessionId: "pi-sess-stable", + entryId: "entry-stable-1", + role: "user", + text: "hello", + ts: "2026-06-01T12:00:00Z", + }; + const first = mapPiEventPayload(payload); + const second = mapPiEventPayload(payload); + expect(first?.event_id).toBe(second?.event_id); + expect(first?.event_id).toBe("pi:pi-sess-stable:entry-stable-1"); + }); + + it("event_id is invariant to ts (different or absent)", () => { + // Dedup key must not incorporate wall-clock ts — retries with a fresh + // clock or missing ts must collapse to the same raw-event identity. + const base = { + piEvent: "message_end", + sessionId: "pi-sess-ts-invariant", + entryId: "entry-ts-1", + role: "assistant", + text: "same logical message", + }; + const withTsA = mapPiEventPayload({ ...base, ts: "2026-01-01T00:00:00Z" }); + const withTsB = mapPiEventPayload({ ...base, ts: "2026-12-31T23:59:59Z" }); + const withoutTs = mapPiEventPayload({ ...base }); + expect(withTsA?.event_id).toBe("pi:pi-sess-ts-invariant:entry-ts-1"); + expect(withTsB?.event_id).toBe(withTsA?.event_id); + expect(withoutTs?.event_id).toBe(withTsA?.event_id); + // ts itself may differ; only event_id must be stable. + expect(withTsA?.ts).not.toBe(withTsB?.ts); + }); + + it("uses the pi:: format", () => { + const event = mapPiEventPayload({ + piEvent: "tool_call", + sessionId: "S", + toolCallId: "T", + toolName: "read", + ts: "2026-06-01T12:00:00Z", + }); + expect(event?.event_id).toMatch(/^pi:[^:]+:.+$/); + expect(event?.event_id).toBe("pi:S:T"); + }); + }); + + describe("fork id change → new stream identity", () => { + it("different sessionId yields different event_id and session_id", () => { + const base = { + piEvent: "message_end" as const, + entryId: "entry-same", + role: "user", + text: "same text", + ts: "2026-06-01T12:00:00Z", + }; + const parent = mapPiEventPayload({ ...base, sessionId: "sess-parent" }); + const fork = mapPiEventPayload({ ...base, sessionId: "sess-fork" }); + + expect(parent?.session_id).toBe("sess-parent"); + expect(fork?.session_id).toBe("sess-fork"); + expect(parent?.event_id).toBe("pi:sess-parent:entry-same"); + expect(fork?.event_id).toBe("pi:sess-fork:entry-same"); + expect(parent?.event_id).not.toBe(fork?.event_id); + }); + + it("envelope stream identity follows the forked session id", () => { + const parentEnv = buildRawEventEnvelopeFromPiEvent({ + piEvent: "session_start", + sessionId: "sess-parent", + ts: "2026-06-01T12:00:00Z", + }); + const forkEnv = buildRawEventEnvelopeFromPiEvent({ + piEvent: "session_start", + sessionId: "sess-fork", + ts: "2026-06-01T12:00:00Z", + }); + + expect(parentEnv?.session_stream_id).toBe("sess-parent"); + expect(forkEnv?.session_stream_id).toBe("sess-fork"); + expect(parentEnv?.session_stream_id).not.toBe(forkEnv?.session_stream_id); + expect(parentEnv?.source).toBe("pi"); + expect(forkEnv?.source).toBe("pi"); + }); + }); + + describe("snake_case field aliases", () => { + it("accepts session_id / pi_event / entry_id aliases", () => { + const event = mapPiEventPayload({ + pi_event: "message_end", + session_id: "pi-snake", + entry_id: "e-snake", + role: "user", + text: "aliased", + ts: "2026-06-01T12:00:00Z", + }); + expect(event?.session_id).toBe("pi-snake"); + expect(event?.event_id).toBe("pi:pi-snake:e-snake"); + expect(event?.source).toBe("pi"); + }); + }); +}); + +// --------------------------------------------------------------------------- +// buildPiFlushSignalFromEvent +// --------------------------------------------------------------------------- + +describe("buildPiFlushSignalFromEvent", () => { + it("returns a flush signal for session_before_compact", () => { + const signal = buildPiFlushSignalFromEvent({ + piEvent: "session_before_compact", + sessionId: "pi-sess-compact", + cwd: "/tmp/repo", + project: "repo", + ts: "2026-06-01T14:00:00Z", + }); + + expect(signal).not.toBeNull(); + expect(signal?.kind).toBe("flush"); + expect(signal?.reason).toBe("session_before_compact"); + expect(signal?.source).toBe("pi"); + expect(signal?.session_id).toBe("pi-sess-compact"); + expect(signal?.ts).toBe("2026-06-01T14:00:00Z"); + // Must never look like a compaction object for pi to apply. + expect(signal && "compaction" in signal).toBe(false); + }); + + it("returns null for transcript events", () => { + expect( + buildPiFlushSignalFromEvent({ + piEvent: "session_start", + sessionId: "pi-sess-1", + }), + ).toBeNull(); + }); + + it("returns null without sessionId", () => { + expect( + buildPiFlushSignalFromEvent({ + piEvent: "session_before_compact", + }), + ).toBeNull(); + }); + + it("does not produce a raw envelope or ingest payload for compaction", () => { + const payload = { + piEvent: "session_before_compact", + sessionId: "pi-sess-compact", + ts: "2026-06-01T14:00:00Z", + }; + expect(mapPiEventPayload(payload)).toBeNull(); + expect(buildRawEventEnvelopeFromPiEvent(payload)).toBeNull(); + expect(buildIngestPayloadFromPiEvent(payload)).toBeNull(); + expect(buildPiFlushSignalFromEvent(payload)).not.toBeNull(); + }); +}); + +// --------------------------------------------------------------------------- +// buildRawEventEnvelopeFromPiEvent +// --------------------------------------------------------------------------- + +describe("buildRawEventEnvelopeFromPiEvent", () => { + it("returns null for unsupported event", () => { + expect( + buildRawEventEnvelopeFromPiEvent({ + piEvent: "before_agent_start", + sessionId: "pi-sess-1", + }), + ).toBeNull(); + }); + + it("wraps adapter events for raw-event ingestion with source pi", () => { + const envelope = buildRawEventEnvelopeFromPiEvent({ + piEvent: "session_start", + sessionId: "pi-sess-env", + ts: "2026-06-01T12:00:00Z", + cwd: "/tmp/repo", + project: "repo", + }); + + expect(envelope).not.toBeNull(); + expect(envelope?.source).toBe("pi"); + expect(envelope?.event_type).toBe("pi.hook"); + expect(envelope?.session_stream_id).toBe("pi-sess-env"); + expect(envelope?.session_id).toBe("pi-sess-env"); + expect(envelope?.opencode_session_id).toBe("pi-sess-env"); + expect(envelope?.started_at).toBe("2026-06-01T12:00:00Z"); + expect(envelope?.event_id).toBe("pi:pi-sess-env:session_start"); + expect(envelope?.payload.type).toBe("pi.hook"); + expect((envelope?.payload._adapter as Record).source).toBe("pi"); + expect((envelope?.payload._adapter as Record).schema_version).toBe("1.0"); + expect((envelope?.payload._adapter as Record).event_type).toBe( + "session_start", + ); + }); + + it("sets started_at only for session_start", () => { + const envelope = buildRawEventEnvelopeFromPiEvent({ + piEvent: "message_end", + sessionId: "pi-sess-env", + entryId: "e1", + role: "user", + text: "hi", + ts: "2026-06-01T12:01:00Z", + }); + expect(envelope?.started_at).toBeNull(); + }); +}); + +// --------------------------------------------------------------------------- +// buildIngestPayloadFromPiEvent +// --------------------------------------------------------------------------- + +describe("buildIngestPayloadFromPiEvent", () => { + it("returns null for unsupported event", () => { + expect( + buildIngestPayloadFromPiEvent({ + piEvent: "unknown", + sessionId: "pi-sess-1", + }), + ).toBeNull(); + }); + + it("wraps adapter event in session_context with source pi and all aliases", () => { + const ingest = buildIngestPayloadFromPiEvent({ + piEvent: "session_start", + sessionId: "pi-sess-xyz", + cwd: "/tmp/repo", + ts: "2026-06-01T12:00:00Z", + }); + + expect(ingest).not.toBeNull(); + const ctx = ingest?.session_context as Record; + expect(ctx.source).toBe("pi"); + expect(ctx.stream_id).toBe("pi-sess-xyz"); + expect(ctx.session_stream_id).toBe("pi-sess-xyz"); + expect(ctx.session_id).toBe("pi-sess-xyz"); + expect(ctx.opencode_session_id).toBe("pi-sess-xyz"); + + const events = ingest?.events as Array>; + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe("pi.hook"); + expect((events[0]?._adapter as Record).source).toBe("pi"); + expect((events[0]?._adapter as Record).event_type).toBe("session_start"); + }); + + it("sets cwd from pi payload", () => { + const ingest = buildIngestPayloadFromPiEvent({ + piEvent: "message_end", + sessionId: "pi-sess-cwd", + entryId: "e-cwd", + role: "user", + text: "hello", + cwd: "/home/user/myrepo", + ts: "2026-06-01T12:00:00Z", + }); + expect(ingest?.cwd).toBe("/home/user/myrepo"); + }); +}); + +// --------------------------------------------------------------------------- +// Attribution: pi-ingested rows carry source = "pi" +// --------------------------------------------------------------------------- + +describe("pi source attribution via recordRawEvent", () => { + let tmpDir: string; + let store: MemoryStore; + + beforeEach(() => { + tmpDir = mkdtempSync(join(tmpdir(), "codemem-pi-hooks-test-")); + const dbPath = join(tmpDir, "test.sqlite"); + const db = connect(dbPath); + initTestSchema(db); + db.close(); + store = new MemoryStore(dbPath); + }); + + afterEach(() => { + store.close(); + rmSync(tmpDir, { recursive: true, force: true }); + }); + + it("stores raw events with source=pi and creates no opencode rows", () => { + const envelope = requireEnvelope( + buildRawEventEnvelopeFromPiEvent({ + piEvent: "message_end", + sessionId: "pi-attr-sess", + entryId: "entry-attr-1", + role: "user", + text: "attribute me to pi", + ts: "2026-06-01T15:00:00Z", + cwd: tmpDir, + project: "codemem", + }), + ); + // Explicit source from envelope — never rely on recordRawEvent default. + expect(envelope.source).toBe("pi"); + + const inserted = store.recordRawEvent({ + opencodeSessionId: envelope.session_stream_id, + source: envelope.source, + eventId: envelope.event_id, + eventType: envelope.event_type, + payload: envelope.payload, + tsWallMs: envelope.ts_wall_ms, + }); + expect(inserted).toBe(true); + + const piRows = store.db + .prepare(`SELECT source, stream_id, event_id, event_type FROM raw_events WHERE source = ?`) + .all("pi") as Array<{ + source: string; + stream_id: string; + event_id: string; + event_type: string; + }>; + expect(piRows).toHaveLength(1); + expect(piRows[0]?.source).toBe("pi"); + expect(piRows[0]?.stream_id).toBe("pi-attr-sess"); + expect(piRows[0]?.event_id).toBe("pi:pi-attr-sess:entry-attr-1"); + expect(piRows[0]?.event_type).toBe("pi.hook"); + + const opencodeRows = store.db + .prepare(`SELECT COUNT(*) AS n FROM raw_events WHERE source = ?`) + .get("opencode") as { n: number }; + expect(Number(opencodeRows.n)).toBe(0); + + const sessionRows = store.db + .prepare(`SELECT source, stream_id FROM raw_event_sessions`) + .all() as Array<{ source: string; stream_id: string }>; + expect(sessionRows).toHaveLength(1); + expect(sessionRows[0]?.source).toBe("pi"); + expect(sessionRows[0]?.stream_id).toBe("pi-attr-sess"); + }); + + it("dedupes retries by (source, stream, event_id) for pi", () => { + const envelope = requireEnvelope( + buildRawEventEnvelopeFromPiEvent({ + piEvent: "tool_call", + sessionId: "pi-dedupe-sess", + toolCallId: "tc-dedupe", + toolName: "read", + toolInput: { path: "README.md" }, + ts: "2026-06-01T15:01:00Z", + }), + ); + + const write = () => + store.recordRawEvent({ + opencodeSessionId: envelope.session_stream_id, + source: envelope.source, + eventId: envelope.event_id, + eventType: envelope.event_type, + payload: envelope.payload, + tsWallMs: envelope.ts_wall_ms, + }); + expect(write()).toBe(true); + expect(write()).toBe(false); + + const count = store.db + .prepare(`SELECT COUNT(*) AS n FROM raw_events WHERE source = ? AND stream_id = ?`) + .get("pi", "pi-dedupe-sess") as { n: number }; + expect(Number(count.n)).toBe(1); + }); + + it("forked session id creates a separate pi stream partition", () => { + for (const sessionId of ["sess-parent", "sess-fork"]) { + const envelope = requireEnvelope( + buildRawEventEnvelopeFromPiEvent({ + piEvent: "message_end", + sessionId, + entryId: "entry-shared-logical", + role: "user", + text: "fork test", + ts: "2026-06-01T15:02:00Z", + }), + ); + store.recordRawEvent({ + opencodeSessionId: envelope.session_stream_id, + source: envelope.source, + eventId: envelope.event_id, + eventType: envelope.event_type, + payload: envelope.payload, + tsWallMs: envelope.ts_wall_ms, + }); + } + + const rows = store.db + .prepare( + `SELECT source, stream_id, event_id FROM raw_events WHERE source = ? ORDER BY stream_id`, + ) + .all("pi") as Array<{ source: string; stream_id: string; event_id: string }>; + expect(rows).toHaveLength(2); + expect(rows.map((r) => r.stream_id).sort()).toEqual(["sess-fork", "sess-parent"]); + expect(rows.every((r) => r.source === "pi")).toBe(true); + expect(new Set(rows.map((r) => r.event_id)).size).toBe(2); + + const opencodeCount = store.db + .prepare(`SELECT COUNT(*) AS n FROM raw_events WHERE source = ?`) + .get("opencode") as { n: number }; + expect(Number(opencodeCount.n)).toBe(0); + }); +}); + +describe("ingestRawEvents accepts pi envelopes", () => { + let tmpDir: string; + let store: MemoryStore; + + beforeEach(() => { + tmpDir = mkdtempSync(join(tmpdir(), "codemem-pi-hooks-ingest-")); + const dbPath = join(tmpDir, "test.sqlite"); + const db = connect(dbPath); + initTestSchema(db); + db.close(); + store = new MemoryStore(dbPath); + }); + + afterEach(() => { + store.close(); + rmSync(tmpDir, { recursive: true, force: true }); + }); + + it("persists source pi with no opencode rows", () => { + const envelope = requireEnvelope( + buildRawEventEnvelopeFromPiEvent({ + piEvent: "message_end", + sessionId: "pi-canonical-1", + entryId: "entry-1", + role: "user", + text: "hello", + ts: "2026-06-01T16:00:00Z", + }), + ); + const result = ingestRawEvents(store, envelope); + expect(result.inserted).toBe(1); + expect(result.skipped).toBe(0); + + const row = store.db.prepare(`SELECT source, stream_id, event_id FROM raw_events`).get() as { + source: string; + stream_id: string; + event_id: string; + }; + expect(row.source).toBe("pi"); + expect(row.stream_id).toBe("pi-canonical-1"); + expect(row.event_id).toBe(envelope.event_id); + + const opencodeCount = store.db + .prepare(`SELECT COUNT(*) AS n FROM raw_events WHERE source = ?`) + .get("opencode") as { n: number }; + expect(Number(opencodeCount.n)).toBe(0); + }); +}); diff --git a/packages/core/src/pi-hooks.ts b/packages/core/src/pi-hooks.ts new file mode 100644 index 000000000..70f262e74 --- /dev/null +++ b/packages/core/src/pi-hooks.ts @@ -0,0 +1,450 @@ +/** + * Pi extension event payload mapping. + * + * Normalizes pi coding-agent extension events into AdapterEvent v1 envelopes + * for the shared raw-event sweeper pipeline. Mirrors claude-hooks.ts / + * codex-hooks.ts structure with source hard-coded to "pi". + * + * Entry points: + * mapPiEventPayload(payload) → adapter event or null + * buildRawEventEnvelopeFromPiEvent(...) → raw event envelope or null + * buildIngestPayloadFromPiEvent(...) → ingest payload or null + * buildPiFlushSignalFromEvent(...) → flush signal (compaction only) + * + * session_before_compact is observe-only: it never becomes a transcript event + * and never returns a compaction object for pi to apply. + */ + +import { normalizeProjectLabel, resolveHookProject } from "./claude-hooks.js"; + +// --------------------------------------------------------------------------- +// Constants +// --------------------------------------------------------------------------- + +/** Pi extension events that map to AdapterEvent v1 transcript types. */ +export const MAPPABLE_PI_EVENTS = new Set([ + "session_start", + "session_shutdown", + "message_end", + "turn_end", + "tool_call", + "tool_result", +]); + +/** Events that only signal a boundary flush (never stored as transcript). */ +export const PI_FLUSH_ONLY_EVENTS = new Set(["session_before_compact"]); + +// --------------------------------------------------------------------------- +// Timestamp helpers +// --------------------------------------------------------------------------- + +function nowIso(): string { + return new Date().toISOString().replace(/\.(\d{3})\d*Z$/, ".$1Z"); +} + +function normalizeIsoTs(value: unknown): string | null { + if (typeof value !== "string") return null; + const text = value.trim(); + if (!text) return null; + const hasTimezone = + /[Zz]$/.test(text) || /[+-]\d{2}:\d{2}$/.test(text) || /[+-]\d{4}$/.test(text); + const parsed = new Date(hasTimezone ? text : `${text}Z`); + if (Number.isNaN(parsed.getTime())) return null; + const hasFractional = /\.\d+([Zz+-]|$)/.test(text); + return hasFractional + ? parsed.toISOString().replace(/\.(\d{3})Z$/, ".$1000Z") + : parsed.toISOString().replace(/\.\d{3}Z$/, "Z"); +} + +function isoToWallMs(value: string): number { + return new Date(value).getTime(); +} + +// --------------------------------------------------------------------------- +// Coercion helpers +// --------------------------------------------------------------------------- + +function coerceString(value: unknown): string { + return typeof value === "string" ? value.trim() : ""; +} + +/** Read the first defined field among camelCase / snake_case aliases. */ +function field(payload: Record, ...keys: string[]): unknown { + for (const key of keys) { + if (Object.hasOwn(payload, key) && payload[key] !== undefined) return payload[key]; + } + return undefined; +} + +function coerceSessionId(payload: Record): string | null { + const value = coerceString(field(payload, "sessionId", "session_id")); + return value || null; +} + +function coercePiEventName(payload: Record): string { + return coerceString(field(payload, "piEvent", "pi_event", "event", "type")); +} + +function objectOrEmpty(value: unknown): Record { + return value != null && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : {}; +} + +function coerceBool(value: unknown): boolean { + if (typeof value === "boolean") return value; + if (typeof value === "number") return value !== 0; + if (typeof value === "string") { + const t = value.trim().toLowerCase(); + return t === "true" || t === "1" || t === "yes"; + } + return false; +} + +/** + * Deterministic event id: `pi::`. + * Same logical event must yield the same id across HTTP/CLI retries. + */ +function buildPiEventId(sessionId: string, stablePart: string): string { + const part = stablePart.trim(); + if (!part) throw new Error("pi event id stable part is required"); + return `pi:${sessionId}:${part}`; +} + +// --------------------------------------------------------------------------- +// mapPiEventPayload +// --------------------------------------------------------------------------- + +export interface PiHookAdapterEvent { + schema_version: "1.0"; + source: "pi"; + session_id: string; + event_id: string; + event_type: string; + ts: string; + ordering_confidence: "low"; + cwd: string | null; + payload: Record; + meta: Record; +} + +/** + * Map a pi extension event payload to a normalized AdapterEvent v1. + * Returns null if the event type is unsupported, flush-only, or required + * fields are missing. + */ +export function mapPiEventPayload(payload: Record): PiHookAdapterEvent | null { + const piEvent = coercePiEventName(payload); + if (!piEvent) return null; + + // Compaction is observe-only — never a transcript AdapterEvent. + if (PI_FLUSH_ONLY_EVENTS.has(piEvent)) return null; + if (!MAPPABLE_PI_EVENTS.has(piEvent)) return null; + + const sessionId = coerceSessionId(payload); + if (!sessionId) return null; + + const normalizedRawTs = normalizeIsoTs(field(payload, "ts", "timestamp")); + const ts = normalizedRawTs ?? nowIso(); + + const entryId = coerceString(field(payload, "entryId", "entry_id")); + const toolCallId = coerceString(field(payload, "toolCallId", "tool_call_id")); + const cwdRaw = field(payload, "cwd"); + const cwd = typeof cwdRaw === "string" ? cwdRaw : null; + + const consumed = new Set([ + "piEvent", + "pi_event", + "event", + "type", + "sessionId", + "session_id", + "entryId", + "entry_id", + "toolCallId", + "tool_call_id", + "cwd", + "ts", + "timestamp", + "project", + ]); + + let eventType: string; + let eventPayload: Record; + let idPart: string | null = null; + + if (piEvent === "session_start") { + eventType = "session_start"; + eventPayload = {}; + idPart = entryId || "session_start"; + } else if (piEvent === "session_shutdown") { + const reason = field(payload, "reason"); + eventType = "session_end"; + eventPayload = { reason: reason ?? null }; + idPart = entryId || "session_end"; + consumed.add("reason"); + } else if (piEvent === "message_end") { + const role = coerceString(field(payload, "role")).toLowerCase(); + const text = coerceString(field(payload, "text", "content", "prompt")); + if (!text) return null; + if (!entryId) return null; + if (role === "user") { + eventType = "prompt"; + eventPayload = { text }; + } else if (role === "assistant") { + eventType = "assistant"; + eventPayload = { text }; + } else { + return null; + } + idPart = entryId; + consumed.add("role"); + consumed.add("text"); + consumed.add("content"); + consumed.add("prompt"); + } else if (piEvent === "turn_end") { + // Design D2: turn_end → assistant. Extension emits message_end for completed + // assistant turns; agent_end is intentionally not in the mappable set. + const text = coerceString(field(payload, "text", "content")); + if (!text) return null; + if (!entryId && !toolCallId) return null; + eventType = "assistant"; + eventPayload = { text }; + idPart = entryId || toolCallId; + consumed.add("role"); + consumed.add("text"); + consumed.add("content"); + } else if (piEvent === "tool_call") { + const toolName = coerceString(field(payload, "toolName", "tool_name", "name")); + if (!toolName) return null; + if (!toolCallId && !entryId) return null; + const toolInput = objectOrEmpty(field(payload, "toolInput", "tool_input", "args", "input")); + eventType = "tool_call"; + eventPayload = { tool_name: toolName, tool_input: toolInput }; + idPart = toolCallId || entryId; + consumed.add("toolName"); + consumed.add("tool_name"); + consumed.add("name"); + consumed.add("toolInput"); + consumed.add("tool_input"); + consumed.add("args"); + consumed.add("input"); + } else if (piEvent === "tool_result") { + const toolName = coerceString(field(payload, "toolName", "tool_name", "name")); + if (!toolName) return null; + if (!toolCallId && !entryId) return null; + const toolInput = objectOrEmpty(field(payload, "toolInput", "tool_input", "args", "input")); + const isError = coerceBool(field(payload, "isError", "is_error")); + const toolOutput = field(payload, "toolOutput", "tool_output", "output", "result") ?? null; + const error = field(payload, "error", "tool_error") ?? null; + eventType = "tool_result"; + if (isError) { + eventPayload = { + tool_name: toolName, + status: "error", + tool_input: toolInput, + tool_output: null, + tool_error: error ?? true, + error: error ?? true, + }; + } else { + eventPayload = { + tool_name: toolName, + status: "ok", + tool_input: toolInput, + tool_output: toolOutput, + tool_error: null, + }; + } + // tool_result ids prefer toolCallId so call/result pair on the same id root. + idPart = toolCallId ? `${toolCallId}:result` : `${entryId}:result`; + consumed.add("toolName"); + consumed.add("tool_name"); + consumed.add("name"); + consumed.add("toolInput"); + consumed.add("tool_input"); + consumed.add("args"); + consumed.add("input"); + consumed.add("toolOutput"); + consumed.add("tool_output"); + consumed.add("output"); + consumed.add("result"); + consumed.add("isError"); + consumed.add("is_error"); + consumed.add("error"); + consumed.add("tool_error"); + } else { + return null; + } + + if (!idPart) return null; + + const meta: Record = { + pi_event: piEvent, + ordering_confidence: "low", + }; + if (entryId) meta.entry_id = entryId; + if (toolCallId) meta.tool_call_id = toolCallId; + if (normalizedRawTs === null) meta.ts_normalized = "generated"; + + const unknown: Record = {}; + for (const [key, value] of Object.entries(payload)) { + if (!consumed.has(key)) unknown[key] = value; + } + if (Object.keys(unknown).length > 0) meta.pi_fields = unknown; + + return { + schema_version: "1.0", + source: "pi", + session_id: sessionId, + event_id: buildPiEventId(sessionId, idPart), + event_type: eventType, + ts, + ordering_confidence: "low", + cwd, + payload: eventPayload, + meta, + }; +} + +// --------------------------------------------------------------------------- +// Flush signal (session_before_compact — observe only) +// --------------------------------------------------------------------------- + +export interface PiFlushSignal { + kind: "flush"; + reason: "session_before_compact"; + source: "pi"; + session_id: string; + ts: string; + cwd: string | null; + project: string | null; +} + +/** + * Build a flush signal for pi compaction boundaries. + * Returns null unless the payload is a session_before_compact event with a + * session id. Never produces a compaction object for pi to apply. + */ +export function buildPiFlushSignalFromEvent( + payload: Record, +): PiFlushSignal | null { + const piEvent = coercePiEventName(payload); + if (!PI_FLUSH_ONLY_EVENTS.has(piEvent)) return null; + + const sessionId = coerceSessionId(payload); + if (!sessionId) return null; + + const normalizedRawTs = normalizeIsoTs(field(payload, "ts", "timestamp")); + const ts = normalizedRawTs ?? nowIso(); + const cwdRaw = field(payload, "cwd"); + const cwd = typeof cwdRaw === "string" ? cwdRaw : null; + const project = + resolveHookProject(cwd, field(payload, "project")) ?? + normalizeProjectLabel(field(payload, "project")); + + return { + kind: "flush", + reason: "session_before_compact", + source: "pi", + session_id: sessionId, + ts, + cwd, + project, + }; +} + +// --------------------------------------------------------------------------- +// buildRawEventEnvelopeFromPiEvent +// --------------------------------------------------------------------------- + +export interface PiHookRawEventEnvelope { + session_stream_id: string; + session_id: string; + opencode_session_id: string; + source: "pi"; + event_id: string; + event_type: "pi.hook"; + payload: Record; + ts_wall_ms: number; + cwd: string | null; + project: string | null; + started_at: string | null; +} + +/** + * Build a raw event envelope from a pi extension event payload. + * Returns null if the payload is unsupported, flush-only, or missing fields. + * Source is always the literal "pi" — never falls through to a default. + */ +export function buildRawEventEnvelopeFromPiEvent( + piPayload: Record, +): PiHookRawEventEnvelope | null { + const adapterEvent = mapPiEventPayload(piPayload); + if (adapterEvent === null) return null; + + const sessionId = adapterEvent.session_id.trim(); + if (!sessionId) return null; + const ts = adapterEvent.ts.trim(); + if (!ts) return null; + + const cwdRaw = field(piPayload, "cwd"); + const cwd = typeof cwdRaw === "string" ? cwdRaw : null; + const project = + resolveHookProject(cwd, field(piPayload, "project")) ?? + normalizeProjectLabel(field(piPayload, "project")); + const piEvent = coercePiEventName(piPayload); + + return { + session_stream_id: sessionId, + session_id: sessionId, + opencode_session_id: sessionId, + source: "pi", + event_id: adapterEvent.event_id, + event_type: "pi.hook", + payload: { + type: "pi.hook", + timestamp: ts, + _adapter: adapterEvent, + }, + ts_wall_ms: isoToWallMs(ts), + cwd, + project, + started_at: piEvent === "session_start" ? ts : null, + }; +} + +// --------------------------------------------------------------------------- +// buildIngestPayloadFromPiEvent +// --------------------------------------------------------------------------- + +/** + * Build an ingest pipeline payload from a pi extension event. + * Used by the direct-ingest path. Source is always the literal "pi". + * Returns null if the payload is unsupported or flush-only. + */ +export function buildIngestPayloadFromPiEvent( + piPayload: Record, +): Record | null { + const adapterEvent = mapPiEventPayload(piPayload); + if (adapterEvent === null) return null; + + const sessionId = adapterEvent.session_id; + return { + cwd: field(piPayload, "cwd") ?? null, + events: [ + { + type: "pi.hook", + timestamp: adapterEvent.ts, + _adapter: adapterEvent, + }, + ], + session_context: { + source: "pi", + stream_id: sessionId, + session_stream_id: sessionId, + session_id: sessionId, + opencode_session_id: sessionId, + }, + }; +} diff --git a/packages/core/src/store.test.ts b/packages/core/src/store.test.ts index 5b39a7383..f283c6f71 100644 --- a/packages/core/src/store.test.ts +++ b/packages/core/src/store.test.ts @@ -2336,6 +2336,12 @@ describe("buildFilterClauses", () => { expect(result.params).toEqual(["discovery"]); }); + it("ignores a non-string kind instead of binding it", () => { + const result = buildFilterClauses({ kind: 5 as unknown as string }); + expect(result.clauses).toEqual([]); + expect(result.params).toEqual([]); + }); + it("builds include_visibility filter", () => { const result = buildFilterClauses({ include_visibility: ["private", "shared"] }); expect(result.clauses).toHaveLength(1); diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index ebbdc9b3b..0fa58df86 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -31,6 +31,7 @@ import { } from "./db.js"; import { buildFilterClausesWithContext, type OwnershipFilterContext } from "./filters.js"; import { buildMemoryDedupKey, normalizeMemoryDedupTitle } from "./memory-dedup.js"; +import { validateMemoryKind } from "./memory-kinds.js"; import { readCodememConfigFile } from "./observer-config.js"; import type { PackArtifacts } from "./pack.js"; import { @@ -75,19 +76,6 @@ import type { } from "./types.js"; import { storeVectors } from "./vectors.js"; -// Memory kind validation (mirrors codemem/memory_kinds.py) - -const ALLOWED_MEMORY_KINDS = new Set([ - "discovery", - "change", - "feature", - "bugfix", - "refactor", - "decision", - "exploration", - "session_summary", -]); - // Locally reviewed flows that can establish same-person ownership. Coordinator // enrollment remains discovery evidence and must never grant write authority. const SAME_PERSON_BINDING_PROVENANCE = new Set([ @@ -97,17 +85,6 @@ const SAME_PERSON_BINDING_PROVENANCE = new Set([ "review_resolution", ]); -/** Normalize and validate a memory kind. Throws on invalid kinds. */ -function validateMemoryKind(kind: string): string { - const normalized = kind.trim().toLowerCase(); - if (!ALLOWED_MEMORY_KINDS.has(normalized)) { - throw new Error( - `Invalid memory kind "${kind}". Allowed: ${[...ALLOWED_MEMORY_KINDS].join(", ")}`, - ); - } - return normalized; -} - // Helpers /** ISO 8601 timestamp in UTC. */ diff --git a/packages/mcp-server/src/memory-kinds.ts b/packages/mcp-server/src/memory-kinds.ts index bdef8082a..c91337b5b 100644 --- a/packages/mcp-server/src/memory-kinds.ts +++ b/packages/mcp-server/src/memory-kinds.ts @@ -1,9 +1,5 @@ -export const MEMORY_KINDS: Record = { - discovery: "Something learned about the codebase, architecture, or tools", - change: "A code change that was made", - feature: "A new feature that was implemented", - bugfix: "A bug that was found and fixed", - refactor: "Code that was refactored or restructured", - decision: "A design or architecture decision", - exploration: "An experiment or investigation (may not have shipped)", -}; +/** + * MCP memory-kind catalog — the seven kinds the remember tools accept. + * Re-exported from the shared core catalog so MCP and store never drift. + */ +export { MEMORY_KIND_DESCRIPTIONS as MEMORY_KINDS } from "@codemem/core"; diff --git a/packages/mcp-server/src/schemas.test.ts b/packages/mcp-server/src/schemas.test.ts new file mode 100644 index 000000000..41d779ee6 --- /dev/null +++ b/packages/mcp-server/src/schemas.test.ts @@ -0,0 +1,16 @@ +import { REMEMBER_MEMORY_KINDS } from "@codemem/core"; +import { describe, expect, it } from "vitest"; +import { memoryKindSchema } from "./schemas.js"; + +describe("memoryKindSchema", () => { + it("accepts every remember kind from the core catalog", () => { + for (const kind of REMEMBER_MEMORY_KINDS) { + expect(memoryKindSchema.safeParse(kind).success).toBe(true); + } + }); + + it("rejects session_summary and unknown kinds", () => { + expect(memoryKindSchema.safeParse("session_summary").success).toBe(false); + expect(memoryKindSchema.safeParse("not-a-kind").success).toBe(false); + }); +}); diff --git a/packages/mcp-server/src/schemas.ts b/packages/mcp-server/src/schemas.ts index 52c1b3892..d32d0b3f2 100644 --- a/packages/mcp-server/src/schemas.ts +++ b/packages/mcp-server/src/schemas.ts @@ -1,3 +1,4 @@ +import { REMEMBER_MEMORY_KINDS } from "@codemem/core"; import { z } from "zod"; const scopeFilterSchema = { @@ -29,14 +30,6 @@ export const filterSchema = { widen_shared_min_personal_score: z.number().optional(), }; -export const memoryKindSchema = z.enum([ - "discovery", - "change", - "feature", - "bugfix", - "refactor", - "decision", - "exploration", -]); +export const memoryKindSchema = z.enum(REMEMBER_MEMORY_KINDS); export const filterNames = Object.keys(filterSchema).toSorted(); diff --git a/packages/viewer-server/src/helpers.test.ts b/packages/viewer-server/src/helpers.test.ts index 6f8068c45..11abac315 100644 --- a/packages/viewer-server/src/helpers.test.ts +++ b/packages/viewer-server/src/helpers.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { queryInt } from "./helpers.js"; +import { parseJsonObjectBody, queryInt } from "./helpers.js"; describe("queryInt", () => { it("parses full integer strings", () => { @@ -14,3 +14,77 @@ describe("queryInt", () => { expect(queryInt("", 25)).toBe(25); }); }); + +describe("parseJsonObjectBody", () => { + function makeContext(body: BodyInit | null, headers: Record = {}) { + const request = new Request("http://viewer.test/api/raw-events", { + method: "POST", + body, + headers, + // Required by undici for stream bodies, and harmless for null bodies. + duplex: "half", + }); + return { + req: { + header: (name: string) => request.headers.get(name) ?? undefined, + raw: request, + }, + json: (data: unknown, status = 200) => new Response(JSON.stringify(data), { status }), + }; + } + + it("rejects oversized bodies with 413 based on the declared content-length", async () => { + // A stream body so the browser-style content-length recompute + // cannot override the declared header. + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("{}")); + controller.close(); + }, + }); + const result = await parseJsonObjectBody( + makeContext(stream, { "content-length": "999999" }), + 1024, + ); + expect(result).toBeInstanceOf(Response); + const response = result as Response; + expect(response.status).toBe(413); + expect(await response.json()).toEqual({ error: "payload too large", max_bytes: 1024 }); + }); + + it("rejects oversized bodies with 413 while streaming when no content-length is declared", async () => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("x".repeat(800))); + controller.enqueue(new TextEncoder().encode("y".repeat(800))); + controller.close(); + }, + }); + const result = await parseJsonObjectBody(makeContext(stream), 1024); + expect(result).toBeInstanceOf(Response); + const response = result as Response; + expect(response.status).toBe(413); + expect(await response.json()).toEqual({ error: "payload too large", max_bytes: 1024 }); + }); + + it("parses a well-formed JSON object body", async () => { + const result = await parseJsonObjectBody(makeContext(JSON.stringify({ events: [] })), 1024); + expect(result).toEqual({ events: [] }); + }); + + it("rejects invalid JSON with 400", async () => { + const result = await parseJsonObjectBody(makeContext("{ nope"), 1024); + expect(result).toBeInstanceOf(Response); + const response = result as Response; + expect(response.status).toBe(400); + expect(await response.json()).toEqual({ error: "invalid json" }); + }); + + it("rejects non-object JSON payloads with 400", async () => { + const result = await parseJsonObjectBody(makeContext("[1,2,3]"), 1024); + expect(result).toBeInstanceOf(Response); + const response = result as Response; + expect(response.status).toBe(400); + expect(await response.json()).toEqual({ error: "payload must be an object" }); + }); +}); diff --git a/packages/viewer-server/src/helpers.ts b/packages/viewer-server/src/helpers.ts index 0484344e2..11a9f5302 100644 --- a/packages/viewer-server/src/helpers.ts +++ b/packages/viewer-server/src/helpers.ts @@ -37,3 +37,70 @@ export function queryBool(value: string | undefined): boolean { if (value == null) return false; return value === "1" || value === "true" || value === "yes"; } + +/** + * Parse and validate a JSON object body, enforcing size limits. + * Returns the parsed payload or a Hono Response on error. + */ +export async function parseJsonObjectBody( + c: { + req: { header: (name: string) => string | undefined; raw: Request }; + json: (data: unknown, status?: number) => Response; + }, + maxBytes: number, +): Promise | Response> { + const contentLength = Number.parseInt(c.req.header("content-length") ?? "0", 10); + if (Number.isNaN(contentLength) || contentLength < 0) { + return c.json({ error: "invalid content-length" }, 400); + } + if (contentLength > maxBytes) { + return c.json({ error: "payload too large", max_bytes: maxBytes }, 413); + } + let raw = ""; + try { + const body = c.req.raw.body; + if (body !== null) { + const reader = body.getReader(); + const chunks: Uint8Array[] = []; + let totalBytes = 0; + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + totalBytes += value.byteLength; + if (totalBytes > maxBytes) { + try { + await reader.cancel(); + } catch { + // The size rejection is authoritative even if cancellation fails. + } + return c.json({ error: "payload too large", max_bytes: maxBytes }, 413); + } + chunks.push(value); + } + } finally { + reader.releaseLock(); + } + + const bytes = new Uint8Array(totalBytes); + let offset = 0; + for (const chunk of chunks) { + bytes.set(chunk, offset); + offset += chunk.byteLength; + } + raw = new TextDecoder("utf-8", { fatal: true }).decode(bytes); + } + } catch { + return c.json({ error: "invalid json" }, 400); + } + let parsed: unknown; + try { + parsed = raw ? JSON.parse(raw) : {}; + } catch { + return c.json({ error: "invalid json" }, 400); + } + if (parsed == null || typeof parsed !== "object" || Array.isArray(parsed)) { + return c.json({ error: "payload must be an object" }, 400); + } + return parsed as Record; +} diff --git a/packages/viewer-server/src/routes/raw-events.ts b/packages/viewer-server/src/routes/raw-events.ts index d540e8504..b80016217 100644 --- a/packages/viewer-server/src/routes/raw-events.ts +++ b/packages/viewer-server/src/routes/raw-events.ts @@ -16,7 +16,7 @@ import { import { desc } from "drizzle-orm"; import { drizzle } from "drizzle-orm/better-sqlite3"; import { Hono } from "hono"; -import { queryInt } from "../helpers.js"; +import { parseJsonObjectBody, queryInt } from "../helpers.js"; import { validateViewerTarget } from "./target-validation.js"; type StoreFactory = () => MemoryStore; @@ -87,73 +87,6 @@ function codexTranscriptRoot(): string { return join(process.env.CODEX_HOME?.trim() || join(homedir(), ".codex"), "sessions"); } -/** - * Parse and validate a JSON object body, enforcing size limits. - * Returns the parsed payload or a Hono Response on error. - */ -async function parseJsonObjectBody( - c: { - req: { header: (name: string) => string | undefined; raw: Request }; - json: (data: unknown, status?: number) => Response; - }, - maxBytes: number, -): Promise | Response> { - const contentLength = Number.parseInt(c.req.header("content-length") ?? "0", 10); - if (Number.isNaN(contentLength) || contentLength < 0) { - return c.json({ error: "invalid content-length" }, 400); - } - if (contentLength > maxBytes) { - return c.json({ error: "payload too large", max_bytes: maxBytes }, 413); - } - let raw = ""; - try { - const body = c.req.raw.body; - if (body !== null) { - const reader = body.getReader(); - const chunks: Uint8Array[] = []; - let totalBytes = 0; - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - totalBytes += value.byteLength; - if (totalBytes > maxBytes) { - try { - await reader.cancel(); - } catch { - // The size rejection is authoritative even if cancellation fails. - } - return c.json({ error: "payload too large", max_bytes: maxBytes }, 413); - } - chunks.push(value); - } - } finally { - reader.releaseLock(); - } - - const bytes = new Uint8Array(totalBytes); - let offset = 0; - for (const chunk of chunks) { - bytes.set(chunk, offset); - offset += chunk.byteLength; - } - raw = new TextDecoder("utf-8", { fatal: true }).decode(bytes); - } - } catch { - return c.json({ error: "invalid json" }, 400); - } - let parsed: unknown; - try { - parsed = raw ? JSON.parse(raw) : {}; - } catch { - return c.json({ error: "invalid json" }, 400); - } - if (parsed == null || typeof parsed !== "object" || Array.isArray(parsed)) { - return c.json({ error: "payload must be an object" }, 400); - } - return parsed as Record; -} - /** Nudge the sweeper safely — never crashes the caller. */ function nudgeSweeper( sweeper: RawEventSweeper | null | undefined,