From 5e0f8afd98b5ad6837ce70d7f08b42a7aaac1b65 Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Thu, 22 Oct 2026 20:10:07 +0200 Subject: [PATCH 1/6] Deduplicate daemon worker hook invocations Co-Authored-By: GPT-6.1 Sol --- CHANGELOG.md | 4 + .../hooks/worker-invocation-dedup.test.ts | 336 ++++++++++++++++++ src/hooks/custom-hooks-loader.ts | 24 +- src/hooks/handler.ts | 41 ++- src/hooks/worker-server.ts | 76 +++- 5 files changed, 466 insertions(+), 15 deletions(-) create mode 100644 __tests__/hooks/worker-invocation-dedup.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index d4f9e68b9..38aeea3c8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## 1.0.12-beta.0 — 2026-10-22 +### Fixes + +- Deduplicate hook invocations in the daemon worker when the same tool call arrives from multiple installation scopes. Matching invocations share one policy evaluation, activity record and telemetry event, and receive the same reply; different runtimes and calls without stable invocation IDs continue to evaluate independently. + ### Dependencies - Combine the open Dependabot PRs (#921, #922, #923) into one: `certifi` 2026.7.22 → 2026.10.19 (GHSA-6q48-6c7w-9gw2) in `fp-cloud-cli` and the Python SDK's test lockfile, and `vite` 8.3.0 → 8.3.4 in the TypeScript SDK's dev lockfile. The Supply Chain gate scans every lockfile, so none of the three could pass on its own (#926) diff --git a/__tests__/hooks/worker-invocation-dedup.test.ts b/__tests__/hooks/worker-invocation-dedup.test.ts new file mode 100644 index 000000000..ab633c0a6 --- /dev/null +++ b/__tests__/hooks/worker-invocation-dedup.test.ts @@ -0,0 +1,336 @@ +// @vitest-environment node +import { afterEach, beforeEach, describe, expect, it, vi, type MockInstance } from "vitest"; +import { createConnection, type Server, type Socket } from "node:net"; +import { mkdtempSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import * as handler from "../../src/hooks/handler"; +import * as evaluator from "../../src/hooks/policy-evaluator"; +import * as loader from "../../src/hooks/custom-hooks-loader"; +import * as cloudPolicies from "../../src/hooks/cloud-managed-policies"; +import * as logger from "../../src/hooks/hook-logger"; +import { getAllHookActivityEntries, _resetForTest } from "../../src/hooks/hook-activity-store"; +import { trackHookEvent } from "../../src/hooks/hook-telemetry"; +import { startWorkerServer } from "../../src/hooks/worker-server"; + +const runtime = vi.hoisted(() => ({ version: "test-version-a" })); +vi.mock("../../package.json", async (importOriginal) => ({ + ...await importOriginal(), + get version() { return runtime.version; }, +})); +// Keep evaluation and activity/stat persistence real; replace network telemetry. +vi.mock("../../src/hooks/hook-telemetry", () => ({ + trackHookEvent: vi.fn(() => Promise.resolve()), + flushHookTelemetry: vi.fn(() => Promise.resolve()), +})); + +function deferred() { + let resolve!: () => void; + const promise = new Promise((res) => { resolve = res; }); + return { promise, resolve }; +} + +function encodeFrame(value: unknown): Buffer { + const body = Buffer.from(JSON.stringify(value)); + const header = Buffer.alloc(4); + header.writeUInt32BE(body.length); + return Buffer.concat([header, body]); +} + +type Reply = Record; + +function readFrame(socket: Socket): Promise { + return new Promise((resolve, reject) => { + let buf = Buffer.alloc(0); + socket.on("data", (chunk: Buffer) => { + buf = Buffer.concat([buf, chunk]); + if (buf.length < 4 || buf.length < 4 + buf.readUInt32BE(0)) return; + resolve(JSON.parse(buf.subarray(4, 4 + buf.readUInt32BE(0)).toString())); + }); + socket.on("error", reject); + }); +} + +describe("worker invocation dedup (real socket, evaluator and activity store)", () => { + let dir: string; + let project: string; + let socketPath: string; + let server: Server; + const sockets: Socket[] = []; + let evaluate: MockInstance; + const realEvaluate = evaluator.evaluatePolicies; + const realHandler = handler.evaluateHookEvent; + let releaseRegistry: (() => void) | undefined; + + beforeEach(async () => { + dir = mkdtempSync(join(tmpdir(), "fpai-worker-dedup-")); + project = join(dir, "project"); + mkdirSync(join(project, ".failproofai"), { recursive: true }); + vi.stubEnv("FAILPROOFAI_HOME", join(dir, "home")); + vi.stubEnv("FAILPROOFAI_LOG_LEVEL", "error"); + vi.stubEnv("FAILPROOFAI_PACKAGE_ROOT", "package-a"); + runtime.version = "test-version-a"; + _resetForTest(join(dir, "activity")); + writeFileSync(join(project, ".failproofai", "policies-config.json"), JSON.stringify({ enabledPolicies: ["block-sudo"] })); + evaluate = vi.spyOn(evaluator, "evaluatePolicies"); + vi.spyOn(handler, "evaluateHookEvent").mockImplementation((event, cli, stdin, opts) => { + releaseRegistry = opts?.releaseRegistry; + return realHandler(event, cli, stdin, opts); + }); + vi.mocked(trackHookEvent).mockClear(); + socketPath = join(dir, "worker.sock"); + server = startWorkerServer(socketPath); + await new Promise((resolve, reject) => { + server.once("error", reject); + if (server.listening) resolve(); + else server.once("listening", resolve); + }); + }); + + afterEach(async () => { + vi.useRealTimers(); + for (const socket of sockets.splice(0)) socket.destroy(); + await new Promise((resolve) => server.close(() => resolve())); + vi.restoreAllMocks(); + vi.unstubAllEnvs(); + _resetForTest(); + // Retain isolated artifacts for inspection; no writes reach the real home. + }); + + function request(payload: Record = {}, hookEvent = "PreToolUse", cli = "claude") { + return { + type: "hook", hookEvent, cli, cwd: project, + stdin: JSON.stringify({ + session_id: "dedup-session", tool_use_id: "tool-a", cwd: project, + tool_name: "Bash", tool_input: { command: "sudo whoami" }, ...payload, + }), + }; + } + + function send(req = request()): Promise { + const socket = createConnection({ path: socketPath }, () => socket.write(encodeFrame(req))); + sockets.push(socket); + return readFrame(socket).then((reply) => { socket.end(); return reply; }); + } + + function holdOwner(fail = false) { + const started = deferred(); + const finish = deferred(); + evaluate.mockImplementationOnce(async (...args) => { + const result = await realEvaluate(...args); + // Model the two-tier boundary: registry reads have finished. + releaseRegistry?.(); + started.resolve(); + await finish.promise; + if (fail) throw new Error("owner evaluation failed"); + return result; + }); + return { started: started.promise, finish: finish.resolve }; + } + + function observeNextLoad() { + const loaded = deferred(); + const realLoad = loader.loadAllCustomHooks; + vi.spyOn(loader, "loadAllCustomHooks").mockImplementationOnce(async (...args) => { + const result = await realLoad(...args); + loaded.resolve(); + return result; + }); + return loaded.promise; + } + + it("evaluates and records concurrent scope duplicates once and replays the exact deny reply", async () => { + const [owner, duplicate] = await Promise.all([send(), send()]); + expect(evaluate).toHaveBeenCalledTimes(1); + expect(owner.type).toBe("hookResult"); + expect(JSON.parse(owner.stdout as string).hookSpecificOutput.permissionDecision).toBe("deny"); + expect(duplicate).toEqual(owner); + expect(getAllHookActivityEntries()).toHaveLength(1); + expect(JSON.parse(readFileSync(join(dir, "activity", "stats.json"), "utf8"))).toMatchObject({ totalEvents: 1, denyCount: 1 }); + expect(vi.mocked(trackHookEvent).mock.calls.filter((call) => call[1] === "hook_policy_triggered")).toHaveLength(1); + }); + + it("replays a completed allow reply", async () => { + const req = request({ tool_input: { command: "echo allowed" } }); + const owner = await send(req); + const duplicate = await send(req); + expect(evaluate).toHaveBeenCalledTimes(1); + expect(owner).toMatchObject({ type: "hookResult", exitCode: 0, stdout: "", stderr: "", evaluation: { decision: "allow" } }); + expect(duplicate).toEqual(owner); + expect(getAllHookActivityEntries()).toHaveLength(1); + }); + + it("emits a custom-policy load diagnostic once for duplicate scope invocations", async () => { + const broken = join(project, "broken.mjs"); + writeFileSync(broken, "export const = ;"); + writeFileSync(join(project, ".failproofai", "policies-config.json"), JSON.stringify({ + enabledPolicies: ["block-sudo"], customPoliciesPaths: [broken], + })); + vi.spyOn(logger, "hookLogError").mockImplementation(() => {}); + const owner = await send(); + expect(await send()).toEqual(owner); + expect(vi.mocked(trackHookEvent).mock.calls.filter((call) => call[1] === "custom_hooks_load_error")).toHaveLength(1); + expect(vi.mocked(trackHookEvent).mock.calls.filter((call) => call[1] === "hook_policy_triggered")).toHaveLength(1); + expect(evaluate).toHaveBeenCalledTimes(1); + }); + + it("emits a cloud-policy load diagnostic once for duplicate scope invocations", async () => { + vi.spyOn(cloudPolicies, "readActiveCloudManagedPolicies").mockImplementation(() => { + throw new Error("invalid cloud policy manifest"); + }); + const owner = await send(); + expect(await send()).toEqual(owner); + expect(vi.mocked(trackHookEvent).mock.calls.filter((call) => call[1] === "cloud_managed_load_error")).toHaveLength(1); + expect(evaluate).toHaveBeenCalledTimes(1); + }); + + it("replays exit code and stderr exactly for an exit-2 deny", async () => { + const req = request({}, "PreToolUse", "factory"); + const owner = await send(req); + const duplicate = await send(req); + expect(owner).toMatchObject({ exitCode: 2, stdout: "" }); + expect(owner.stderr).toBeTruthy(); + expect(duplicate).toEqual(owner); + expect(evaluate).toHaveBeenCalledTimes(1); + }); + + it("shares a pending owner's reply after the owner releases the registry", async () => { + const gate = holdOwner(); + const owner = send(); + await gate.started; + const loaded = observeNextLoad(); + const duplicate = send(); + await loaded; + gate.finish(); + const [a, b] = await Promise.all([owner, duplicate]); + expect(b).toEqual(a); + expect(evaluate).toHaveBeenCalledTimes(1); + }); + + it("releases failed-owner waiters to normal evaluation and does not cache the error", async () => { + const gate = holdOwner(true); + const owner = send(); + await gate.started; + const loaded = observeNextLoad(); + const duplicate = send(); + await loaded; + gate.finish(); + expect(await owner).toMatchObject({ type: "error", message: "owner evaluation failed" }); + const retry = await duplicate; + expect(retry).toMatchObject({ type: "hookResult", evaluation: { decision: "deny" } }); + expect(await send()).toEqual(retry); + expect(evaluate).toHaveBeenCalledTimes(2); + expect(getAllHookActivityEntries()).toHaveLength(1); + }); + + it.each([ + ["tool invocation", { tool_use_id: "tool-b" }, "PreToolUse", "claude"], + ["session", { session_id: "other-session" }, "PreToolUse", "claude"], + ["event", {}, "PostToolUse", "claude"], + ["CLI", {}, "PreToolUse", "factory"], + ])("evaluates again when only the %s identity differs", async (_name, payload, event, cli) => { + const original = request(); + const owner = await send(original); + expect(await send(original)).toEqual(owner); + expect(evaluate).toHaveBeenCalledTimes(1); + await send(request(payload, event, cli)); + expect(evaluate).toHaveBeenCalledTimes(2); + expect(getAllHookActivityEntries()).toHaveLength(2); + }); + + it.each(["package root", "package version", "merged config", "custom hooks", "cwd", "fallback cwd"])( + "evaluates again when only runtime %s differs", async (component) => { + const req = request(component === "fallback cwd" ? { cwd: undefined } : {}); + const owner = await send(req); + expect(await send(req)).toEqual(owner); + expect(evaluate).toHaveBeenCalledTimes(1); + if (component === "package root") vi.stubEnv("FAILPROOFAI_PACKAGE_ROOT", "package-b"); + if (component === "package version") runtime.version = "test-version-b"; + if (component === "merged config") { + writeFileSync(join(project, ".failproofai", "policies-config.local.json"), JSON.stringify({ policyParams: { "block-sudo": { hint: "changed local hint" } } })); + } + if (component === "custom hooks") { + vi.spyOn(loader, "loadAllCustomHooks").mockResolvedValue({ + hooks: [{ name: "runtime-hook", match: {}, fn: async () => ({ decision: "allow" }) }], + conventionSources: [], + }); + } + let changed = req; + if (component === "cwd" || component === "fallback cwd") { + const nested = join(project, "nested"); + mkdirSync(nested); + changed = { ...req, cwd: nested }; + if (component === "cwd") changed.stdin = JSON.stringify({ ...JSON.parse(req.stdin), cwd: nested }); + } + await send(changed); + expect(evaluate).toHaveBeenCalledTimes(2); + }, + ); + + it.each([ + ["missing session", { session_id: undefined }], + ["missing invocation", { tool_use_id: undefined }], + ["empty session", { session_id: "" }], + ["empty invocation", { tool_use_id: "" }], + ["non-string invocation", { tool_use_id: 42 }], + ])("never suppresses or waits for %s", async (_name, payload) => { + const gate = holdOwner(); + const owner = send(request(payload)); + await gate.started; + try { + const independent = await send(request(payload)); + expect(independent.type).toBe("hookResult"); + expect(evaluate).toHaveBeenCalledTimes(2); + } finally { + gate.finish(); + await owner; + } + }); + + it("expires completed entries so the same invocation evaluates again", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const owner = await send(); + expect(await send()).toEqual(owner); + expect(evaluate).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(10 * 60_000 + 1); + await send(); + expect(evaluate).toHaveBeenCalledTimes(2); + expect(getAllHookActivityEntries()).toHaveLength(2); + }); + + it("bounds a pending duplicate's wait and evaluates normally when its owner never settles", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const gate = holdOwner(); + const owner = send(); + await gate.started; + const loaded = observeNextLoad(); + const duplicate = send(); + await loaded; + await vi.advanceTimersByTimeAsync(30_001); + try { + expect(await duplicate).toMatchObject({ type: "hookResult", evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(2); + } finally { + gate.finish(); + await owner; + } + }); + + it("uses the normal deny path when runtime identity computation throws", async () => { + const fn = async () => ({ decision: "allow" as const }); + fn.toString = () => { throw new Error("cannot fingerprint hook"); }; + vi.spyOn(loader, "loadAllCustomHooks").mockResolvedValue({ hooks: [{ name: "opaque", fn }], conventionSources: [] }); + const first = await send(); + const second = await send(); + expect(first).toMatchObject({ type: "hookResult", evaluation: { decision: "deny" } }); + expect(second).toMatchObject({ type: "hookResult", evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(2); + }); + + it("preserves errors on the normal path and recovers for subsequent requests", async () => { + evaluate.mockRejectedValueOnce(new Error("evaluation failed")); + expect(await send()).toMatchObject({ type: "error", message: "evaluation failed" }); + expect(await send()).toMatchObject({ type: "hookResult", evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/hooks/custom-hooks-loader.ts b/src/hooks/custom-hooks-loader.ts index 965f78f2e..10df13492 100644 --- a/src/hooks/custom-hooks-loader.ts +++ b/src/hooks/custom-hooks-loader.ts @@ -195,6 +195,8 @@ async function loadSingleFile( * file, whether or not Jev is configured — the declaration is dead either way. */ outsidePack?: boolean; + /** The worker defers diagnostics until invocation ownership is known. */ + deferTelemetry?: (emit: () => void) => void; }, ): Promise { const g = globalThis as Record; @@ -257,12 +259,16 @@ async function loadSingleFile( : /timed out loading module/i.test(msg) ? "load_timeout" : "runtime_error"; - void trackHookEvent(getInstanceId(), "custom_hooks_load_error", { - error_type: errorType, - is_convention: !!opts?.conventionScope, - convention_scope: opts?.conventionScope ?? null, - file_basename: basename(absPath), - }); + const emit = () => { + void trackHookEvent(getInstanceId(), "custom_hooks_load_error", { + error_type: errorType, + is_convention: !!opts?.conventionScope, + convention_scope: opts?.conventionScope ?? null, + file_basename: basename(absPath), + }); + }; + if (opts?.deferTelemetry) opts.deferTelemetry(emit); + else emit(); if (opts?.strict) throw new Error(`Failed to load custom hooks from ${absPath}: ${msg}`); hookLogError(`failed to load custom hooks from ${absPath}: ${msg}`); return { type: errorType, reason: msg }; @@ -412,6 +418,7 @@ export async function loadAllCustomHooks( customPoliciesEnabled?: boolean; cloudManagedPolicies?: CloudManagedPolicyArtifact[]; packs?: ResolvedPack[]; + deferTelemetry?: (emit: () => void) => void; }, ): Promise { clearCustomHooks(); @@ -654,6 +661,7 @@ export async function loadAllCustomHooks( verifyEntrySha: cloudManaged?.sha256 ?? pack?.sha256, outsidePack: !pack, + deferTelemetry: opts?.deferTelemetry, }); // Every id behind these bytes. One artifact, one import, one failure — // but as many fail-closed guards as there are packs depending on it. @@ -734,7 +742,7 @@ export async function loadAllCustomHooks( for (const file of projectFiles) { loadedPaths.add(file); const hooksBefore = getCustomHooks().length; - await loadSingleFile(file, { conventionScope: projectScope, outsidePack: true }); + await loadSingleFile(file, { conventionScope: projectScope, outsidePack: true, deferTelemetry: opts?.deferTelemetry }); const newHooks = getCustomHooks().slice(hooksBefore); for (const hook of newHooks) { (hook as CustomHook & { __policyId?: string }).__policyId = conventionPolicyId(projectScope, basename(file), hook.name); @@ -774,7 +782,7 @@ export async function loadAllCustomHooks( for (const file of userFiles) { loadedPaths.add(file); const hooksBefore = getCustomHooks().length; - await loadSingleFile(file, { conventionScope: "user", outsidePack: true }); + await loadSingleFile(file, { conventionScope: "user", outsidePack: true, deferTelemetry: opts?.deferTelemetry }); const newHooks = getCustomHooks().slice(hooksBefore); for (const hook of newHooks) { (hook as CustomHook & { __policyId?: string }).__policyId = conventionPolicyId("user", basename(file), hook.name); diff --git a/src/hooks/handler.ts b/src/hooks/handler.ts index 8eaef5499..4b42f2a7d 100644 --- a/src/hooks/handler.ts +++ b/src/hooks/handler.ts @@ -179,6 +179,15 @@ export interface EvaluateHookEventOptions { * leaves it unset. */ releaseRegistry?: () => void; + /** The worker may reuse a reply once the effective runtime is known, before + * evaluating policies or recording the decision. Loading stays on the same + * registry queue and is not repeated just to fingerprint custom hooks. */ + reuseOutcome?: (runtime: { + config: HooksConfig; + customHooks: CustomHook[]; + session: SessionMetadata; + payload: Record; + }) => Promise; } /** @@ -442,6 +451,11 @@ export async function evaluateHookEvent( opts?: EvaluateHookEventOptions, ): Promise { const startTime = performance.now(); + const loadTelemetry: Array<() => void> = []; + const deferTelemetry = opts?.reuseOutcome + ? (emit: () => void) => { loadTelemetry.push(emit); } + : undefined; + let replayed = false; // A home from another layout resolves to no global config, so every builtin // silently stops firing — the machine looks protected and is not. Say so on // every call rather than deleting anything (a hook is unattended) or denying @@ -624,10 +638,14 @@ export async function evaluateHookEvent( `[failproofai] cloud-managed policies could NOT be loaded and are not being enforced: ${msg}. ` + `Local policies are unaffected. Run \`failproofai config --status\` to check this machine's enrolment.`, ); - void trackHookEvent(getInstanceId(), "cloud_managed_load_error", { - event_type: eventType, - cli, - }); + const emit = () => { + void trackHookEvent(getInstanceId(), "cloud_managed_load_error", { + event_type: eventType, + cli, + }); + }; + if (deferTelemetry) deferTelemetry(emit); + else emit(); } // Recorded on every row once a machine is managed, whether or not a cloud // policy decided this event: "what was deployed here at the time" is a @@ -667,8 +685,20 @@ export async function evaluateHookEvent( customPoliciesEnabled: config.customPoliciesEnabled, ...(cloudManagedPolicies.length > 0 ? { cloudManagedPolicies } : {}), ...(installedPacks.length > 0 ? { packs: installedPacks } : {}), + ...(deferTelemetry ? { deferTelemetry } : {}), }); customHooksList = loadResult.hooks; + if (opts?.reuseOutcome) { + try { + const reused = await opts.reuseOutcome({ config, customHooks: customHooksList, session, payload: parsed }); + if (reused) { + replayed = true; + return reused; + } + } catch { + // Reuse is an optimization; its failure must still evaluate policy. + } + } const disabledCustomPolicies = new Set(config.disabledCustomPolicies ?? []); conventionHookNames = new Set(loadResult.conventionSources.flatMap((s) => s.hookNames)); @@ -1096,6 +1126,9 @@ export async function evaluateHookEvent( }, }; } finally { + if (!replayed) { + for (const emit of loadTelemetry) emit(); + } if (opts?.awaitTelemetryFlush ?? true) { // Await any un-awaited (`void trackHookEvent(...)`) events fired during // this invocation. The one-shot CLI process calls process.exit() the diff --git a/src/hooks/worker-server.ts b/src/hooks/worker-server.ts index 81c2e4cb0..485de1d97 100644 --- a/src/hooks/worker-server.ts +++ b/src/hooks/worker-server.ts @@ -24,11 +24,45 @@ */ import { createConnection, createServer, type Server, type Socket } from "node:net"; import { existsSync, unlinkSync } from "node:fs"; -import { evaluateHookEvent } from "./handler"; +import { createHash } from "node:crypto"; +import { version } from "../../package.json"; +import { evaluateHookEvent, type HookEventOutcome } from "./handler"; import type { IntegrationType } from "./types"; import { hookLogWarn } from "./hook-logger"; const MAX_FRAME_LEN = 16 * 1024 * 1024; +const INVOCATION_TTL_MS = 10 * 60_000; +const INVOCATION_WAIT_MS = 30_000; + +interface HookInvocation { + result: Promise; + complete: (outcome?: HookEventOutcome) => void; +} + +/** A failed or over-budget owner releases its waiters into normal evaluation. */ +function claimInvocation(invocations: Map, key: string): HookInvocation { + let resolve!: (outcome?: HookEventOutcome) => void; + let settled = false; + const entry: HookInvocation = { + result: new Promise((r) => { resolve = r; }), + complete: (outcome) => { + if (settled) return; + settled = true; + clearTimeout(timer); + if (outcome) { + timer = setTimeout(() => invocations.delete(key), INVOCATION_TTL_MS); + timer.unref?.(); + } else if (invocations.get(key) === entry) { + invocations.delete(key); + } + resolve(outcome); + }, + }; + let timer = setTimeout(() => entry.complete(), INVOCATION_WAIT_MS); + timer.unref?.(); + invocations.set(key, entry); + return entry; +} interface WorkerHookRequest { type: "hook"; @@ -195,7 +229,7 @@ function enqueue(task: (release: () => void) => Promise): void { processingChain = processingChain.then(() => new Promise((settle) => runQueuedTask(task, settle))).catch(() => {}); } -function handleConnection(socket: Socket, shutdown: () => void): void { +function handleConnection(socket: Socket, shutdown: () => void, invocations: Map): void { let recvBuf = Buffer.alloc(0); let declaredLen: number | null = null; /** @@ -272,6 +306,7 @@ function handleConnection(socket: Socket, shutdown: () => void): void { .catch(() => {}); enqueue(async (release) => { + let invocation: HookInvocation | undefined; try { const result = await evaluateHookEvent(request.hookEvent, request.cli, request.stdin, { awaitTelemetryFlush: false, @@ -279,7 +314,39 @@ function handleConnection(socket: Socket, shutdown: () => void): void { // "absent" as null. fallbackCwd: request.cwd ?? undefined, releaseRegistry: release, + reuseOutcome: async ({ config, customHooks, session, payload }) => { + const sessionId = payload.session_id; + const toolUseId = payload.tool_use_id; + if (typeof sessionId !== "string" || !sessionId || typeof toolUseId !== "string" || !toolUseId) return; + const customHookIdentity = customHooks.map((hook) => ({ + name: hook.name, + description: hook.description, + match: hook.match, + implementation: String(hook.fn), + conventionScope: (hook as typeof hook & { __conventionScope?: string }).__conventionScope, + })); + const runtimeIdentity = createHash("sha256").update(JSON.stringify({ + version, + packageRoot: process.env.FAILPROOFAI_PACKAGE_ROOT ?? process.argv[1] ?? "unknown", + cwd: session.cwd ?? process.cwd(), + config, + customHookIdentity, + })).digest("hex"); + const key = createHash("sha256").update(JSON.stringify([ + request.cli, request.hookEvent, sessionId, toolUseId, runtimeIdentity, + ])).digest("hex"); + const existing = invocations.get(key); + if (existing) { + // The owner can be finishing a released two-tier review. This + // request keeps the registry while it waits, so a failed owner + // can fall back without another request clearing its policies. + const outcome = await existing.result; + if (outcome) return outcome; + } + invocation = claimInvocation(invocations, key); + }, }); + invocation?.complete(result); deliver( encodeFrame({ type: "hookResult", @@ -293,6 +360,8 @@ function handleConnection(socket: Socket, shutdown: () => void): void { const msg = err instanceof Error ? err.message : String(err); hookLogWarn(`worker: evaluateHookEvent threw: ${msg}`); deliver(encodeFrame({ type: "error", message: msg })); + } finally { + invocation?.complete(); } }); } @@ -346,8 +415,9 @@ function requestExistingWorkerShutdown(socketPath: string): Promise<"stale" | "s } export function startWorkerServer(socketPath: string, shutdown?: () => void): Server { + const invocations = new Map(); const server = createServer((socket) => - handleConnection(socket, shutdown ?? (() => server.close())), + handleConnection(socket, shutdown ?? (() => server.close()), invocations), ); const listen = () => { From 7eb920c374bfe0bf45a8a3f57c904a60f1c58a36 Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Thu, 22 Oct 2026 20:19:08 +0200 Subject: [PATCH 2/6] Complete policy loader fixtures in dedup tests Co-Authored-By: GPT-6.1 Sol --- __tests__/hooks/worker-invocation-dedup.test.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/__tests__/hooks/worker-invocation-dedup.test.ts b/__tests__/hooks/worker-invocation-dedup.test.ts index ab633c0a6..5a8095cf5 100644 --- a/__tests__/hooks/worker-invocation-dedup.test.ts +++ b/__tests__/hooks/worker-invocation-dedup.test.ts @@ -253,6 +253,8 @@ describe("worker invocation dedup (real socket, evaluator and activity store)", vi.spyOn(loader, "loadAllCustomHooks").mockResolvedValue({ hooks: [{ name: "runtime-hook", match: {}, fn: async () => ({ decision: "allow" }) }], conventionSources: [], + packFailures: new Map(), + packAliases: new Map(), }); } let changed = req; @@ -319,7 +321,10 @@ describe("worker invocation dedup (real socket, evaluator and activity store)", it("uses the normal deny path when runtime identity computation throws", async () => { const fn = async () => ({ decision: "allow" as const }); fn.toString = () => { throw new Error("cannot fingerprint hook"); }; - vi.spyOn(loader, "loadAllCustomHooks").mockResolvedValue({ hooks: [{ name: "opaque", fn }], conventionSources: [] }); + vi.spyOn(loader, "loadAllCustomHooks").mockResolvedValue({ + hooks: [{ name: "opaque", fn }], conventionSources: [], + packFailures: new Map(), packAliases: new Map(), + }); const first = await send(); const second = await send(); expect(first).toMatchObject({ type: "hookResult", evaluation: { decision: "deny" } }); From ca11c4242e660b743c0e1a19b6580fc612650120 Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Thu, 22 Oct 2026 20:30:08 +0200 Subject: [PATCH 3/6] Preserve the worker evaluator import contract Co-Authored-By: GPT-6.1 Sol --- src/hooks/worker-server.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/hooks/worker-server.ts b/src/hooks/worker-server.ts index 485de1d97..87509710a 100644 --- a/src/hooks/worker-server.ts +++ b/src/hooks/worker-server.ts @@ -26,7 +26,8 @@ import { createConnection, createServer, type Server, type Socket } from "node:n import { existsSync, unlinkSync } from "node:fs"; import { createHash } from "node:crypto"; import { version } from "../../package.json"; -import { evaluateHookEvent, type HookEventOutcome } from "./handler"; +import { evaluateHookEvent } from "./handler"; +import type { HookEventOutcome } from "./handler"; import type { IntegrationType } from "./types"; import { hookLogWarn } from "./hook-logger"; From 3394019dcee4dce7f4e8d327382f259f5a3bbfdb Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Thu, 22 Oct 2026 21:32:45 +0200 Subject: [PATCH 4/6] Keep invocation reuse aligned with runtime and queue ownership Co-Authored-By: GPT-6.1 Sol --- CHANGELOG.md | 2 +- .../hooks/worker-invocation-dedup.test.ts | 204 +++++++++++++++++- src/hooks/custom-hooks-loader.ts | 31 ++- src/hooks/handler.ts | 35 ++- src/hooks/worker-server.ts | 138 ++++++------ 5 files changed, 325 insertions(+), 85 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 38aeea3c8..8bf0fb088 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ ### Fixes -- Deduplicate hook invocations in the daemon worker when the same tool call arrives from multiple installation scopes. Matching invocations share one policy evaluation, activity record and telemetry event, and receive the same reply; different runtimes and calls without stable invocation IDs continue to evaluate independently. +- Deduplicate hook invocations in the daemon worker when the same tool call arrives from multiple installation scopes. Matching invocations share one policy evaluation, activity record and telemetry event, and receive the same reply; changed source graphs or enforcement metadata invalidate reuse. Waiting duplicates release the registry queue and reload before retrying failed owners, so unrelated calls keep running. Calls without stable invocation IDs continue to evaluate independently. ### Dependencies diff --git a/__tests__/hooks/worker-invocation-dedup.test.ts b/__tests__/hooks/worker-invocation-dedup.test.ts index 5a8095cf5..0cce5dd1e 100644 --- a/__tests__/hooks/worker-invocation-dedup.test.ts +++ b/__tests__/hooks/worker-invocation-dedup.test.ts @@ -1,6 +1,7 @@ // @vitest-environment node import { afterEach, beforeEach, describe, expect, it, vi, type MockInstance } from "vitest"; import { createConnection, type Server, type Socket } from "node:net"; +import { createHash } from "node:crypto"; import { mkdtempSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -9,6 +10,8 @@ import * as evaluator from "../../src/hooks/policy-evaluator"; import * as loader from "../../src/hooks/custom-hooks-loader"; import * as cloudPolicies from "../../src/hooks/cloud-managed-policies"; import * as logger from "../../src/hooks/hook-logger"; +import * as packs from "../../src/hooks/pack-manifest"; +import type { ResolvedPack } from "../../src/hooks/pack-manifest"; import { getAllHookActivityEntries, _resetForTest } from "../../src/hooks/hook-activity-store"; import { trackHookEvent } from "../../src/hooks/hook-telemetry"; import { startWorkerServer } from "../../src/hooks/worker-server"; @@ -139,6 +142,189 @@ describe("worker invocation dedup (real socket, evaluator and activity store)", return loaded.promise; } + function policyFile(source: string) { + const path = join(project, "runtime-policy.mjs"); + writeFileSync(path, source); + writeFileSync(join(project, ".failproofai", "policies-config.json"), JSON.stringify({ + enabledPolicies: [], customPoliciesPaths: [path], + })); + return { path, sha256: createHash("sha256").update(source).digest("hex") }; + } + + it("reloads a changed captured dependency instead of replaying the old allow", async () => { + const dependency = join(project, "decision.mjs"); + writeFileSync(dependency, 'export const decision = "allow";'); + policyFile(`import { customPolicies } from "failproofai"; +import { decision } from "./decision.mjs"; +customPolicies.add({ name: "runtime-guard", match: { events: ["PreToolUse"] }, + fn: async () => ({ decision, reason: "captured dependency" }) });`); + const req = request({ tool_input: { command: "echo guarded" } }); + expect(await send(req)).toMatchObject({ evaluation: { decision: "allow" } }); + writeFileSync(dependency, 'export const decision = "deny";'); + expect(await send(req)).toMatchObject({ evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(2); + }); + + it("still reuses an unchanged source graph's captured closure", async () => { + policyFile(`import { customPolicies } from "failproofai"; +let calls = 0; +customPolicies.add({ name: "runtime-guard", match: { events: ["PreToolUse"] }, + fn: async () => ({ decision: "deny", reason: "call-" + (++calls) }) });`); + const req = request({ tool_input: { command: "echo guarded" } }); + const owner = await send(req); + expect(owner).toMatchObject({ evaluation: { decision: "deny", reason: "call-1" } }); + expect(await send(req)).toEqual(owner); + expect(evaluate).toHaveBeenCalledTimes(1); + expect(getAllHookActivityEntries()).toHaveLength(1); + }); + + it("invalidates a changed source graph even when it reuses a global function", async () => { + const source = `import { customPolicies } from "failproofai"; +globalThis.__fpaiDedupDecision = "allow"; +globalThis.__fpaiDedupGuard ??= async () => ({ decision: globalThis.__fpaiDedupDecision, reason: "global guard" }); +customPolicies.add({ name: "runtime-guard", match: { events: ["PreToolUse"] }, fn: globalThis.__fpaiDedupGuard });`; + const artifact = policyFile(source); + const req = request({ tool_input: { command: "echo guarded" } }); + try { + expect(await send(req)).toMatchObject({ evaluation: { decision: "allow" } }); + writeFileSync(artifact.path, source.replace('__fpaiDedupDecision = "allow"', '__fpaiDedupDecision = "deny"')); + expect(await send(req)).toMatchObject({ evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(2); + } finally { + delete (globalThis as Record).__fpaiDedupDecision; + delete (globalThis as Record).__fpaiDedupGuard; + } + }); + + it("evaluates normally when a hook has no verified source graph identity", async () => { + let decision: "allow" | "deny" = "allow"; + const hook = { name: "unverified", fn: async () => ({ decision, reason: "unverified runtime" }) }; + vi.spyOn(loader, "loadAllCustomHooks").mockResolvedValue({ + hooks: [hook], conventionSources: [], packFailures: new Map(), packAliases: new Map(), + }); + const req = request({ tool_input: { command: "echo guarded" } }); + expect(await send(req)).toMatchObject({ evaluation: { decision: "allow" } }); + decision = "deny"; + expect(await send(req)).toMatchObject({ evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(2); + }); + + it("includes source graphs that register no hooks but change a guard's state", async () => { + const statePath = join(project, "runtime-state.mjs"); + writeFileSync(statePath, 'globalThis.__fpaiDedupState = "allow";'); + const guard = policyFile(`import { customPolicies } from "failproofai"; +customPolicies.add({ name: "runtime-guard", match: { events: ["PreToolUse"] }, + fn: async () => ({ decision: globalThis.__fpaiDedupState, reason: "shared state" }) });`); + writeFileSync(join(project, ".failproofai", "policies-config.json"), JSON.stringify({ + enabledPolicies: [], customPoliciesPaths: [statePath, guard.path], + })); + const req = request({ tool_input: { command: "echo guarded" } }); + try { + expect(await send(req)).toMatchObject({ evaluation: { decision: "allow" } }); + writeFileSync(statePath, 'globalThis.__fpaiDedupState = "deny";'); + expect(await send(req)).toMatchObject({ evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(2); + } finally { + delete (globalThis as Record).__fpaiDedupState; + } + }); + + it.each(["cloud", "pack"] as const)("does not alias %s observe and enforce metadata", async (kind) => { + const artifact = policyFile(`import { customPolicies } from "failproofai"; +customPolicies.add({ name: "runtime-guard", match: { events: ["PreToolUse"] }, + fn: async () => ({ decision: "deny", reason: "managed denial" }) });`); + let enforce: () => void; + if (kind === "cloud") { + const metadata = { ...artifact, id: "runtime-guard", version: 1, deployment: 1, effect: "observe" as const }; + const read = vi.spyOn(cloudPolicies, "readActiveCloudManagedPolicies").mockReturnValue([metadata]); + enforce = () => { read.mockReturnValue([{ ...metadata, effect: "enforce" }]); }; + } else { + const metadata: ResolvedPack = { + ...artifact, id: "acme/dedup", version: "1.0.0", source: "github:acme/dedup@v1.0.0", + effect: "observe", policies: [], enabled: null, clis: null, + }; + vi.spyOn(packs, "hasInstalledRegexPacks").mockReturnValue(true); + const read = vi.spyOn(packs, "readInstalledPacks").mockReturnValue({ packs: [metadata], errors: [], warnings: [] }); + enforce = () => { read.mockReturnValue({ packs: [{ ...metadata, effect: "enforce" }], errors: [], warnings: [] }); }; + } + const req = request({ tool_input: { command: "echo guarded" } }); + expect(await send(req)).toMatchObject({ evaluation: { decision: "allow" } }); + enforce(); + expect(await send(req)).toMatchObject({ evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(2); + }); + + it.each(["selection", "default parameters"] as const)("includes pack %s in runtime equivalence", async (change) => { + const artifact = policyFile(`import { customPolicies } from "failproofai"; +customPolicies.add({ name: "runtime-guard", match: { events: ["PreToolUse"] }, + fn: async (ctx) => ({ decision: ctx.params?.blocked ? "deny" : "allow", reason: "pack params" }) });`); + const metadata: ResolvedPack = { + ...artifact, id: "acme/dedup", version: "1.0.0", source: "github:acme/dedup@v1.0.0", + effect: "enforce", enabled: change === "selection" ? [] : null, clis: null, + policies: [{ name: "runtime-guard", description: "runtime guard", defaultEnabled: true, + category: "Custom", match: { events: ["PreToolUse"] }, + params: { blocked: { type: "boolean", description: "block this call", default: change === "selection" } } }], + }; + vi.spyOn(packs, "hasInstalledRegexPacks").mockReturnValue(true); + const read = vi.spyOn(packs, "readInstalledPacks").mockReturnValue({ packs: [metadata], errors: [], warnings: [] }); + const req = request({ tool_input: { command: "echo guarded" } }); + expect(await send(req)).toMatchObject({ evaluation: { decision: "allow" } }); + const changed: ResolvedPack = change === "selection" + ? { ...metadata, enabled: ["runtime-guard"] } + : { ...metadata, policies: metadata.policies.map((p) => ({ ...p, + params: { blocked: { type: "boolean", description: "block this call", default: true } } })) }; + read.mockReturnValue({ packs: [changed], errors: [], warnings: [] }); + expect(await send(req)).toMatchObject({ evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(2); + }); + + it("does not replay an allow when a newly failed pack must fail closed", async () => { + const read = vi.spyOn(packs, "readInstalledPacks").mockReturnValue({ packs: [], errors: [], warnings: [] }); + const req = request({ tool_input: { command: "echo guarded" } }); + expect(await send(req)).toMatchObject({ evaluation: { decision: "allow" } }); + read.mockReturnValue({ packs: [], errors: [{ id: "acme/dedup", reason: "artifact unavailable", effect: "enforce" }], warnings: [] }); + expect(await send(req)).toMatchObject({ evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(2); + }); + + it.each([[false, false], [true, false], [true, true]])("lets a third no-ID request finish before a pending duplicate (owner fails: %s, retry fails: %s)", async (fail, retryFails) => { + const other = join(dir, "other-project"); + mkdirSync(join(other, ".failproofai"), { recursive: true }); + writeFileSync(join(other, ".failproofai", "policies-config.json"), JSON.stringify({ enabledPolicies: [] })); + const gate = holdOwner(fail); + if (retryFails) { + evaluate.mockImplementationOnce(realEvaluate).mockRejectedValueOnce(new Error("retry evaluation failed")); + } + const owner = send(); + await gate.started; + const loaded = observeNextLoad(); + const duplicate = send(); + await loaded; + const independent = send(request({ tool_use_id: undefined, cwd: other })); + let timer: ReturnType | undefined; + const first = await Promise.race([ + independent.then((reply) => ({ kind: "independent", reply })), + new Promise<{ kind: string; reply?: Reply }>((resolve) => { + timer = setTimeout(() => resolve({ kind: "blocked" }), 1_000); + }), + ]); + clearTimeout(timer); + gate.finish(); + const [a, b, c] = await Promise.all([owner, duplicate, independent]); + expect(first.kind).toBe("independent"); + expect(c).toMatchObject({ evaluation: { decision: "allow" } }); + if (fail) { + expect(a).toMatchObject({ type: "error", message: "owner evaluation failed" }); + expect(b).toMatchObject(retryFails + ? { type: "error", message: "retry evaluation failed" } + : { evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(3); + } else { + expect(b).toEqual(a); + expect(evaluate).toHaveBeenCalledTimes(2); + } + }); + it("evaluates and records concurrent scope duplicates once and replays the exact deny reply", async () => { const [owner, duplicate] = await Promise.all([send(), send()]); expect(evaluate).toHaveBeenCalledTimes(1); @@ -275,6 +461,9 @@ describe("worker invocation dedup (real socket, evaluator and activity store)", ["empty session", { session_id: "" }], ["empty invocation", { tool_use_id: "" }], ["non-string invocation", { tool_use_id: 42 }], + ["numeric session", { session_id: 42 }], + ["null session", { session_id: null }], + ["object session", { session_id: {} }], ])("never suppresses or waits for %s", async (_name, payload) => { const gate = holdOwner(); const owner = send(request(payload)); @@ -319,11 +508,16 @@ describe("worker invocation dedup (real socket, evaluator and activity store)", }); it("uses the normal deny path when runtime identity computation throws", async () => { - const fn = async () => ({ decision: "allow" as const }); - fn.toString = () => { throw new Error("cannot fingerprint hook"); }; - vi.spyOn(loader, "loadAllCustomHooks").mockResolvedValue({ - hooks: [{ name: "opaque", fn }], conventionSources: [], - packFailures: new Map(), packAliases: new Map(), + const artifact = policyFile(`import { customPolicies } from "failproofai"; +customPolicies.add({ name: "opaque", fn: async () => ({ decision: "allow" }) });`); + writeFileSync(join(project, ".failproofai", "policies-config.json"), JSON.stringify({ + enabledPolicies: ["block-sudo"], customPoliciesPaths: [artifact.path], + })); + const realLoad = loader.loadAllCustomHooks; + vi.spyOn(loader, "loadAllCustomHooks").mockImplementation(async (...args) => { + const result = await realLoad(...args); + result.hooks[0].fn.toString = () => { throw new Error("cannot fingerprint hook"); }; + return result; }); const first = await send(); const second = await send(); diff --git a/src/hooks/custom-hooks-loader.ts b/src/hooks/custom-hooks-loader.ts index 10df13492..6c3e75b12 100644 --- a/src/hooks/custom-hooks-loader.ts +++ b/src/hooks/custom-hooks-loader.ts @@ -56,6 +56,13 @@ interface CachedPolicyModule { * functions prevents an unbounded ESM module-cache leak in the resident worker. * A source-tree change replaces this entry and imports one new module. */ const policyModuleCache = new Map(); +const loadSourceFingerprints = new WeakMap>(); + +/** Every attempted source graph, including modules that register no hooks. */ +export function getCustomHooksSourceFingerprints(result: LoadAllResult): ReadonlyMap | undefined { + return loadSourceFingerprints.get(result); +} + /** * Cap, for the reason `gitBranchCache` states and this cache did not carry * over: a warm worker touching many projects over its lifetime must not grow @@ -197,6 +204,7 @@ async function loadSingleFile( outsidePack?: boolean; /** The worker defers diagnostics until invocation ownership is known. */ deferTelemetry?: (emit: () => void) => void; + recordSourceFingerprint?: (fingerprint: string | undefined) => void; }, ): Promise { const g = globalThis as Record; @@ -204,6 +212,7 @@ async function loadSingleFile( let tmpFiles: string[] = []; try { + opts?.recordSourceFingerprint?.(undefined); const distIndex = await findDistIndex(); const distUrl = distIndex ? pathToFileURL(distIndex).href : null; @@ -223,6 +232,7 @@ async function loadSingleFile( ); const fingerprint = await fingerprintTemporaryTree(tmpFiles, tmpSuffix); + opts?.recordSourceFingerprint?.(fingerprint); const cached = policyModuleCache.get(absPath); if (cached?.fingerprint === fingerprint) { for (const hook of cached.hooks) customPolicies.add({ ...hook }); @@ -234,6 +244,7 @@ async function loadSingleFile( const hooksBefore = getCustomHooks().length; const semanticBefore = getSemanticRegistrations().length; await importWithDeadline(fileUrl); + const loadedHooks = getCustomHooks().slice(hooksBefore); const ignored = getSemanticRegistrations().slice(semanticBefore).map((d) => d?.name); if (opts?.outsidePack && ignored.length > 0 && !warnedSemanticOutsidePack.has(absPath)) { warnedSemanticOutsidePack.add(absPath); @@ -245,9 +256,7 @@ async function loadSingleFile( if (policyModuleCache.size >= POLICY_MODULE_CACHE_MAX_ENTRIES) policyModuleCache.clear(); policyModuleCache.set(absPath, { fingerprint, - hooks: getCustomHooks() - .slice(hooksBefore) - .map((hook) => ({ ...hook })), + hooks: loadedHooks.map((hook) => ({ ...hook })), }); return null; } catch (err) { @@ -425,6 +434,7 @@ export async function loadAllCustomHooks( const conventionSources: ConventionSource[] = []; const packFailures = new Map(); + const sourceFingerprints = new Map(); const projectRoot = findProjectConfigDir(opts?.sessionCwd ?? process.cwd()); @@ -662,6 +672,7 @@ export async function loadAllCustomHooks( cloudManaged?.sha256 ?? pack?.sha256, outsidePack: !pack, deferTelemetry: opts?.deferTelemetry, + recordSourceFingerprint: (fingerprint) => { sourceFingerprints.set(absPath, fingerprint); }, }); // Every id behind these bytes. One artifact, one import, one failure — // but as many fail-closed guards as there are packs depending on it. @@ -742,7 +753,10 @@ export async function loadAllCustomHooks( for (const file of projectFiles) { loadedPaths.add(file); const hooksBefore = getCustomHooks().length; - await loadSingleFile(file, { conventionScope: projectScope, outsidePack: true, deferTelemetry: opts?.deferTelemetry }); + await loadSingleFile(file, { + conventionScope: projectScope, outsidePack: true, deferTelemetry: opts?.deferTelemetry, + recordSourceFingerprint: (fingerprint) => { sourceFingerprints.set(file, fingerprint); }, + }); const newHooks = getCustomHooks().slice(hooksBefore); for (const hook of newHooks) { (hook as CustomHook & { __policyId?: string }).__policyId = conventionPolicyId(projectScope, basename(file), hook.name); @@ -782,7 +796,10 @@ export async function loadAllCustomHooks( for (const file of userFiles) { loadedPaths.add(file); const hooksBefore = getCustomHooks().length; - await loadSingleFile(file, { conventionScope: "user", outsidePack: true, deferTelemetry: opts?.deferTelemetry }); + await loadSingleFile(file, { + conventionScope: "user", outsidePack: true, deferTelemetry: opts?.deferTelemetry, + recordSourceFingerprint: (fingerprint) => { sourceFingerprints.set(file, fingerprint); }, + }); const newHooks = getCustomHooks().slice(hooksBefore); for (const hook of newHooks) { (hook as CustomHook & { __policyId?: string }).__policyId = conventionPolicyId("user", basename(file), hook.name); @@ -838,5 +855,7 @@ export async function loadAllCustomHooks( const winner = packByPath.get(path); if (winner && ids.length > 1) packAliases.set(winner.id, ids); } - return { hooks: allHooks, conventionSources, packFailures, packAliases }; + const result = { hooks: allHooks, conventionSources, packFailures, packAliases }; + loadSourceFingerprints.set(result, sourceFingerprints); + return result; } diff --git a/src/hooks/handler.ts b/src/hooks/handler.ts index 4b42f2a7d..a578b0af0 100644 --- a/src/hooks/handler.ts +++ b/src/hooks/handler.ts @@ -24,7 +24,7 @@ import { OPENCLAW_EVENT_MAP, ANTIGRAVITY_EVENT_MAP, } from "./types"; -import { existsSync } from "node:fs"; +import { existsSync, readFileSync } from "node:fs"; import { canonicalizeToolName, canonicalizeToolInput } from "./tool-name-canonicalize"; import { normalizeCliPayload } from "./normalize-cli-payload"; import type { PolicyFunction, PolicyResult, HooksConfig } from "./policy-types"; @@ -37,7 +37,7 @@ import type { TwoTierReview } from "./semantic/combine"; import type { JevActivityFields } from "./semantic/combine"; import type { JevConfig } from "./semantic/jev-config"; import { clearPolicies, registerPolicy, getPoliciesForEvent } from "./policy-registry"; -import { loadAllCustomHooks } from "./custom-hooks-loader"; +import { loadAllCustomHooks, getCustomHooksSourceFingerprints } from "./custom-hooks-loader"; import { contestedReviewerNames, effectiveReviewerNames, jevChecksInstalled } from "./effective-reviewers"; import { authorityDeclarationFor, @@ -187,7 +187,8 @@ export interface EvaluateHookEventOptions { customHooks: CustomHook[]; session: SessionMetadata; payload: Record; - }) => Promise; + policyState: unknown; + }) => Promise } | undefined>; } /** @@ -630,9 +631,11 @@ export async function evaluateHookEvent( // than the status quo, which already failed open on five of the twelve — // just accidentally, and without saying so. let cloudManagedPolicies: CloudManagedPolicyArtifact[] = []; + let cloudManagedLoadFailed = false; try { cloudManagedPolicies = readActiveCloudManagedPolicies(); } catch (err) { + cloudManagedLoadFailed = true; const msg = err instanceof Error ? err.message : String(err); hookLogWarn( `[failproofai] cloud-managed policies could NOT be loaded and are not being enforced: ${msg}. ` + @@ -658,6 +661,7 @@ export async function evaluateHookEvent( // enforcing underneath. See the fail-open note in pack-manifest.ts — the // day builtins become a fetched pack this posture has to change with them. let installedPacks: ResolvedPack[] = []; + let packReadFailed = false; try { const packResult = readInstalledPacks(); installedPacks = packResult.packs; @@ -666,6 +670,7 @@ export async function evaluateHookEvent( hookLogWarn(`pack ${err.id ?? "(unnamed)"} not loaded: ${err.reason}`); } } catch (err) { + packReadFailed = true; hookLogWarn(`pack manifest unreadable: ${err instanceof Error ? err.message : String(err)}`); } @@ -689,15 +694,31 @@ export async function evaluateHookEvent( }); customHooksList = loadResult.hooks; if (opts?.reuseOutcome) { + let reused: Awaited>>; try { - const reused = await opts.reuseOutcome({ config, customHooks: customHooksList, session, payload: parsed }); - if (reused) { - replayed = true; - return reused; + const sourceFingerprints = getCustomHooksSourceFingerprints(loadResult); + // Unverified graphs cannot establish runtime equivalence safely. + if (sourceFingerprints && [...sourceFingerprints.values()].every((fingerprint) => fingerprint !== undefined)) { + reused = await opts.reuseOutcome({ + config, customHooks: customHooksList, session, payload: parsed, + policyState: { + session, activePause, packsInstalledHere, installedPacks, packErrors, packReadFailed, + cloudManagedPolicies, cloudManagedLoadFailed, + sourceFingerprints: [...sourceFingerprints], + packFailures: [...loadResult.packFailures], packAliases: [...loadResult.packAliases], + jevConfig: existsSync(jevConfigFile()) ? readFileSync(jevConfigFile(), "utf8") : null, + }, + }); } } catch { // Reuse is an optimization; its failure must still evaluate policy. } + if (reused) { + replayed = true; + // A released duplicate retries on the worker queue. Errors from that + // fresh evaluation must escape, never resume this discarded registry. + return "pending" in reused ? await reused.pending : reused; + } } const disabledCustomPolicies = new Set(config.disabledCustomPolicies ?? []); conventionHookNames = new Set(loadResult.conventionSources.flatMap((s) => s.hookNames)); diff --git a/src/hooks/worker-server.ts b/src/hooks/worker-server.ts index 87509710a..3eab096ad 100644 --- a/src/hooks/worker-server.ts +++ b/src/hooks/worker-server.ts @@ -17,10 +17,9 @@ * evaluations could have request B's `clearPolicies()` wipe request A's * in-flight registration. Serializing keeps the existing registry code * correct with zero changes; a worker_threads/process pool is a valid - * future enhancement if this becomes a real throughput bottleneck. The one - * exception is the network wait of a two-tier (Jev) review, which happens - * after the registry is no longer read and so does not hold the queue (see - * `enqueue`). + * future enhancement if this becomes a real throughput bottleneck. Two-tier + * (Jev) network waits and duplicates waiting for their owners happen outside + * registry access and do not hold the queue (see `enqueue`). */ import { createConnection, createServer, type Server, type Socket } from "node:net"; import { existsSync, unlinkSync } from "node:fs"; @@ -138,10 +137,11 @@ const TASK_DEADLINE_MS = 60_000; * for anyway, but recovered on the next call instead of never. * * A task may hand its place in the chain back early by calling `release`: - * the next request starts while the task finishes on its own. Only the - * two-tier (Jev) path does, at the point it stops reading the registry and - * starts waiting on Jev's network answer (`EvaluateHookEventOptions. - * releaseRegistry`). Without that, every hook on the machine — PostToolUse, + * the next request starts while the task finishes on its own. The two-tier + * (Jev) path releases when it stops reading the registry and starts waiting on + * Jev's network answer (`EvaluateHookEventOptions.releaseRegistry`). A waiting + * duplicate also releases; a retry reenters the queue and reloads its runtime. + * Without that, every hook on the machine — PostToolUse, * UserPromptSubmit, other agents' calls — would queue behind each gated call's * round trip, and a slow or unreachable provider would push the calls at the * back of the queue past the client's 30 s budget into fail-closed denies. @@ -230,6 +230,58 @@ function enqueue(task: (release: () => void) => Promise): void { processingChain = processingChain.then(() => new Promise((settle) => runQueuedTask(task, settle))).catch(() => {}); } +function evaluateRequest(request: WorkerHookRequest, invocations: Map): Promise { + return new Promise((resolve, reject) => { + enqueue(async (release) => { + let invocation: HookInvocation | undefined; + try { + const result = await evaluateHookEvent(request.hookEvent, request.cli, request.stdin, { + awaitTelemetryFlush: false, + // Normalised here so no consumer has to know the wire spells + // "absent" as null. + fallbackCwd: request.cwd ?? undefined, + releaseRegistry: release, + reuseOutcome: async ({ config, customHooks, session, payload, policyState }) => { + const sessionId = payload.session_id; + const toolUseId = payload.tool_use_id; + if (typeof sessionId !== "string" || !sessionId || typeof toolUseId !== "string" || !toolUseId) return; + const customHookIdentity = customHooks.map((hook) => ({ + ...hook, + implementation: String(hook.fn), + })); + const runtimeIdentity = createHash("sha256").update(JSON.stringify({ + version, + packageRoot: process.env.FAILPROOFAI_PACKAGE_ROOT ?? process.argv[1] ?? "unknown", + cwd: session.cwd ?? process.cwd(), + config, + customHookIdentity, + policyState, + })).digest("hex"); + const key = createHash("sha256").update(JSON.stringify([ + request.cli, request.hookEvent, sessionId, toolUseId, runtimeIdentity, + ])).digest("hex"); + const existing = invocations.get(key); + if (existing) { + // Waiting consumes no registry slot. A failed owner retries through + // this queue and reloads before evaluating against any registry. + release(); + return { pending: existing.result.then((outcome) => + outcome ?? evaluateRequest(request, invocations)) }; + } + invocation = claimInvocation(invocations, key); + }, + }); + invocation?.complete(result); + resolve(result); + } catch (err) { + reject(err); + } finally { + invocation?.complete(); + } + }); + }); +} + function handleConnection(socket: Socket, shutdown: () => void, invocations: Map): void { let recvBuf = Buffer.alloc(0); let declaredLen: number | null = null; @@ -306,64 +358,18 @@ function handleConnection(socket: Socket, shutdown: () => void, invocations: Map }) .catch(() => {}); - enqueue(async (release) => { - let invocation: HookInvocation | undefined; - try { - const result = await evaluateHookEvent(request.hookEvent, request.cli, request.stdin, { - awaitTelemetryFlush: false, - // Normalised here so no consumer has to know the wire spells - // "absent" as null. - fallbackCwd: request.cwd ?? undefined, - releaseRegistry: release, - reuseOutcome: async ({ config, customHooks, session, payload }) => { - const sessionId = payload.session_id; - const toolUseId = payload.tool_use_id; - if (typeof sessionId !== "string" || !sessionId || typeof toolUseId !== "string" || !toolUseId) return; - const customHookIdentity = customHooks.map((hook) => ({ - name: hook.name, - description: hook.description, - match: hook.match, - implementation: String(hook.fn), - conventionScope: (hook as typeof hook & { __conventionScope?: string }).__conventionScope, - })); - const runtimeIdentity = createHash("sha256").update(JSON.stringify({ - version, - packageRoot: process.env.FAILPROOFAI_PACKAGE_ROOT ?? process.argv[1] ?? "unknown", - cwd: session.cwd ?? process.cwd(), - config, - customHookIdentity, - })).digest("hex"); - const key = createHash("sha256").update(JSON.stringify([ - request.cli, request.hookEvent, sessionId, toolUseId, runtimeIdentity, - ])).digest("hex"); - const existing = invocations.get(key); - if (existing) { - // The owner can be finishing a released two-tier review. This - // request keeps the registry while it waits, so a failed owner - // can fall back without another request clearing its policies. - const outcome = await existing.result; - if (outcome) return outcome; - } - invocation = claimInvocation(invocations, key); - }, - }); - invocation?.complete(result); - deliver( - encodeFrame({ - type: "hookResult", - exitCode: result.exitCode, - stdout: result.stdout, - stderr: result.stderr, - evaluation: result.evaluation, - }), - ); - } catch (err) { - const msg = err instanceof Error ? err.message : String(err); - hookLogWarn(`worker: evaluateHookEvent threw: ${msg}`); - deliver(encodeFrame({ type: "error", message: msg })); - } finally { - invocation?.complete(); - } + void evaluateRequest(request, invocations).then((result) => { + deliver(encodeFrame({ + type: "hookResult", + exitCode: result.exitCode, + stdout: result.stdout, + stderr: result.stderr, + evaluation: result.evaluation, + })); + }, (err) => { + const msg = err instanceof Error ? err.message : String(err); + hookLogWarn(`worker: evaluateHookEvent threw: ${msg}`); + deliver(encodeFrame({ type: "error", message: msg })); }); } }); From 46cf5c89083c728d2263ee86c18de126358f1822 Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Fri, 23 Oct 2026 00:07:17 +0200 Subject: [PATCH 5/6] Reserve fallback evaluation time within the daemon response budget Co-Authored-By: GPT-6.1 Sol --- .../hooks/worker-invocation-dedup.test.ts | 21 +++++++++++++++++++ src/hooks/daemon-client.ts | 2 +- src/hooks/worker-server.ts | 5 ++++- 3 files changed, 26 insertions(+), 2 deletions(-) diff --git a/__tests__/hooks/worker-invocation-dedup.test.ts b/__tests__/hooks/worker-invocation-dedup.test.ts index 0cce5dd1e..49a0b1c34 100644 --- a/__tests__/hooks/worker-invocation-dedup.test.ts +++ b/__tests__/hooks/worker-invocation-dedup.test.ts @@ -489,6 +489,27 @@ customPolicies.add({ name: "runtime-guard", match: { events: ["PreToolUse"] }, expect(getAllHookActivityEntries()).toHaveLength(2); }); + it("answers a pending duplicate with its fallback within the caller's transport budget", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const gate = holdOwner(); + const owner = send(); + await gate.started; + const loaded = observeNextLoad(); + const duplicate = send(); + let settled = false; + void duplicate.then(() => { settled = true; }); + await loaded; + await vi.advanceTimersByTimeAsync(29_999); + try { + expect(settled).toBe(true); + expect(await duplicate).toMatchObject({ type: "hookResult", evaluation: { decision: "deny" } }); + expect(evaluate).toHaveBeenCalledTimes(2); + } finally { + gate.finish(); + await owner; + } + }); + it("bounds a pending duplicate's wait and evaluates normally when its owner never settles", async () => { vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); const gate = holdOwner(); diff --git a/src/hooks/daemon-client.ts b/src/hooks/daemon-client.ts index 6b5683168..47ced9f47 100644 --- a/src/hooks/daemon-client.ts +++ b/src/hooks/daemon-client.ts @@ -41,7 +41,7 @@ const PROTOCOL_VERSION = 1; * working on. */ const DAEMON_CONNECT_TIMEOUT_MS = 150; -const DAEMON_RESPONSE_TIMEOUT_MS = 30_000; +export const DAEMON_RESPONSE_TIMEOUT_MS = 30_000; const MAX_FRAME_LEN = 16 * 1024 * 1024; diff --git a/src/hooks/worker-server.ts b/src/hooks/worker-server.ts index 3eab096ad..aefa8b23d 100644 --- a/src/hooks/worker-server.ts +++ b/src/hooks/worker-server.ts @@ -29,10 +29,13 @@ import { evaluateHookEvent } from "./handler"; import type { HookEventOutcome } from "./handler"; import type { IntegrationType } from "./types"; import { hookLogWarn } from "./hook-logger"; +import { DAEMON_RESPONSE_TIMEOUT_MS } from "./daemon-client"; const MAX_FRAME_LEN = 16 * 1024 * 1024; const INVOCATION_TTL_MS = 10 * 60_000; -const INVOCATION_WAIT_MS = 30_000; +// The duplicate shares the caller's transport budget; reserve a window for fallback evaluation. +const INVOCATION_FALLBACK_RESERVE_MS = 10_000; +const INVOCATION_WAIT_MS = DAEMON_RESPONSE_TIMEOUT_MS - INVOCATION_FALLBACK_RESERVE_MS; interface HookInvocation { result: Promise; From 0480e7e36bfa1b7776b3e993be33b2ba7b7d944f Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Fri, 23 Oct 2026 05:13:23 +0200 Subject: [PATCH 6/6] Remove the worker-dedup test's temp root in teardown Co-Authored-By: GPT-6.1 Sol --- __tests__/hooks/worker-invocation-dedup.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/__tests__/hooks/worker-invocation-dedup.test.ts b/__tests__/hooks/worker-invocation-dedup.test.ts index 49a0b1c34..e37274eed 100644 --- a/__tests__/hooks/worker-invocation-dedup.test.ts +++ b/__tests__/hooks/worker-invocation-dedup.test.ts @@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi, type MockInstance } from "vitest"; import { createConnection, type Server, type Socket } from "node:net"; import { createHash } from "node:crypto"; -import { mkdtempSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { existsSync, mkdtempSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import * as handler from "../../src/hooks/handler"; @@ -97,7 +97,8 @@ describe("worker invocation dedup (real socket, evaluator and activity store)", vi.restoreAllMocks(); vi.unstubAllEnvs(); _resetForTest(); - // Retain isolated artifacts for inspection; no writes reach the real home. + rmSync(dir, { recursive: true, force: true }); + expect(existsSync(dir)).toBe(false); }); function request(payload: Record = {}, hookEvent = "PreToolUse", cli = "claude") {