diff --git a/.changeset/fn-9181-recall-capture-bounded-audit.md b/.changeset/fn-9181-recall-capture-bounded-audit.md new file mode 100644 index 0000000000..9a0f85a575 --- /dev/null +++ b/.changeset/fn-9181-recall-capture-bounded-audit.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Prevent stalled recall telemetry from retaining detached memory captures. +category: fix +dev: Routes packages/core/src/memory/recall-capture.ts through emitBoundedRunAudit. diff --git a/AGENTS.md b/AGENTS.md index 28e9d8e72d..7b18fce4ab 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -295,6 +295,7 @@ canonical emitters remain explicit exclusions until their separately scoped hard - FNXC:RunAudit 2026-08-20-05:49: FN-9177 requires new core best-effort emitters to use `packages/core/src/run-audit/emit-bounded-run-audit.ts`. It deliberately mirrors the engine seam because core cannot import engine; transactional and deliberately awaited durability writers remain unbounded. - FNXC:RunAudit 2026-08-20-07:14: FN-9182 requires core emitters whose behavior branches on audit success to use `emitBoundedRunAuditWithOutcome`; `emitBoundedRunAudit` remains the default best-effort seam, while transactional and deliberately awaited durability writers remain unbounded. - FN-9180: The `task-deleted-outbox:*` catch-up, reconciliation-fallback, lease-fenced, and retention-pruned emitters use `packages/core/src/run-audit/emit-bounded-run-audit.ts`, remain awaited at their post-cursor/post-DELETE positions, and require hostile-sink production-path coverage. +- FN-9181: Detached recall-capture `memory:capture-recorded` and `memory:capture-failed` emissions use the core bounded seam, preserving the injectable test adapter and preventing stalled telemetry from retaining detached capture work; see `docs/run-audit.md`. - FN-9109: `session:cross-runtime-fallback-engaged` records a single retryable-failure handoff from a primary runtime to a deferred CLI runtime. Metadata is ids/outcomes-only (`sessionPurpose`, primary/fallback provider and model IDs, trigger point, failure category, `contextTransferred`); never record error prose or transferred transcript text. - FN-8958: `merge:orphan-write-fenced` is emitted once per orphan merge body at its fence's first interaction. Metadata is ids/counts/outcomes-only: `{ taskId, category, interaction, suppressedCount }`; `suppressedCount` is the emit-time count (`1` for `interaction:"suppressed"`, `0` for `interaction:"rejected"`), never a cumulative body total. diff --git a/docs/run-audit.md b/docs/run-audit.md index 8b1d6fb0d7..d8968bdc8d 100644 --- a/docs/run-audit.md +++ b/docs/run-audit.md @@ -82,10 +82,10 @@ This applies to executor, run-auditor, self-healing, merger, PR reconciliation, ### Core emit-seam policy -Core best-effort emitters use `packages/core/src/run-audit/emit-bounded-run-audit.ts`. This is a deliberate copy of the engine seam because `@fusion/core` cannot import `@fusion/engine`; it synchronously invokes a valid sink, then absorbs throws, rejection, timeout, and late settlement without making telemetry lifecycle-load-bearing. `emitBoundedRunAudit` is the default void seam. `emitBoundedRunAuditWithOutcome` returns `recorded`, `absent`, `failed` (with the original error), or `timed-out` where a forensic throw ordering or caller-visible skipped payload depends on the audit result; workflow-switch torn reconciliation and phantom committed-reservation reconciliation use it. Transactional writers and explicitly awaited durability/ordering writers remain unbounded. `packages/core/src/__tests__/core-run-audit-sink-health.test.ts` and `core-run-audit-emitter-isolation.test.ts` respectively enforce hostile-sink behavior and source routing. +Core best-effort emitters use `packages/core/src/run-audit/emit-bounded-run-audit.ts`. This is a deliberate copy of the engine seam because `@fusion/core` cannot import `@fusion/engine`; it synchronously invokes a valid sink, then absorbs throws, rejection, timeout, and late settlement without making telemetry lifecycle-load-bearing. `emitBoundedRunAudit` is the default void seam. `emitBoundedRunAuditWithOutcome` returns `recorded`, `absent`, `failed` (with the original error), or `timed-out` where a forensic throw ordering or caller-visible skipped payload depends on the audit result; workflow-switch torn reconciliation and phantom committed-reservation reconciliation use it. FN-9181 applies FN-9178's class-A decision to detached recall capture: `memory:capture-recorded` and `memory:capture-failed` are bounded, while the injectable `deps.audit` adapter remains a test seam with its existing bare-metadata contract. Transactional writers and explicitly awaited durability/ordering writers remain unbounded. `packages/core/src/__tests__/core-run-audit-sink-health.test.ts` and `core-run-audit-emitter-isolation.test.ts` respectively enforce hostile-sink behavior and source routing. ### Awaited core exclusion decision -FN-9178 classified awaited sites with hostile-sink characterization tests. FN-9180 routed the class-A `task-deleted-outbox:catch-up`, `:reconciliation-fallback`, `:lease-fenced`, and `:retention-pruned` rows through `emitBoundedRunAudit`; each remains awaited at its post-acknowledgement, post-cursor, or post-DELETE position so bounded telemetry preserves ordering. Recall capture remains class A and is owned by FN-9181. `task:workflow-switch-torn` and `task:reconcile-phantom-committed-reservation` are class B and use the bounded outcome seam because their throw/result payload depends on audit outcome. `task:bypass-review`, `task:resume-step`, and both resurrection-blocked records are class C and intentionally unbounded: they claim persistence before return, destructive cleanup, or a forensic throw. +FN-9178 classified awaited sites with hostile-sink characterization tests. FN-9180 routed the class-A `task-deleted-outbox:catch-up`, `:reconciliation-fallback`, `:lease-fenced`, and `:retention-pruned` rows through `emitBoundedRunAudit`; each remains awaited at its post-acknowledgement, post-cursor, or post-DELETE position so bounded telemetry preserves ordering. FN-9181 routed detached recall capture through the same bounded seam. `task:workflow-switch-torn` and `task:reconcile-phantom-committed-reservation` are class B and use the bounded outcome seam because their throw/result payload depends on audit outcome. `task:bypass-review`, `task:resume-step`, and both resurrection-blocked records are class C and intentionally unbounded: they claim persistence before return, destructive cleanup, or a forensic throw. All `recordRunAuditEventWithinTransaction(tx, ...)` calls and the `recordRunAuditEventBackend(tx, ...)` transactional call are permanently out of scope. Their audit row shares a transaction with the mutation it describes; bounding would split that atomicity. The full matrix and evidence pointers are in the FN-9178 `decision` task document; `excluded-awaited-run-audit-store-sites.test.ts`, `excluded-awaited-run-audit-layer-sites.test.ts`, and the core routing ratchet pin this boundary. diff --git a/packages/core/src/__tests__/core-run-audit-emitter-isolation.test.ts b/packages/core/src/__tests__/core-run-audit-emitter-isolation.test.ts index 38dea19abf..f916531375 100644 --- a/packages/core/src/__tests__/core-run-audit-emitter-isolation.test.ts +++ b/packages/core/src/__tests__/core-run-audit-emitter-isolation.test.ts @@ -14,7 +14,6 @@ const awaitedClassifications = { "store.ts:task:resume-step": "C", "task-store/task-creation.ts:intake:resurrection-blocked": "C", "task-store/task-id-integrity.ts:task:resurrection-blocked": "C", - "memory/recall-capture.ts:memory:capture-recorded|memory:capture-failed": "A", "task-store/project-store-ops.ts:recordRunAuditEventImpl": "permanent-sink", } as const; @@ -32,7 +31,7 @@ const files = [ "../task-store/workflow-workitems-ops.ts", "../task-store/workflow-workitems-ops-2.ts", "../task-store/task-artifacts-ops.ts", "../task-store/lifecycle-ops.ts", "../task-store/task-id-integrity.ts", "../task-store/workflow-definitions.ts", "../task-store/async/async-phantom-reservations.ts", "../task-store/task-deleted-outbox-consumer.ts", - "../task-store/task-lifecycle-event-retention.ts", + "../task-store/task-lifecycle-event-retention.ts", "../memory/recall-capture.ts", ]; const sourceRoot = fileURLToPath(new URL("..", import.meta.url)); diff --git a/packages/core/src/__tests__/core-run-audit-sink-health.test.ts b/packages/core/src/__tests__/core-run-audit-sink-health.test.ts index 6197ea12c2..b4367b31eb 100644 --- a/packages/core/src/__tests__/core-run-audit-sink-health.test.ts +++ b/packages/core/src/__tests__/core-run-audit-sink-health.test.ts @@ -1,7 +1,8 @@ import { afterEach, describe, expect, it, vi } from "vitest"; -const { asyncAuditSink, workflowReader, resolvedWorkflowIr } = vi.hoisted(() => ({ +const { asyncAuditSink, recallAppend, workflowReader, resolvedWorkflowIr } = vi.hoisted(() => ({ asyncAuditSink: { current: undefined as undefined | ((event: unknown) => unknown) }, + recallAppend: { current: undefined as undefined | ((input: unknown) => unknown) }, workflowReader: { current: undefined as undefined | (() => unknown) }, resolvedWorkflowIr: { current: undefined as unknown }, })); @@ -14,6 +15,10 @@ vi.mock("../postgres/data-layer.js", async (importOriginal) => { return { ...actual, recordRunAuditEvent: (_layer: unknown, event: unknown) => asyncAuditSink.current?.(event) }; }); +vi.mock("../memory/recall/recall-store.js", () => ({ + appendRecall: (_layer: unknown, input: unknown) => recallAppend.current?.(input), +})); + vi.mock("../task-store/async/async-persistence.js", () => ({ readTaskRow: vi.fn(async () => workflowReader.current?.()), })); @@ -54,6 +59,7 @@ import { emitOverseerRetry, emitOverseerSteering, } from "../planner/planner-overseer-events.js"; +import { createRecallCaptureWriter } from "../memory/recall-capture.js"; import { CORE_RUN_AUDIT_EMIT_TIMEOUT_MS } from "../run-audit/emit-bounded-run-audit.js"; import { runPluginColumnTransitionHooksImpl } from "../task-store/audit-ops.js"; import { rehomeOccupantImpl } from "../task-store/branch-group-ops.js"; @@ -102,9 +108,61 @@ async function settleBounded(promise: Promise, mode: SinkMode, settle: () await expect(promise).resolves.toBeUndefined(); } -afterEach(() => vi.useRealTimers()); +afterEach(() => { + asyncAuditSink.current = undefined; + recallAppend.current = undefined; + workflowReader.current = undefined; + resolvedWorkflowIr.current = undefined; + vi.useRealTimers(); +}); describe("core audit emitters tolerate hostile sinks", () => { + it.each(["absent", "throw", "reject", "never", "late-resolve", "late-reject"] as const)( + "keeps production recall-capture audit bounded for %s layer sinks", async (mode) => { + vi.useFakeTimers(); + const hostile = hostileSink(mode); + asyncAuditSink.current = mode === "absent" ? undefined : hostile.sink; + recallAppend.current = () => ({ + status: "created", + record: { id: "recall-9177", projectId: "project", kind: "decision", content: "stored", contentHash: "hash", source: { origin: "other" }, tags: [], graphNodeIds: [], createdAt: "2026-08-20T00:00:00.000Z", updatedAt: "2026-08-20T00:00:00.000Z" }, + }); + const writer = createRecallCaptureWriter({ layer: {} as never, logger: { warn: vi.fn() } }); + expect(writer.capture({ origin: "insight", summary: "must not reach audit", insightId: "INS-9177" })).toBeUndefined(); + const drain = writer.flushPendingCaptures(); + if (mode === "never" || mode.startsWith("late-")) { + await vi.advanceTimersByTimeAsync(CORE_RUN_AUDIT_EMIT_TIMEOUT_MS); + hostile.settle(); + } + await expect(drain).resolves.toBeUndefined(); + await expect(writer.flushPendingCaptures()).resolves.toBeUndefined(); + if (mode !== "absent") expect(hostile.sink).toHaveBeenCalled(); + if (mode === "late-reject") await Promise.resolve(); + if (mode === "absent") return; + const event = hostile.sink.mock.calls[0]?.[0] as Record; + expect(event).toMatchObject({ agentId: "memory-capture", domain: "database", mutationType: "memory:capture-recorded", target: "INS-9177" }); + expect(event.metadata).toMatchObject({ origin: "insight", recallRecordId: "recall-9177", outcome: "created" }); + expect(JSON.stringify(event)).not.toContain("must not reach audit"); + }, + ); + + it("records the production layer failure event and suppresses the layer when append is injected", async () => { + const sink = vi.fn(); + asyncAuditSink.current = sink; + recallAppend.current = () => Promise.reject(new TypeError("append failure")); + const writer = createRecallCaptureWriter({ layer: {} as never, logger: { warn: vi.fn() } }); + writer.capture({ origin: "research-finding", summary: "private failure summary", findingId: "RF-9177" }); + await writer.flushPendingCaptures(); + expect(sink).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "memory:capture-failed", target: "RF-9177", metadata: { origin: "research-finding", errorClass: "TypeError" }, + })); + + const injectedAppend = vi.fn(async () => ({ status: "created" as const, record: { id: "recall-injected" } })); + const injectedWriter = createRecallCaptureWriter({ layer: {} as never, logger: { warn: vi.fn() }, append: injectedAppend as never }); + injectedWriter.capture({ origin: "insight", summary: "append injection suppresses audit" }); + await injectedWriter.flushPendingCaptures(); + expect(sink).toHaveBeenCalledTimes(1); + }); + it.each(["absent", "throw", "reject", "never", "late-resolve", "late-reject"] as const)( "keeps every planner façade bounded for %s sinks", async (mode) => { vi.useFakeTimers(); diff --git a/packages/core/src/__tests__/memory/recall-capture.test.ts b/packages/core/src/__tests__/memory/recall-capture.test.ts index 14dd70cd9e..0fd676cb3c 100644 --- a/packages/core/src/__tests__/memory/recall-capture.test.ts +++ b/packages/core/src/__tests__/memory/recall-capture.test.ts @@ -1,4 +1,5 @@ -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { CORE_RUN_AUDIT_EMIT_TIMEOUT_MS } from "../../run-audit/emit-bounded-run-audit.js"; import { buildRecallCaptureContent, createRecallCaptureWriter, @@ -11,6 +12,31 @@ import type { RecallAppendInput } from "../../memory/recall/recall-types.js"; const layer = {} as never; const logger = { warn: vi.fn() }; +type AuditSinkMode = "absent" | "throw" | "reject" | "never" | "late-resolve" | "late-reject"; + +function hostileAudit(mode: AuditSinkMode) { + let settle: (() => void) | undefined; + const audit = vi.fn(() => { + if (mode === "throw") throw new Error("audit throw"); + if (mode === "reject") return Promise.reject(new Error("audit rejection")); + if (mode === "never") return new Promise(() => undefined); + if (mode === "late-resolve") return new Promise((resolve) => { settle = resolve; }); + if (mode === "late-reject") return new Promise((_resolve, reject) => { settle = () => reject(new Error("late audit rejection")); }); + return undefined; + }); + return { audit, settle: () => settle?.() }; +} + +async function drainBounded(writer: ReturnType, mode: AuditSinkMode, settle: () => void) { + const drain = writer.flushPendingCaptures(); + if (mode === "never" || mode.startsWith("late-")) { + await vi.advanceTimersByTimeAsync(CORE_RUN_AUDIT_EMIT_TIMEOUT_MS); + settle(); + } + await expect(drain).resolves.toBeUndefined(); + await expect(writer.flushPendingCaptures()).resolves.toBeUndefined(); +} + function created(input: RecallAppendInput) { return { status: "created" as const, @@ -22,6 +48,8 @@ function created(input: RecallAppendInput) { }; } +afterEach(() => vi.useRealTimers()); + describe("recall capture writer", () => { it("maps every automatic origin to FN-8922 kinds and source provenance", async () => { const append = vi.fn(async (input: RecallAppendInput) => created(input)); @@ -64,6 +92,47 @@ describe("recall capture writer", () => { expect(warn).toHaveBeenCalledWith("Automatic recall capture failed for research-finding"); }); + it.each(["absent", "throw", "reject", "never", "late-resolve", "late-reject"] as const)( + "keeps injectable capture audit bounded for %s sinks", async (mode) => { + vi.useFakeTimers(); + const hostile = hostileAudit(mode); + const append = vi.fn(async (input: RecallAppendInput) => created(input)); + const warn = vi.fn(); + const writer = createRecallCaptureWriter({ + layer, + logger: { warn }, + append, + ...(mode === "absent" ? {} : { audit: hostile.audit }), + }); + + expect(writer.capture({ origin: "insight", summary: "secret capture summary", insightId: "INS-1" })).toBeUndefined(); + await drainBounded(writer, mode, hostile.settle); + expect(append).toHaveBeenCalledTimes(1); + if (mode === "absent") expect(hostile.audit).not.toHaveBeenCalled(); + else expect(hostile.audit).toHaveBeenCalledWith({ type: "memory:capture-recorded", metadata: { recallRecordId: "recall-1", outcome: "created" } }); + expect(JSON.stringify([hostile.audit.mock.calls, warn.mock.calls])).not.toContain("secret capture summary"); + }, + ); + + it("preserves duplicate and failure adapter metadata without invoking the layer sink", async () => { + const audit = vi.fn(async () => {}); + const duplicate = { ...created({ kind: "decision", content: "x", source: { origin: "other" } }), id: "recall-duplicate" }; + const append = vi.fn() + .mockResolvedValueOnce({ status: "duplicate" as const, duplicateOf: duplicate, similarity: 1 }) + .mockRejectedValueOnce(new TypeError("append failure")); + const warn = vi.fn(); + const writer = createRecallCaptureWriter({ layer, logger: { warn }, append, audit }); + + writer.capture({ origin: "insight", summary: "never expose this summary" }); + writer.capture({ origin: "research-finding", summary: "nor this failure summary" }); + await writer.flushPendingCaptures(); + + expect(audit).toHaveBeenNthCalledWith(1, { type: "memory:capture-recorded", metadata: { recallRecordId: "recall-duplicate", outcome: "duplicate" } }); + expect(audit).toHaveBeenNthCalledWith(2, { type: "memory:capture-failed", metadata: { errorClass: "TypeError" } }); + expect(warn).toHaveBeenCalledWith("Automatic recall capture failed for research-finding"); + expect(JSON.stringify([audit.mock.calls, warn.mock.calls])).not.toContain("summary"); + }); + it("records ids-only capture audit metadata after persistence", async () => { const audit = vi.fn(async () => {}); const append = vi.fn(async (input: RecallAppendInput) => created(input)); diff --git a/packages/core/src/memory/recall-capture.ts b/packages/core/src/memory/recall-capture.ts index 105a14a15e..5103c12911 100644 --- a/packages/core/src/memory/recall-capture.ts +++ b/packages/core/src/memory/recall-capture.ts @@ -1,5 +1,7 @@ import { recordRunAuditEvent, type AsyncDataLayer } from "../postgres/data-layer.js"; import type { Logger } from "../process/logger.js"; +import { emitBoundedRunAudit } from "../run-audit/emit-bounded-run-audit.js"; +import type { RunAuditEventInput } from "../types.js"; import { appendRecall } from "./recall/recall-store.js"; import type { RecallAppendInput, RecallAppendResult, RecallKind, RecallOrigin } from "./recall/recall-types.js"; @@ -77,6 +79,11 @@ FNXC:MemoryRecallCapture 2026-08-11-10:57: FN-8922 has no insight-specific RecallOrigin literal, so insight outcomes use its real "other" origin while completed tasks and research use "task-completion" and "deep-research" respectively. The test drain exists solely for deterministic detached-work tests and production must never call it. + +FNXC:RunAudit 2026-08-20-07:03: +FN-9181 implements FN-9178's class-A decision: bounded audit emission prevents a stalled sink from +retaining a detached capture promise. Preserve the injectable adapter's bare-metadata contract, and +suppress the layer path whenever append is injected so the deterministic test seam remains isolated. */ export const NOOP_RECALL_CAPTURE_WRITER: RecallCaptureWriter = Object.freeze({ capture: () => {}, @@ -134,10 +141,18 @@ export function createRecallCaptureWriter( const pending = new Set>(); const append = deps.append ?? ((input: RecallAppendInput) => appendRecall(deps.layer, input)); const recordAudit = async (type: "memory:capture-recorded" | "memory:capture-failed", input: RecallCaptureInput, metadata: Record) => { + const log = { warn: () => deps.logger.warn(`Automatic recall capture audit failed for ${input.origin}`) }; if (deps.audit) { - await deps.audit({ type, metadata }); + await emitBoundedRunAudit({ + recordRunAuditEvent: (event) => deps.audit?.({ + type: event.mutationType as "memory:capture-recorded" | "memory:capture-failed", + metadata, + }), + }, { mutationType: type }, { log }); } else if (!deps.append) { - await recordRunAuditEvent(deps.layer, { + await emitBoundedRunAudit({ + recordRunAuditEvent: (event) => recordRunAuditEvent(deps.layer, event as RunAuditEventInput), + }, { agentId: input.agentId ?? "memory-capture", runId: `memory-capture:${input.origin}:${Date.now()}`, taskId: input.taskId, @@ -145,7 +160,7 @@ export function createRecallCaptureWriter( mutationType: type, target: input.insightId ?? input.findingId ?? input.researchRunId ?? input.taskId ?? input.origin, metadata: { origin: input.origin, ...metadata }, - }); + }, { log }); } }; @@ -154,24 +169,16 @@ export function createRecallCaptureWriter( const operation = (async () => { try { const result = await append(toAppendInput(input)); - try { - await recordAudit("memory:capture-recorded", input, { - recallRecordId: result.status === "created" ? result.record.id : result.duplicateOf.id, - outcome: result.status, - }); - } catch { - deps.logger.warn(`Automatic recall capture audit failed for ${input.origin}`); - } + await recordAudit("memory:capture-recorded", input, { + recallRecordId: result.status === "created" ? result.record.id : result.duplicateOf.id, + outcome: result.status, + }); } catch (error) { // Recall content can be sensitive, so diagnostics identify only the bounded origin. deps.logger.warn(`Automatic recall capture failed for ${input.origin}`); - try { - await recordAudit("memory:capture-failed", input, { - errorClass: error instanceof Error ? error.name : "unknown", - }); - } catch { - deps.logger.warn(`Automatic recall capture audit failed for ${input.origin}`); - } + await recordAudit("memory:capture-failed", input, { + errorClass: error instanceof Error ? error.name : "unknown", + }); } })(); pending.add(operation);