diff --git a/.changeset/fn-9180-outbox-run-audit.md b/.changeset/fn-9180-outbox-run-audit.md new file mode 100644 index 0000000000..503d29d2e9 --- /dev/null +++ b/.changeset/fn-9180-outbox-run-audit.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Keep deleted-task outbox delivery resilient when audit telemetry fails. +category: fix +dev: Routes catch-up, reconciliation-fallback, lease-fenced, and retention-pruned through the core bounded audit seam. diff --git a/AGENTS.md b/AGENTS.md index e1f6c26143..28e9d8e72d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -294,6 +294,7 @@ canonical emitters remain explicit exclusions until their separately scoped hard - FN-9175: New engine run-audit emitters must use `emitBoundedRunAudit` from `packages/engine/src/util/emit-bounded-run-audit.ts`; it absorbs absent, throwing, rejecting, hanging, and late-settling sinks without changing the owning branch, and requires behavioral sink-health coverage. - 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-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 2bcbb263c9..8b1d6fb0d7 100644 --- a/docs/run-audit.md +++ b/docs/run-audit.md @@ -86,6 +86,6 @@ Core best-effort emitters use `packages/core/src/run-audit/emit-bounded-run-audi ### Awaited core exclusion decision -FN-9178 classified the remaining awaited sites with hostile-sink characterization tests. `task-deleted-outbox:catch-up`, `:reconciliation-fallback`, `:lease-fenced`, `:retention-pruned`, and recall-capture audit events are class A (bound-safe) candidates. `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. 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. 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 9641cea8f4..38dea19abf 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 @@ -5,19 +5,15 @@ import { describe, expect, it } from "vitest"; /* * FNXC:RunAudit 2026-08-20-07:16: * FN-9178 makes every direct awaited core audit writer a named decision rather than an implicit - * exception. Class A sites are evaluated candidates, C retains its ordering claim, and - * transactional/sink writers remain permanent atomicity boundaries. FN-9182 migrated class B - * sites to the reporting bounded seam, so they no longer appear in this direct-await inventory. + * exception. Class C retains its ordering claim while transactional/sink writers remain permanent + * atomicity boundaries. FN-9180 routed class-A outbox rows and FN-9182 routed class-B sites + * through bounded seams, so neither class remains in this direct-await inventory. */ const awaitedClassifications = { "store.ts:task:bypass-review": "C", "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", - "task-store/task-deleted-outbox-consumer.ts:task-deleted-outbox:catch-up": "A", - "task-store/task-deleted-outbox-consumer.ts:task-deleted-outbox:reconciliation-fallback": "A", - "task-store/task-deleted-outbox-consumer.ts:task-deleted-outbox:lease-fenced": "A", - "task-store/task-lifecycle-event-retention.ts:task-deleted-outbox:retention-pruned": "A", "memory/recall-capture.ts:memory:capture-recorded|memory:capture-failed": "A", "task-store/project-store-ops.ts:recordRunAuditEventImpl": "permanent-sink", } as const; @@ -35,7 +31,8 @@ const files = [ "../task-store/merge-queue-ops-2.ts", "../task-store/task-mutation-ops.ts", "../task-store/workflow-integrity.ts", "../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/async/async-phantom-reservations.ts", "../task-store/task-deleted-outbox-consumer.ts", + "../task-store/task-lifecycle-event-retention.ts", ]; const sourceRoot = fileURLToPath(new URL("..", import.meta.url)); diff --git a/packages/core/src/__tests__/excluded-awaited-run-audit-layer-sites.test.ts b/packages/core/src/__tests__/excluded-awaited-run-audit-layer-sites.test.ts index 89a7f492ae..9bc4aed970 100644 --- a/packages/core/src/__tests__/excluded-awaited-run-audit-layer-sites.test.ts +++ b/packages/core/src/__tests__/excluded-awaited-run-audit-layer-sites.test.ts @@ -1,26 +1,8 @@ import { afterEach, describe, expect, it, vi } from "vitest"; -const audit = vi.hoisted(() => vi.fn()); const asyncAudit = vi.hoisted(() => vi.fn()); const softDelete = vi.hoisted(() => vi.fn().mockResolvedValue(undefined)); const readTaskRow = vi.hoisted(() => vi.fn()); -const lifecycle = vi.hoisted(() => ({ - acknowledgeTaskLifecycleEvent: vi.fn(async () => true), - acquireTaskLifecycleLease: vi.fn(async () => ({ token: "lease", fencingToken: 1n })), - advanceTaskLifecycleConsumerCursor: vi.fn(async () => true), - hasTaskLifecycleConsumerReceipt: vi.fn(async () => false), - listTaskLifecycleEvents: vi.fn(async () => []), - readTaskLifecycleConsumerCursor: vi.fn(async () => ({ fencingToken: 1n, lastAckedSeq: 0n, retryAttempts: 0, updatedAt: new Date().toISOString() })), - readTaskLifecycleEventBounds: vi.fn(async () => ({ headSeq: 4n, oldestSeq: 1n })), - registerTaskLifecycleConsumer: vi.fn(async () => undefined), - releaseTaskLifecycleLease: vi.fn(async () => undefined), - renewTaskLifecycleLease: vi.fn(async () => true), - setTaskLifecycleConsumerActive: vi.fn(async () => undefined), -})); -vi.mock("../postgres/data-layer.js", async (importOriginal) => ({ - ...(await importOriginal()), - recordRunAuditEvent: audit, -})); vi.mock("../task-store/async/async-audit.js", async (importOriginal) => ({ ...(await importOriginal()), recordRunAuditEvent: asyncAudit, @@ -30,44 +12,17 @@ vi.mock("../task-store/async/async-persistence.js", async (importOriginal) => ({ softDeleteTaskRow: softDelete, readTaskRow, })); -vi.mock("../task-store/task-lifecycle-consumer-registry.js", async (importOriginal) => ({ - ...(await importOriginal()), - acknowledgeTaskLifecycleEvent: lifecycle.acknowledgeTaskLifecycleEvent, - acquireTaskLifecycleLease: lifecycle.acquireTaskLifecycleLease, - advanceTaskLifecycleConsumerCursor: lifecycle.advanceTaskLifecycleConsumerCursor, - hasTaskLifecycleConsumerReceipt: lifecycle.hasTaskLifecycleConsumerReceipt, - listTaskLifecycleEvents: lifecycle.listTaskLifecycleEvents, - readTaskLifecycleConsumerCursor: lifecycle.readTaskLifecycleConsumerCursor, - readTaskLifecycleEventBounds: lifecycle.readTaskLifecycleEventBounds, - registerTaskLifecycleConsumer: lifecycle.registerTaskLifecycleConsumer, - releaseTaskLifecycleLease: lifecycle.releaseTaskLifecycleLease, - renewTaskLifecycleLease: lifecycle.renewTaskLifecycleLease, - setTaskLifecycleConsumerActive: lifecycle.setTaskLifecycleConsumerActive, -})); import { createRecallCaptureWriter } from "../memory/recall-capture.js"; import { resolveSameAgentDuplicateIntake } from "../task-store/task-creation.js"; import { maybeResolveTombstonedTaskIdImpl } from "../task-store/task-id-integrity.js"; import { TombstonedTaskResurrectionError } from "../task-store/errors.js"; -import { pruneTaskLifecycleEvents } from "../task-store/task-lifecycle-event-retention.js"; -import { TaskDeletedOutboxConsumer } from "../task-store/task-deleted-outbox-consumer.js"; /* - * FNXC:RunAudit 2026-08-20-06:40: - * FN-9178 invokes real helper-owned entry points with hostile audit helpers. It records existing - * awaiting/ordering behavior only; fake timers make the never-settling observation deterministic. + * FNXC:RunAudit 2026-08-20-07:42: + * FN-9180 routes class-A outbox rows through bounded telemetry, so this characterization retains + * only intentionally awaited class-C and recall-capture behavior. */ -function emptyQuery() { - const query: Record = {}; - for (const method of ["from", "where", "orderBy", "limit"]) query[method] = () => query; - query.then = (resolve: (value: unknown[]) => unknown) => resolve([]); - return query; -} - -function retentionLayer() { - return { projectId: "project", db: { select: vi.fn(() => emptyQuery()) } } as never; -} - describe("FN-9178 awaited data-layer audit characterization", () => { afterEach(() => { vi.clearAllMocks(); vi.useRealTimers(); }); @@ -151,158 +106,6 @@ describe("FN-9178 awaited data-layer audit characterization", () => { await expect(operation).rejects.toThrow(_state === "synchronous throw" ? "sync" : "reject"); } }); - it.each([ - ["absent", () => undefined], - ["synchronous throw", () => { throw new Error("sync"); }], - ["rejection", () => Promise.reject(new Error("rejected"))], - ])("drives retention pruning through a %s audit helper", async (_state, sink) => { - audit.mockImplementation(sink as never); - const layer = retentionLayer(); - const operation = pruneTaskLifecycleEvents(layer, "project"); - if (_state === "absent") await expect(operation).resolves.toMatchObject({ prunedCount: 0 }); - else await expect(operation).rejects.toThrow(); - expect(audit).toHaveBeenCalledWith(layer, expect.objectContaining({ mutationType: "task-deleted-outbox:retention-pruned" })); - }); - - it("retention pruning remains pending for a never-settling audit helper", async () => { - const layer = retentionLayer(); - audit.mockImplementation(() => new Promise(() => undefined)); - vi.useFakeTimers(); - try { - let settled = false; - void pruneTaskLifecycleEvents(layer, "project").finally(() => { settled = true; }).catch(() => undefined); - await vi.advanceTimersByTimeAsync(2_100); - expect(settled).toBe(false); - } finally { vi.useRealTimers(); } - }); - - it("a late retention audit delays the real return until it settles", async () => { - const layer = retentionLayer(); - let resolve!: () => void; - audit.mockImplementation(() => new Promise((done) => { resolve = done; })); - audit.mockClear(); - const operation = pruneTaskLifecycleEvents(layer, "project"); - await vi.waitFor(() => expect(audit).toHaveBeenCalled()); - resolve(); - await expect(operation).resolves.toMatchObject({ prunedCount: 0 }); - }); - - it.each([ - ["absent", () => undefined, true], - ["synchronous throw", () => { throw new Error("sync"); }, false], - ["rejection", () => Promise.reject(new Error("rejected")), false], - ])("reconciliation fallback advances its cursor before a %s audit helper", async (_state, sink, completes) => { - audit.mockImplementation(sink as never); - const layer = retentionLayer(); - const store = { asyncLayer: layer, consumerId: "consumer", taskCache: new Map(), emitObservedTaskDeleted: vi.fn() } as never; - const operation = (new TaskDeletedOutboxConsumer(store) as never).reconcile(0n, { fencingToken: 1n }, "pruned-gap"); - if (completes) await expect(operation).resolves.toBe(true); else await expect(operation).rejects.toThrow(); - expect(lifecycle.advanceTaskLifecycleConsumerCursor).toHaveBeenCalledOnce(); - expect(audit).toHaveBeenCalledWith(layer, expect.objectContaining({ mutationType: "task-deleted-outbox:reconciliation-fallback" })); - }); - - it.each(["never-settling", "late-settling"])("reconciliation fallback waits after its cursor advance for %s audit", async (state) => { - const layer = retentionLayer(); - audit.mockImplementation((state === "never-settling" ? () => new Promise(() => undefined) : () => new Promise((resolve) => setTimeout(resolve, 2_100))) as never); - const store = { asyncLayer: layer, consumerId: "consumer", taskCache: new Map(), emitObservedTaskDeleted: vi.fn() } as never; - vi.useFakeTimers(); - let settled = false; - void (new TaskDeletedOutboxConsumer(store) as never).reconcile(0n, { fencingToken: 1n }, "pruned-gap").finally(() => { settled = true; }).catch(() => undefined); - await vi.advanceTimersByTimeAsync(2_100); - expect(lifecycle.advanceTaskLifecycleConsumerCursor).toHaveBeenCalledOnce(); - expect(settled).toBe(state === "late-settling"); - }); - - function catchUpStore() { - return { - asyncLayer: retentionLayer(), consumerId: "consumer", taskCache: new Map([["FN-DELETED", { id: "FN-DELETED" }]]), - emitObservedTaskDeleted: vi.fn(), - } as never; - } - - function deletedEvent() { - return { - eventId: "event-1", eventType: "task:deleted", taskId: "FN-DELETED", occurredAt: new Date().toISOString(), seq: 1n, - payload: { taskId: "FN-DELETED", previousColumn: "todo", previousStatus: null, deletedAt: new Date().toISOString(), allowResurrection: false, githubIssueAction: null, closureContext: null, deletedBy: null }, - }; - } - - it.each([ - ["absent", () => undefined, true], - ["synchronous throw", () => { throw new Error("sync"); }, false], - ["rejection", () => Promise.reject(new Error("rejected")), false], - ])("drives catch-up through the production poll batch before a %s audit helper", async (_state, sink, completes) => { - lifecycle.listTaskLifecycleEvents.mockResolvedValueOnce([deletedEvent()]); - audit.mockImplementation(sink as never); - const consumer = new TaskDeletedOutboxConsumer(catchUpStore()); - /* - * FNXC:RunAudit 2026-08-20-07:04: - * The characterization must use poll's production batch path. Set only its lifecycle-owned - * running gate to avoid adding a background scheduler while retaining lease-to-release order. - */ - (consumer as never).running = true; - const operation = consumer.poll(); - if (completes) await expect(operation).resolves.toBe("active"); else await expect(operation).rejects.toThrow(); - expect(lifecycle.acknowledgeTaskLifecycleEvent).toHaveBeenCalledOnce(); - expect(audit).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ mutationType: "task-deleted-outbox:catch-up" })); - expect(lifecycle.releaseTaskLifecycleLease).toHaveBeenCalledOnce(); - }); - - it("keeps the real catch-up poll and its lease cleanup pending for a never-settling audit", async () => { - lifecycle.listTaskLifecycleEvents.mockResolvedValueOnce([deletedEvent()]); - audit.mockImplementation(() => new Promise(() => undefined)); - const consumer = new TaskDeletedOutboxConsumer(catchUpStore()); - (consumer as never).running = true; - vi.useFakeTimers(); - try { - let settled = false; - void consumer.poll().finally(() => { settled = true; }).catch(() => undefined); - await vi.advanceTimersByTimeAsync(2_100); - expect(lifecycle.acknowledgeTaskLifecycleEvent).toHaveBeenCalledOnce(); - expect(settled).toBe(false); - expect(lifecycle.releaseTaskLifecycleLease).not.toHaveBeenCalled(); - } finally { vi.useRealTimers(); } - }); - - it("finishes the real catch-up poll only after a late audit settles beyond the bounded window", async () => { - lifecycle.listTaskLifecycleEvents.mockResolvedValueOnce([deletedEvent()]); - let resolveAudit!: () => void; - audit.mockImplementation(() => new Promise((resolve) => { resolveAudit = resolve; })); - const consumer = new TaskDeletedOutboxConsumer(catchUpStore()); - (consumer as never).running = true; - vi.useFakeTimers(); - try { - const operation = consumer.poll(); - await vi.advanceTimersByTimeAsync(2_100); - expect(lifecycle.acknowledgeTaskLifecycleEvent).toHaveBeenCalledOnce(); - expect(lifecycle.releaseTaskLifecycleLease).not.toHaveBeenCalled(); - resolveAudit(); - await expect(operation).resolves.toBe("active"); - expect(lifecycle.releaseTaskLifecycleLease).toHaveBeenCalledOnce(); - } finally { vi.useRealTimers(); } - }); - - it.each([ - ["absent", () => undefined, true], - ["synchronous throw", () => { throw new Error("sync"); }, false], - ["rejection", () => Promise.reject(new Error("rejected")), false], - ])("drives the real outbox lease-fenced entry through a %s audit helper", async (_state, sink, completes) => { - audit.mockImplementation(sink as never); - const store = { asyncLayer: retentionLayer(), consumerId: "consumer" } as never; - const operation = (new TaskDeletedOutboxConsumer(store) as never).recordLeaseFenced({ fencingToken: 1n }, 1); - if (completes) await expect(operation).resolves.toBeUndefined(); else await expect(operation).rejects.toThrow(); - expect(audit).toHaveBeenCalledWith(store.asyncLayer, expect.objectContaining({ mutationType: "task-deleted-outbox:lease-fenced" })); - }); - - it.each(["never-settling", "late-settling"])("outbox lease-fenced awaits a %s audit helper", async (state) => { - audit.mockImplementation((state === "never-settling" ? () => new Promise(() => undefined) : () => new Promise((resolve) => setTimeout(resolve, 2_100))) as never); - const store = { asyncLayer: retentionLayer(), consumerId: "consumer" } as never; - vi.useFakeTimers(); let settled = false; - void (new TaskDeletedOutboxConsumer(store) as never).recordLeaseFenced({ fencingToken: 1n }, 1).finally(() => { settled = true; }).catch(() => undefined); - await vi.advanceTimersByTimeAsync(2_100); - expect(settled).toBe(state === "late-settling"); - }); - it.each([ ["absent", () => undefined], ["synchronous throw", () => { throw new Error("sync"); }], diff --git a/packages/core/src/__tests__/task-deleted-outbox-audit-sink-health.test.ts b/packages/core/src/__tests__/task-deleted-outbox-audit-sink-health.test.ts new file mode 100644 index 0000000000..e788f6aded --- /dev/null +++ b/packages/core/src/__tests__/task-deleted-outbox-audit-sink-health.test.ts @@ -0,0 +1,117 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { TaskStore } from "../store.js"; +import { CORE_RUN_AUDIT_EMIT_TIMEOUT_MS } from "../run-audit/emit-bounded-run-audit.js"; + +const audit = vi.hoisted(() => ({ sink: undefined as undefined | (() => unknown) })); +const mocks = vi.hoisted(() => ({ + register: vi.fn(), active: vi.fn(), acquire: vi.fn(), readCursor: vi.fn(), bounds: vi.fn(), list: vi.fn(), + receipt: vi.fn(), acknowledge: vi.fn(), advance: vi.fn(), renew: vi.fn(), retry: vi.fn(), park: vi.fn(), release: vi.fn(), +})); + +vi.mock("../postgres/data-layer.js", () => ({ recordRunAuditEvent: vi.fn(() => audit.sink?.()) })); +vi.mock("../task-store/task-lifecycle-consumer-registry.js", () => ({ + registerTaskLifecycleConsumer: mocks.register, setTaskLifecycleConsumerActive: mocks.active, + acquireTaskLifecycleLease: mocks.acquire, readTaskLifecycleConsumerCursor: mocks.readCursor, + readTaskLifecycleEventBounds: mocks.bounds, listTaskLifecycleEvents: mocks.list, + hasTaskLifecycleConsumerReceipt: mocks.receipt, acknowledgeTaskLifecycleEvent: mocks.acknowledge, + advanceTaskLifecycleConsumerCursor: mocks.advance, renewTaskLifecycleLease: mocks.renew, + setTaskLifecycleConsumerRetry: mocks.retry, parkTaskLifecycleConsumerDeadLetter: mocks.park, + releaseTaskLifecycleLease: mocks.release, +})); + +import { TaskDeletedOutboxConsumer } from "../task-store/task-deleted-outbox-consumer.js"; + +type SinkMode = "absent" | "throw" | "reject" | "never" | "late"; +const cursor = (lastAckedSeq = 0n) => ({ lastAckedSeq, retryAttempts: 0, retryBackoffUntil: null, updatedAt: new Date().toISOString() }); +const event = { eventId: "event-1", seq: 1n, eventType: "task:deleted", taskId: "FN-9180", occurredAt: "2026-08-20T00:00:00.000Z", payload: { taskId: "FN-9180", previousColumn: "done", previousStatus: null, deletedAt: "2026-08-20T00:00:00.000Z", allowResurrection: false, githubIssueAction: null, closureContext: null, deletedBy: null } }; + +function configure(mode: SinkMode): () => void { + let settle: (() => void) | undefined; + audit.sink = mode === "absent" ? undefined : () => { + if (mode === "throw") throw new Error("hostile audit"); + if (mode === "reject") return Promise.reject(new Error("hostile audit")); + if (mode === "never") return new Promise(() => undefined); + if (mode === "late") return new Promise((resolve) => { settle = resolve; }); + }; + return () => settle?.(); +} + +async function settle(promise: Promise, mode: SinkMode, late: () => void): Promise { + // Lease fencing can be followed by the batch catch-up row; advance enough bounded windows for + // both awaited class-A emits while keeping the production timeout itself unchanged. + if (mode === "never" || mode === "late") await vi.advanceTimersByTimeAsync(CORE_RUN_AUDIT_EMIT_TIMEOUT_MS * 3); + late(); + return await promise; +} + +function makeConsumer(hasCachedTask = true) { + const store = { asyncLayer: { projectId: "project-9180", db: { select: vi.fn() } }, consumerId: "consumer-9180", taskCache: hasCachedTask ? new Map([["FN-9180", { id: "FN-9180" }]]) : new Map(), emitObservedTaskDeleted: vi.fn() } as unknown as TaskStore; + const consumer = new TaskDeletedOutboxConsumer(store); + (consumer as unknown as { running: boolean }).running = true; + return consumer; +} + +beforeEach(() => { + vi.useFakeTimers(); vi.clearAllMocks(); + mocks.register.mockResolvedValue(undefined); mocks.active.mockResolvedValue(undefined); + mocks.acquire.mockResolvedValue({ token: "lease", fencingToken: 1n, expiresAt: "2026-08-20T00:01:00.000Z" }); + mocks.readCursor.mockResolvedValue(cursor()); mocks.bounds.mockResolvedValue({ oldestSeq: null, oldestOccurredAt: null, headSeq: 1n }); + mocks.list.mockResolvedValue([]); mocks.receipt.mockResolvedValue(false); mocks.acknowledge.mockResolvedValue(true); + mocks.release.mockResolvedValue(undefined); mocks.renew.mockResolvedValue(true); mocks.advance.mockResolvedValue(true); +}); +afterEach(() => { audit.sink = undefined; vi.useRealTimers(); }); + +/* +FNXC:RunAudit 2026-08-20-06:50: +FN-9180 drives the real outbox poll through every class-A audit branch. A bounded awaited emit must +preserve post-cursor ordering while hostile telemetry cannot change delivery's active/idle result. +*/ +describe("TaskDeletedOutboxConsumer bounded audit health", () => { + it.each(["absent", "throw", "reject", "never", "late"] as const)("keeps catch-up delivery active for a %s audit sink", async (mode) => { + const late = configure(mode); mocks.list.mockResolvedValue([event]); + const consumer = makeConsumer(); + await expect(settle(consumer.poll(), mode, late)).resolves.toBe("active"); + expect(mocks.active.mock.invocationCallOrder[0]).toBeGreaterThan(0); + }); + + it.each(["absent", "throw", "reject", "never", "late"] as const)("keeps reconciliation delivery idle for a %s audit sink after cursor advance", async (mode) => { + for (const hasCachedTask of [false, true]) { + const late = configure(mode); mocks.readCursor.mockResolvedValue(cursor(0n)); mocks.bounds.mockResolvedValue({ oldestSeq: 2n, oldestOccurredAt: null, headSeq: 4n }); + const consumer = makeConsumer(hasCachedTask); + const layer = (consumer as unknown as { store: TaskStore }).store.asyncLayer! as unknown as { db: { select: ReturnType } }; + layer.db.select.mockReturnValue({ from: () => ({ where: async () => [] }) }); + await expect(settle(consumer.poll(), mode, late)).resolves.toBe("idle"); + expect(mocks.advance).toHaveBeenCalledTimes(hasCachedTask ? 2 : 1); + } + }); + + it.each(["absent", "throw", "reject", "never", "late"] as const)("keeps lease-fenced delivery active for a %s audit sink", async (mode) => { + const late = configure(mode); mocks.list.mockResolvedValue([event]); mocks.acknowledge.mockResolvedValue(false); + const consumer = makeConsumer(); + await expect(settle(consumer.poll(), mode, late)).resolves.toBe("active"); + expect(mocks.readCursor).toHaveBeenCalledTimes(3); + }); + + it("keeps audit attempts after cursor advance and before consumer activation", async () => { + const order: string[] = []; + audit.sink = () => { order.push("audit"); }; + mocks.active.mockImplementation(async () => { order.push("active"); }); + mocks.advance.mockImplementation(async () => { order.push("advance"); return true; }); + mocks.readCursor.mockResolvedValue(cursor(0n)); mocks.bounds.mockResolvedValue({ oldestSeq: 2n, oldestOccurredAt: null, headSeq: 4n }); + const reconciliationConsumer = makeConsumer(); + const layer = (reconciliationConsumer as unknown as { store: TaskStore }).store.asyncLayer! as unknown as { db: { select: ReturnType } }; + layer.db.select.mockReturnValue({ from: () => ({ where: async () => [] }) }); + await expect(reconciliationConsumer.poll()).resolves.toBe("idle"); + expect(order).toEqual(["advance", "audit", "active"]); + + order.length = 0; mocks.bounds.mockResolvedValue({ oldestSeq: null, oldestOccurredAt: null, headSeq: 1n }); mocks.readCursor.mockResolvedValue(cursor()); mocks.list.mockResolvedValue([event]); + await expect(makeConsumer().poll()).resolves.toBe("active"); + expect(order).toEqual(["audit", "active"]); + }); + + it("does not emit catch-up telemetry for an empty batch", async () => { + const sink = vi.fn(); audit.sink = sink; + await expect(makeConsumer().poll()).resolves.toBe("idle"); + expect(sink).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/core/src/__tests__/task-lifecycle-retention-audit-sink-health.test.ts b/packages/core/src/__tests__/task-lifecycle-retention-audit-sink-health.test.ts new file mode 100644 index 0000000000..9fbb94a39e --- /dev/null +++ b/packages/core/src/__tests__/task-lifecycle-retention-audit-sink-health.test.ts @@ -0,0 +1,64 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { CORE_RUN_AUDIT_EMIT_TIMEOUT_MS } from "../run-audit/emit-bounded-run-audit.js"; + +const audit = vi.hoisted(() => ({ sink: undefined as undefined | (() => unknown) })); +vi.mock("../postgres/data-layer.js", () => ({ recordRunAuditEvent: vi.fn(() => audit.sink?.()) })); + +import { pruneTaskLifecycleEvents } from "../task-store/task-lifecycle-event-retention.js"; + +type SinkMode = "absent" | "throw" | "reject" | "never" | "late"; +function configure(mode: SinkMode): () => void { + let settle: (() => void) | undefined; + audit.sink = mode === "absent" ? undefined : () => { + if (mode === "throw") throw new Error("hostile audit"); + if (mode === "reject") return Promise.reject(new Error("hostile audit")); + if (mode === "never") return new Promise(() => undefined); + if (mode === "late") return new Promise((resolve) => { settle = resolve; }); + }; + return () => settle?.(); +} +async function settle(promise: Promise, mode: SinkMode, late: () => void): Promise { + if (mode === "never" || mode === "late") await vi.advanceTimersByTimeAsync(CORE_RUN_AUDIT_EMIT_TIMEOUT_MS); + late(); return await promise; +} +function layer(candidates: bigint[]) { + const deleted = vi.fn(async () => undefined); + const db = { + select: vi.fn() + .mockReturnValueOnce({ from: () => ({ where: async () => [] }) }) + .mockReturnValueOnce({ from: () => ({ where: () => ({ orderBy: () => ({ limit: async () => candidates.map((seq) => ({ seq })) }) }) }) }) + .mockReturnValueOnce({ from: () => ({ where: () => ({ orderBy: () => ({ limit: async () => [{ seq: 99n }] }) }) }) }), + delete: vi.fn(() => ({ where: deleted })), + }; + return { projectId: "project-9180", db, deleted }; +} + +beforeEach(() => vi.useFakeTimers()); +afterEach(() => { audit.sink = undefined; vi.useRealTimers(); }); + +/* +FNXC:RunAudit 2026-08-20-06:50: +FN-9180 exercises retention through its exported production owner, not the seam alone. The bounded +await occurs after DELETE so an audit failure cannot reclassify a committed prune as a failed sweep. +*/ +describe("task lifecycle retention bounded audit health", () => { + it.each(["absent", "throw", "reject", "never", "late"] as const)("returns the committed prune result for a %s audit sink", async (mode) => { + const late = configure(mode); const fixture = layer([1n]); + const result = await settle(pruneTaskLifecycleEvents(fixture as never, "project-9180", { maxDeletes: 1 }), mode, late); + expect(result).toEqual({ prunedCount: 1, oldestRetainedSeq: 99n, minAckedSeq: null, liveConsumerCount: 0, staleConsumerCount: 0, budgetExhausted: true }); + expect(fixture.deleted).toHaveBeenCalledTimes(1); + }); + + it.each(["absent", "throw", "reject", "never", "late"] as const)("preserves the zero-prune result without a DELETE for a %s audit sink", async (mode) => { + const late = configure(mode); const fixture = layer([]); + await expect(settle(pruneTaskLifecycleEvents(fixture as never, "project-9180"), mode, late)).resolves.toEqual({ prunedCount: 0, oldestRetainedSeq: 99n, minAckedSeq: null, liveConsumerCount: 0, staleConsumerCount: 0, budgetExhausted: false }); + expect(fixture.deleted).not.toHaveBeenCalled(); + }); + + it("attempts retention telemetry only after the bounded DELETE", async () => { + const order: string[] = []; audit.sink = () => { order.push("audit"); }; + const fixture = layer([1n]); fixture.deleted.mockImplementation(async () => { order.push("delete"); }); + await expect(pruneTaskLifecycleEvents(fixture as never, "project-9180", { maxDeletes: 1 })).resolves.toBeDefined(); + expect(order).toEqual(["delete", "audit"]); + }); +}); diff --git a/packages/core/src/task-store/task-deleted-outbox-consumer.ts b/packages/core/src/task-store/task-deleted-outbox-consumer.ts index ad6808b746..d3a5b509dd 100644 --- a/packages/core/src/task-store/task-deleted-outbox-consumer.ts +++ b/packages/core/src/task-store/task-deleted-outbox-consumer.ts @@ -3,6 +3,7 @@ import { and, eq, isNull } from "drizzle-orm"; import type { TaskStore } from "../store.js"; import { createLogger } from "../process/logger.js"; import { recordRunAuditEvent } from "../postgres/data-layer.js"; +import { emitBoundedRunAudit } from "../run-audit/emit-bounded-run-audit.js"; import * as schema from "../postgres/schema/index.js"; import { acknowledgeTaskLifecycleEvent, @@ -282,7 +283,13 @@ export class TaskDeletedOutboxConsumer { } } if (this.running && events.length > 0) { - await recordRunAuditEvent(layer, { + /* + FNXC:RunAudit 2026-08-20-06:50: + FN-9178 classified task-deleted-outbox catch-up, reconciliation, and lease-fence rows as + class A: bounded best-effort telemetry. Keep each emit awaited at its durable-work boundary, + rather than fire-and-forget, so post-acknowledgement/cursor ordering remains observable. + */ + await emitBoundedRunAudit({ recordRunAuditEvent: (input) => recordRunAuditEvent(layer, input) }, { agentId: "system", runId: `task-deleted-outbox:${consumerId}`, domain: "task-lifecycle", @@ -386,7 +393,7 @@ export class TaskDeletedOutboxConsumer { } const advanced = await advanceTaskLifecycleConsumerCursor(layer, consumerId, priorSeq, headSeq, lease.fencingToken); if (!advanced) return false; - await recordRunAuditEvent(layer, { + await emitBoundedRunAudit({ recordRunAuditEvent: (input) => recordRunAuditEvent(layer, input) }, { agentId: "system", runId: `task-deleted-outbox:${consumerId}`, domain: "task-lifecycle", mutationType: "task-deleted-outbox:reconciliation-fallback", target: consumerId, metadata: { projectId: layer.projectId, consumerId, reason, reconciliationHeadSeq: headSeq.toString(), dispatchedCount, scannedCount: liveRows.length }, @@ -399,7 +406,7 @@ export class TaskDeletedOutboxConsumer { const consumerId = this.store.consumerId; if (!layer || !consumerId) return; const cursor = await readTaskLifecycleConsumerCursor(layer, consumerId); - await recordRunAuditEvent(layer, { + await emitBoundedRunAudit({ recordRunAuditEvent: (input) => recordRunAuditEvent(layer, input) }, { agentId: "system", runId: `task-deleted-outbox:${consumerId}`, domain: "task-lifecycle", mutationType: "task-deleted-outbox:lease-fenced", target: consumerId, metadata: { diff --git a/packages/core/src/task-store/task-lifecycle-event-retention.ts b/packages/core/src/task-store/task-lifecycle-event-retention.ts index 4d106918d2..dfd690af16 100644 --- a/packages/core/src/task-store/task-lifecycle-event-retention.ts +++ b/packages/core/src/task-store/task-lifecycle-event-retention.ts @@ -1,6 +1,7 @@ import { and, asc, eq, lt, lte } from "drizzle-orm"; import * as schema from "../postgres/schema/index.js"; import { recordRunAuditEvent, type AsyncDataLayer } from "../postgres/data-layer.js"; +import { emitBoundedRunAudit } from "../run-audit/emit-bounded-run-audit.js"; export const TASK_LIFECYCLE_RETENTION_DAYS = 30; export const TASK_LIFECYCLE_RETENTION_MAX_DELETES = 5_000; @@ -84,7 +85,12 @@ export async function pruneTaskLifecycleEvents( staleConsumerCount, budgetExhausted: candidates.length === maxDeletes, }; - await recordRunAuditEvent(layer, { + /* + FNXC:RunAudit 2026-08-20-06:50: + FN-9178 classifies retention-pruned as best-effort class-A telemetry. Await the bounded seam after + the committed bounded DELETE so a hostile sink cannot fail or indefinitely delay the sweep. + */ + await emitBoundedRunAudit({ recordRunAuditEvent: (input) => recordRunAuditEvent(layer, input) }, { agentId: "system", runId: "task-deleted-outbox:retention", domain: "task-lifecycle",