From 6d51ae84c378351e4359ec7708cb3c6fa68bc807 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 19 Aug 2026 23:20:27 -0700 Subject: [PATCH] FN-9176: Route remaining run-audit emitters through bounded seam Prevent audit sink failures and hangs from blocking engine lifecycle progress. - Route hold release, goal diagnostics, overseer, mesh lease, credential rotation, and workflow boundary telemetry through emitBoundedRunAudit. - Add hostile sink regression coverage and a source-level isolation ratchet for migrated emitters. - Document the bounded audit contract, scoped exclusions, and release impact. Files changed: ...n-9176-remaining-run-audit-emitter-isolation.md | 7 ++ AGENTS.md | 5 + docs/run-audit.md | 2 +- ...on-executor-run-audit-emitter-isolation.test.ts | 8 ++ ...un-audit-sink-health-boundary-heartbeat.test.ts | 116 +++++++++++++++++++++ .../__tests__/run-audit-sink-health-goals.test.ts | 71 +++++++++++++ .../run-audit-sink-health-hold-release.test.ts | 102 ++++++++++++++++++ .../run-audit-sink-health-overseer.test.ts | 71 +++++++++++++ .../run-audit-sink-health-runtime-leases.test.ts | 82 +++++++++++++++ .../engine/src/credential-instance-rotation.ts | 6 +- packages/engine/src/execution/hold-release.ts | 32 +++--- packages/engine/src/goals/goal-anchoring-audit.ts | 9 +- .../engine/src/goals/goal-injection-diagnostics.ts | 5 +- .../src/overseer/overseer-advisor-service.ts | 11 +- packages/engine/src/project-engine.ts | 10 +- packages/engine/src/project/mesh-lease-manager.ts | 17 +-- packages/engine/src/runtimes/in-process-runtime.ts | 32 ++++-- .../engine/src/workflow-column-boundary-hooks.ts | 3 +- 18 files changed, 524 insertions(+), 65 deletions(-) Fusion-Task-Id: FN-9176 Fusion-Task-Lineage: 5205deeb-57e8-424e-aba4-3aed6a66f85c Co-authored-by: Fusion (runfusion.ai) --- ...6-remaining-run-audit-emitter-isolation.md | 7 ++ AGENTS.md | 5 + docs/run-audit.md | 2 +- ...ecutor-run-audit-emitter-isolation.test.ts | 8 ++ ...dit-sink-health-boundary-heartbeat.test.ts | 116 ++++++++++++++++++ .../run-audit-sink-health-goals.test.ts | 71 +++++++++++ ...run-audit-sink-health-hold-release.test.ts | 102 +++++++++++++++ .../run-audit-sink-health-overseer.test.ts | 71 +++++++++++ ...n-audit-sink-health-runtime-leases.test.ts | 82 +++++++++++++ .../src/credential-instance-rotation.ts | 6 +- packages/engine/src/execution/hold-release.ts | 32 ++--- .../engine/src/goals/goal-anchoring-audit.ts | 9 +- .../src/goals/goal-injection-diagnostics.ts | 5 +- .../src/overseer/overseer-advisor-service.ts | 11 +- packages/engine/src/project-engine.ts | 10 +- .../engine/src/project/mesh-lease-manager.ts | 17 +-- .../engine/src/runtimes/in-process-runtime.ts | 32 +++-- .../src/workflow-column-boundary-hooks.ts | 3 +- 18 files changed, 524 insertions(+), 65 deletions(-) create mode 100644 .changeset/fn-9176-remaining-run-audit-emitter-isolation.md create mode 100644 packages/engine/src/__tests__/run-audit-sink-health-boundary-heartbeat.test.ts create mode 100644 packages/engine/src/__tests__/run-audit-sink-health-goals.test.ts create mode 100644 packages/engine/src/__tests__/run-audit-sink-health-hold-release.test.ts create mode 100644 packages/engine/src/__tests__/run-audit-sink-health-overseer.test.ts create mode 100644 packages/engine/src/__tests__/run-audit-sink-health-runtime-leases.test.ts diff --git a/.changeset/fn-9176-remaining-run-audit-emitter-isolation.md b/.changeset/fn-9176-remaining-run-audit-emitter-isolation.md new file mode 100644 index 0000000000..da850c72f4 --- /dev/null +++ b/.changeset/fn-9176-remaining-run-audit-emitter-isolation.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Keep remaining engine audit telemetry from blocking workflow progress. +category: internal +dev: Routes hold-release, goals, overseer, mesh lease, runtime rotation, and column boundaries through packages/engine/src/util/emit-bounded-run-audit.ts while preserving the goal retrieval log anchor. diff --git a/AGENTS.md b/AGENTS.md index 409aefb316..b1bb1945e6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -285,6 +285,11 @@ FNXC:RunAudit 2026-08-20-04:15: FN-9175 promotes this to every engine lane. Direct engine `store.recordRunAuditEvent` calls are an anti-pattern: use `packages/engine/src/util/emit-bounded-run-audit.ts` and prove hostile sink isolation through a behavioral regression. + +FNXC:RunAudit 2026-08-20-05:39: +FN-9176 applies that seam to hold-release, goals, overseer, mesh leases, runtime credential +rotation, and workflow-column boundaries. The bespoke merge-write fence and packages/core +canonical emitters remain explicit exclusions until their separately scoped hardening work lands. --> - 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. - 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. diff --git a/docs/run-audit.md b/docs/run-audit.md index 2c53079303..b7e5a39684 100644 --- a/docs/run-audit.md +++ b/docs/run-audit.md @@ -78,4 +78,4 @@ Adding a new catalogued run-audit event requires updating **both** the typed cat All engine telemetry must use `emitBoundedRunAudit` from `packages/engine/src/util/emit-bounded-run-audit.ts`. It is best-effort and never load-bearing for lifecycle correctness: absent/non-function, synchronously throwing, rejecting, never-settling, and late-settling sinks are absorbed without altering the owning branch. The seam swallow-logs and bounds each write; it intentionally adds no retry, backoff, or queueing. -This applies to executor, run-auditor, self-healing, merger, PR reconciliation, scheduler, project-engine, plugin, and mission-loop emitters. `merge-write-fence.ts` retains its bespoke non-`RunAuditEventInput` recorder; deferred emitters outside those modules remain separate follow-up work. New engine emitters must ship with a behavioral sink-health regression covering hostile sink states, not only a source-routing assertion. \ No newline at end of file +This applies to executor, run-auditor, self-healing, merger, PR reconciliation, scheduler, project-engine, plugin, mission-loop, hold-release, goal diagnostics, overseer advisor, mesh-lease, in-process runtime, credential rotation, and workflow-column-boundary emitters. `packages/engine/src/merge/merge-write-fence.ts` retains its bespoke non-`RunAuditEventInput` recorder; `packages/core` emitters, including `recordPlannerIntervention`, remain separately scoped. New engine emitters must ship with a behavioral sink-health regression covering hostile sink states, not only a source-routing assertion. \ No newline at end of file diff --git a/packages/engine/src/__tests__/non-executor-run-audit-emitter-isolation.test.ts b/packages/engine/src/__tests__/non-executor-run-audit-emitter-isolation.test.ts index c19ee4533e..3c51f2c979 100644 --- a/packages/engine/src/__tests__/non-executor-run-audit-emitter-isolation.test.ts +++ b/packages/engine/src/__tests__/non-executor-run-audit-emitter-isolation.test.ts @@ -12,6 +12,14 @@ const modules = [ "project-engine.ts", "plugins/plugin-runner.ts", "missions/mission-execution-loop.ts", + "execution/hold-release.ts", + "goals/goal-injection-diagnostics.ts", + "goals/goal-anchoring-audit.ts", + "overseer/overseer-advisor-service.ts", + "project/mesh-lease-manager.ts", + "runtimes/in-process-runtime.ts", + "credential-instance-rotation.ts", + "workflow-column-boundary-hooks.ts", ] as const; /** diff --git a/packages/engine/src/__tests__/run-audit-sink-health-boundary-heartbeat.test.ts b/packages/engine/src/__tests__/run-audit-sink-health-boundary-heartbeat.test.ts new file mode 100644 index 0000000000..d7d16784bd --- /dev/null +++ b/packages/engine/src/__tests__/run-audit-sink-health-boundary-heartbeat.test.ts @@ -0,0 +1,116 @@ +import { describe, expect, it, vi } from "vitest"; +import type { TaskStore } from "@fusion/core"; +import { RUN_AUDIT_EMIT_TIMEOUT_MS } from "../util/emit-bounded-run-audit.js"; +import { createExecutorColumnBoundaryHooks } from "../workflow-column-boundary-hooks.js"; +import { HeartbeatMonitor } from "../agent-heartbeat.js"; +import type { Agent, AgentHeartbeatRun, AgentStore } from "@fusion/core"; +import { createBudgetStatus } from "./heartbeat-test-helpers.js"; + +vi.mock("../logger.js", async () => { + const { createMockLogger, formatMockError } = await import("./heartbeat-test-helpers.js"); + return { createLogger: vi.fn(() => createMockLogger()), heartbeatLog: createMockLogger(), formatError: formatMockError }; +}); + +vi.mock("../pi.js", () => ({ + createFnAgent: vi.fn(), + describeModel: vi.fn(() => "mock/mock"), + promptWithFallback: vi.fn(async (session: { prompt: (value: string) => Promise }, value: string) => session.prompt(value)), +})); + +type Sink = undefined | (() => unknown); +const hostileSinks: [string, Sink][] = [ + ["absent", undefined], ["throws", () => { throw new Error("down"); }], + ["rejects", () => Promise.reject(new Error("down"))], ["hangs", () => new Promise(() => {})], + ["late", () => new Promise(() => {})], +]; + +const baseStore = (recordRunAuditEvent: Sink) => ({ + recordRunAuditEvent, + getSettings: vi.fn(async () => ({})), + listWorkflowWorkItemsForTask: vi.fn(async () => []), + moveTask: vi.fn(async () => undefined), +} as unknown as TaskStore); + +function hooks(sink: Sink) { + return createExecutorColumnBoundaryHooks({ store: baseStore(sink), task: { id: "FN-1" } }); +} + +function heartbeatStore(sink: Sink): AgentStore { + const agent = { + id: "agent-1", name: "Recovering", role: "executor", state: "error", lastError: "socket hang up", + metadata: {}, runtimeConfig: { enabled: true }, createdAt: "2026-01-01T00:00:00.000Z", updatedAt: "2026-01-01T00:00:00.000Z", + } as Agent; + const runs = new Map(); + return { + getAgent: vi.fn(async () => agent), getBudgetStatus: vi.fn(async () => createBudgetStatus({ agentId: agent.id })), + startHeartbeatRun: vi.fn(async () => { + const run = { id: "heartbeat-1", agentId: agent.id, source: "on_demand", startedAt: new Date().toISOString(), endedAt: null, status: "active" } as AgentHeartbeatRun; + runs.set(run.id, run); + return run; + }), + saveRun: vi.fn(async (run: AgentHeartbeatRun) => { runs.set(run.id, run); }), + getRunDetail: vi.fn(async (_agentId: string, runId: string) => runs.get(runId)!), + endHeartbeatRun: vi.fn(async () => undefined), recordHeartbeat: vi.fn(async () => undefined), + updateAgentState: vi.fn(async () => undefined), updateAgent: vi.fn(async () => undefined), + getRatingSummary: vi.fn(async () => undefined), appendRunLog: vi.fn(async () => undefined), + getLastBlockedState: vi.fn(async () => null), setLastBlockedState: vi.fn(async () => undefined), clearLastBlockedState: vi.fn(async () => undefined), + } as unknown as AgentStore; +} + +function heartbeatTaskStore(sink: Sink): TaskStore { + return { + recordRunAuditEvent: sink, getSettings: vi.fn(async () => ({})), selectNextTaskForAgent: vi.fn(async () => null), + listTasks: vi.fn(async () => []), getTaskDocuments: vi.fn(async () => []), + } as unknown as TaskStore; +} + +/* +FNXC:RunAudit 2026-08-20-06:01: +Column-boundary telemetry is awaited by workflow ownership code, so both transition shapes must +complete when the optional sink is absent, hostile, stalled, or settles after the bound. +*/ +describe("column-boundary and heartbeat run-audit sink health", () => { + it.each(hostileSinks)("does not abort either production boundary event when sink %s", async (name, sink) => { + if (name === "hangs" || name === "late") vi.useFakeTimers(); + try { + const audit = sink ? vi.fn(sink) : undefined; + const boundary = hooks(audit); + const transition = boundary.emitAudit({ type: "task:column-transition", taskId: "FN-1", workflowId: "wf", fromColumn: "todo", toColumn: "in-progress", nodeId: "execute", irHash: "hash" } as any); + const pinned = boundary.emitAudit({ type: "task:workflow-node-pinned", taskId: "FN-1", workflowId: "wf", pinnedNodeId: "execute", reason: "entry" } as any); + if (name === "hangs" || name === "late") { + for (let turn = 0; turn < 3; turn += 1) await vi.advanceTimersByTimeAsync(RUN_AUDIT_EMIT_TIMEOUT_MS + 1); + } + await expect(Promise.all([transition, pinned])).resolves.toEqual([undefined, undefined]); + if (audit) { + expect(audit).toHaveBeenCalledWith(expect.objectContaining({ mutationType: "task:column-transition", metadata: expect.objectContaining({ fromColumn: "todo", toColumn: "in-progress" }) })); + expect(audit).toHaveBeenCalledWith(expect.objectContaining({ mutationType: "task:workflow-node-pinned", metadata: { taskId: "FN-1", workflowId: "wf", pinnedNodeId: "execute", reason: "entry" } })); + } + } finally { vi.useRealTimers(); } + }); + + it.each(hostileSinks)("completes a real heartbeat recovery run when the auditor sink %s", async (name, sink) => { + if (name === "hangs" || name === "late") vi.useFakeTimers(); + try { + const audit = sink ? vi.fn(sink) : undefined; + const taskStore = heartbeatTaskStore(audit); + const monitor = new HeartbeatMonitor({ store: heartbeatStore(audit), taskStore, rootDir: process.cwd() }); + const pending = monitor.executeHeartbeat({ agentId: "agent-1", source: "on_demand" }); + if (name === "hangs" || name === "late") await vi.advanceTimersByTimeAsync(RUN_AUDIT_EMIT_TIMEOUT_MS); + await expect(pending).resolves.toMatchObject({ status: "completed", agentId: "agent-1" }); + if (audit) expect(audit).toHaveBeenCalledWith(expect.objectContaining({ mutationType: "agent:auto-recover-error-state", target: "agent-1" })); + } finally { vi.useRealTimers(); } + }); + + it("pre-observes a late boundary rejection", async () => { + let reject!: (error: Error) => void; + const late = new Promise((_resolve, rejectPromise) => { reject = rejectPromise; }); + const unhandled = vi.fn(); + process.on("unhandledRejection", unhandled); + try { + await hooks(() => late).emitAudit({ type: "task:column-transition", taskId: "FN-1", workflowId: "wf", fromColumn: "todo", toColumn: "in-progress", nodeId: "execute", irHash: "hash" } as any); + reject(new Error("late")); + await Promise.resolve(); + expect(unhandled).not.toHaveBeenCalled(); + } finally { process.off("unhandledRejection", unhandled); } + }); +}); diff --git a/packages/engine/src/__tests__/run-audit-sink-health-goals.test.ts b/packages/engine/src/__tests__/run-audit-sink-health-goals.test.ts new file mode 100644 index 0000000000..0dfb1e6306 --- /dev/null +++ b/packages/engine/src/__tests__/run-audit-sink-health-goals.test.ts @@ -0,0 +1,71 @@ +import { describe, expect, it, vi } from "vitest"; +import type { TaskStore } from "@fusion/core"; +import { RUN_AUDIT_EMIT_TIMEOUT_MS } from "../util/emit-bounded-run-audit.js"; +import { emitGoalInjectionDiagnostic, resolveAndEmitGoalContext } from "../goals/goal-injection-diagnostics.js"; +import { emitGoalRetrievalAudit } from "../goals/goal-anchoring-audit.js"; + +type Sink = undefined | (() => unknown); +const sinks: [string, Sink][] = [ + ["absent", undefined], + ["throws", () => { throw new Error("sink down"); }], + ["rejects", () => Promise.reject(new Error("sink down"))], + ["hangs", () => new Promise(() => {})], + ["late", () => new Promise(() => {})], +]; + +async function diagnostic(sink: Sink) { + const store = { recordRunAuditEvent: sink } as unknown as TaskStore; + return emitGoalInjectionDiagnostic({ + lane: "heartbeat", outcome: "applied", goalCount: 1, goalIds: ["G-1"], truncated: false, + store, runContext: { runId: "run", agentId: "agent", taskId: "FN-1", phase: "heartbeat" }, + }); +} + +describe("goal run-audit sink health", () => { + it.each(sinks)("preserves the diagnostic through an %s audit sink", async (name, sink) => { + if (name === "hangs") vi.useFakeTimers(); + try { + const pending = diagnostic(sink); + if (name === "hangs") await vi.advanceTimersByTimeAsync(RUN_AUDIT_EMIT_TIMEOUT_MS); + await expect(pending).resolves.toMatchObject({ lane: "heartbeat", goalIds: ["G-1"] }); + } finally { vi.useRealTimers(); } + }); + + it.each(sinks)("keeps retrieval synchronous when sink %s", async (_name, sink) => { + const store = { recordRunAuditEvent: sink } as unknown as TaskStore; + expect(() => emitGoalRetrievalAudit(store, { runId: "run", agentId: "agent", taskId: "FN-1" }, { toolName: "fn_goal_list", resultCount: 1, goalIds: ["G-1"] })).not.toThrow(); + }); + + it.each(sinks)("preserves resolveAndEmitGoalContext results when diagnostic sink %s", async (name, sink) => { + if (name === "hangs" || name === "late") vi.useFakeTimers(); + try { + const store = { + recordRunAuditEvent: sink, + getGoalStore: () => ({ listGoals: async () => [] }), + } as unknown as TaskStore; + const pending = resolveAndEmitGoalContext({ + lane: "heartbeat", store, audit: { database: vi.fn(async () => undefined) } as any, + runContext: { runId: "run", agentId: "agent", phase: "heartbeat" }, + }); + if (name === "hangs" || name === "late") await vi.advanceTimersByTimeAsync(RUN_AUDIT_EMIT_TIMEOUT_MS); + await expect(pending).resolves.toMatchObject({ goalContext: "", classification: { outcome: "no-goals" } }); + } finally { vi.useRealTimers(); } + }); + + it("keeps retrieval synchronous and observes a late rejection", async () => { + vi.useFakeTimers(); + const unhandled = vi.fn(); + process.on("unhandledRejection", unhandled); + let reject!: (error: Error) => void; + const late = new Promise((_resolve, rejectPromise) => { reject = rejectPromise; }); + const sink = vi.fn(() => late); + try { + expect(() => emitGoalRetrievalAudit({ recordRunAuditEvent: sink } as unknown as TaskStore, { runId: "run", agentId: "agent", taskId: "FN-1" }, { toolName: "fn_goal_list", resultCount: 1, goalIds: ["G-1"] })).not.toThrow(); + expect(sink).toHaveBeenCalledWith(expect.objectContaining({ mutationType: "goal:retrieval-invoked", metadata: expect.objectContaining({ goalIds: ["G-1"] }) })); + await vi.advanceTimersByTimeAsync(RUN_AUDIT_EMIT_TIMEOUT_MS); + reject(new Error("late")); + await Promise.resolve(); + expect(unhandled).not.toHaveBeenCalled(); + } finally { process.off("unhandledRejection", unhandled); vi.useRealTimers(); } + }); +}); diff --git a/packages/engine/src/__tests__/run-audit-sink-health-hold-release.test.ts b/packages/engine/src/__tests__/run-audit-sink-health-hold-release.test.ts new file mode 100644 index 0000000000..2376564adc --- /dev/null +++ b/packages/engine/src/__tests__/run-audit-sink-health-hold-release.test.ts @@ -0,0 +1,102 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Task, TaskStore, WorkflowIr } from "@fusion/core"; +import { RUN_AUDIT_EMIT_TIMEOUT_MS } from "../util/emit-bounded-run-audit.js"; +import { promoteHeldTask, releaseHeldTaskByEvent, runHoldReleaseSweep } from "../execution/hold-release.js"; + +type Sink = undefined | (() => unknown); +const hostileSinks: [string, Sink][] = [ + ["absent", undefined], + ["throws", () => { throw new Error("sink down"); }], + ["rejects", () => Promise.reject(new Error("sink down"))], + ["hangs", () => new Promise(() => {})], +]; + +function task(overrides: Partial): Task { + return { + id: "FN-1", title: "held", description: "", column: "todo", status: null, + dependencies: [], steps: [], currentStep: 0, log: [], + createdAt: "2026-01-01T00:00:00.000Z", updatedAt: "2026-01-01T00:00:00.000Z", + columnMovedAt: "2026-01-01T00:00:00.000Z", ...overrides, + } as Task; +} + +function workflow(release: "dependency" | "capacity" | "external-event"): WorkflowIr { + return { + version: "v2", id: "wf", nodes: [], edges: [], columns: [ + { id: "todo", label: "Todo", traits: [{ trait: "hold", config: { release } }] }, + { id: "in-progress", label: "In progress", traits: [{ trait: "wip" }] }, + // Deliberately omit complete so legacy `done` and workflow completion disagree. + { id: "done", label: "Done", traits: [] }, + ], + } as unknown as WorkflowIr; +} + +function storeFor(tasks: Task[], ir: WorkflowIr, recordRunAuditEvent: Sink) { + const byId = new Map(tasks.map((item) => [item.id, item])); + return { + getSettings: vi.fn(async () => ({})), listTasks: vi.fn(async () => tasks), + getTask: vi.fn(async (id: string) => byId.get(id)), + getTaskWorkflowSelectionAsync: vi.fn(async () => ({ workflowId: "wf", stepIds: [] })), + getWorkflowDefinition: vi.fn(async () => ({ ir })), + getCompletionHandoffAcceptedMarker: vi.fn(async () => null), + updateTask: vi.fn(async () => undefined), + moveTaskIf: vi.fn(async () => ({ moved: true })), + recordRunAuditEvent, + } as unknown as TaskStore; +} + +async function settleAwaited(pending: Promise, name: string) { + if (name === "hangs") await vi.advanceTimersByTimeAsync(RUN_AUDIT_EMIT_TIMEOUT_MS); + await expect(pending).resolves.toMatchObject({ released: true, toColumn: "in-progress" }); +} + +describe("hold-release run-audit sink health", () => { + it.each(hostileSinks)("keeps the real dependency-parity sweep result unchanged when audit sink %s", async (_name, sink) => { + const held = task({ id: "FN-1", dependencies: ["FN-2"] }); + const completedLegacyOnly = task({ id: "FN-2", column: "done" }); + const audit = sink ? vi.fn(sink) : undefined; + const result = await runHoldReleaseSweep(storeFor([held, completedLegacyOnly], workflow("dependency"), audit), { now: () => 1 }); + expect(result.released).toEqual(["FN-1"]); + if (audit) expect(audit).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "merge:dependency-parity-diff", taskId: "FN-2", + metadata: { depId: "FN-2", completeFlagResult: false, legacyResult: true, source: "hold-release.dependency" }, + })); + }); + + it.each(hostileSinks)("keeps force-promote observable results unchanged when audit sink %s", async (name, sink) => { + if (name === "hangs") vi.useFakeTimers(); + try { + const held = task({ status: "needs-replan" }); + const audit = sink ? vi.fn(sink) : undefined; + const store = storeFor([held], workflow("capacity"), audit); + await settleAwaited(promoteHeldTask(store, held.id, {}, { force: true }), name); + if (audit) expect(audit).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:promote-forced-unplanned", taskId: "FN-1", agentId: "system", runId: "promote-force-FN-1", + })); + } finally { vi.useRealTimers(); } + }); + + it.each(hostileSinks)("keeps event release observable results unchanged when audit sink %s", async (_name, sink) => { + const held = task({}); + const audit = sink ? vi.fn(sink) : undefined; + const result = await releaseHeldTaskByEvent(storeFor([held], workflow("external-event"), audit), held.id, "webhook"); + expect(result).toEqual({ released: true, toColumn: "in-progress" }); + if (audit) expect(audit).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:hold-release-event", taskId: "FN-1", agentId: "scheduler", runId: "hold-release:event:FN-1", + })); + }); + + it("pre-observes a late fire-and-forget rejection", async () => { + let reject!: (error: Error) => void; + const late = new Promise((_resolve, rejectPromise) => { reject = rejectPromise; }); + const unhandled = vi.fn(); + process.on("unhandledRejection", unhandled); + try { + const held = task({}); + await releaseHeldTaskByEvent(storeFor([held], workflow("external-event"), () => late), held.id, "webhook"); + reject(new Error("late sink failure")); + await Promise.resolve(); + expect(unhandled).not.toHaveBeenCalled(); + } finally { process.off("unhandledRejection", unhandled); } + }); +}); diff --git a/packages/engine/src/__tests__/run-audit-sink-health-overseer.test.ts b/packages/engine/src/__tests__/run-audit-sink-health-overseer.test.ts new file mode 100644 index 0000000000..3f4f05e0de --- /dev/null +++ b/packages/engine/src/__tests__/run-audit-sink-health-overseer.test.ts @@ -0,0 +1,71 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Task } from "@fusion/core"; +import { OverseerAdvisorService, createParsingOverseerAgent } from "../overseer/overseer-advisor-service.js"; +import { ProjectEngine } from "../project-engine.js"; +import { RUN_AUDIT_EMIT_TIMEOUT_MS } from "../util/emit-bounded-run-audit.js"; + +const task = { id: "FN-1", title: "t", column: "in-progress", status: null, createdAt: "2026-01-01T00:00:00.000Z", updatedAt: "2026-01-01T00:00:00.000Z" } as Task; + +describe("overseer run-audit sink health", () => { + it.each([ + ["absent", undefined], ["throws", () => { throw new Error("down"); }], + ["rejects", () => Promise.reject(new Error("down"))], ["hangs", () => new Promise(() => {})], + ["late", () => new Promise(() => {})], + ])("delivers advice when the steering sink %s", async (_name, recordRunAuditEvent) => { + const addSteeringComment = vi.fn(async () => ({})); + const service = new OverseerAdvisorService({ + store: { addSteeringComment, recordRunAuditEvent, getRunAuditEvents: () => [], getTask: async () => task }, + resolveLevel: () => "autonomous", + resolveModel: () => ({ provider: "mock", modelId: "scripted" }), + agentFactory: async ({ systemPrompt, onAdvice }) => createParsingOverseerAgent({ systemPrompt, onAdvice, complete: async () => JSON.stringify({ note: "Check scope", severity: "concern" }) }), + }); + expect(await service.ensureTask(task)).toBe(true); + await service.onExecutorLogDelta(task.id, [{ type: "text", text: "editing source", agent: "executor" }]); + await vi.waitFor(() => expect(addSteeringComment).toHaveBeenCalled()); + }); + + it.each([ + ["absent", undefined], ["throws", () => { throw new Error("down"); }], + ["rejects", () => Promise.reject(new Error("down"))], ["hangs", () => new Promise(() => {})], + ["late", () => new Promise(() => {})], + ])("completes the ProjectEngine escalation caller when the facade sink %s", async (name, recordRunAuditEvent) => { + if (name === "hangs" || name === "late") vi.useFakeTimers(); + try { + // This prototype entry is the exact private ProjectEngine poll helper without booting unrelated runtime subsystems. + const engine = Object.create(ProjectEngine.prototype) as any; + engine.plannerEscalationEmitDedup = new Set(); + const audit = recordRunAuditEvent ? vi.fn(recordRunAuditEvent) : undefined; + const store = { recordRunAuditEvent: audit, getRunAuditEvents: () => [] } as any; + const pending = engine.emitOverseerEscalationDeduped(store, "FN-1", { + watchedStage: "executor", reason: "budget exhausted", attemptCount: 1, attemptLimit: 1, sourceLinks: [], + }); + if (name === "hangs" || name === "late") await vi.advanceTimersByTimeAsync(RUN_AUDIT_EMIT_TIMEOUT_MS); + await expect(pending).resolves.toBeUndefined(); + if (audit) expect(audit).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "overseer:intervention", taskId: "FN-1", metadata: expect.objectContaining({ action: "escalate", outcome: "failed" }), + })); + } finally { vi.useRealTimers(); } + }); + + it("pre-observes a late steering rejection after advice delivery", async () => { + let reject!: (error: Error) => void; + const late = new Promise((_resolve, rejectPromise) => { reject = rejectPromise; }); + const unhandled = vi.fn(); + process.on("unhandledRejection", unhandled); + try { + const addSteeringComment = vi.fn(async () => ({})); + const service = new OverseerAdvisorService({ + store: { addSteeringComment, recordRunAuditEvent: () => late, getRunAuditEvents: () => [], getTask: async () => task }, + resolveLevel: () => "autonomous", + resolveModel: () => ({ provider: "mock", modelId: "scripted" }), + agentFactory: async ({ systemPrompt, onAdvice }) => createParsingOverseerAgent({ systemPrompt, onAdvice, complete: async () => JSON.stringify({ note: "Check scope", severity: "concern" }) }), + }); + await service.ensureTask(task); + await service.onExecutorLogDelta(task.id, [{ type: "text", text: "editing source", agent: "executor" }]); + await vi.waitFor(() => expect(addSteeringComment).toHaveBeenCalled()); + reject(new Error("late")); + await Promise.resolve(); + expect(unhandled).not.toHaveBeenCalled(); + } finally { process.off("unhandledRejection", unhandled); } + }); +}); diff --git a/packages/engine/src/__tests__/run-audit-sink-health-runtime-leases.test.ts b/packages/engine/src/__tests__/run-audit-sink-health-runtime-leases.test.ts new file mode 100644 index 0000000000..c998553b13 --- /dev/null +++ b/packages/engine/src/__tests__/run-audit-sink-health-runtime-leases.test.ts @@ -0,0 +1,82 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Task, TaskStore } from "@fusion/core"; +import { CredentialInstanceRotator } from "../credential-instance-rotation.js"; +import { MeshLeaseManager } from "../project/mesh-lease-manager.js"; +import { RUN_AUDIT_EMIT_TIMEOUT_MS } from "../util/emit-bounded-run-audit.js"; +import { createRuntimeCredentialRotationAuditAdapter } from "../runtimes/in-process-runtime.js"; + +const refs = [{ providerId: "anthropic", instanceId: "a" }, { providerId: "anthropic", instanceId: "b" }]; +type Sink = undefined | ((type: string, metadata: Record) => unknown); +const sinks: [string, Sink][] = [ + ["absent", undefined], ["throws", () => { throw new Error("down"); }], + ["rejects", () => Promise.reject(new Error("down"))], ["hangs", () => new Promise(() => {})], + ["late", () => new Promise(() => {})], +]; + +describe("rotation run-audit sink health", () => { + it.each(sinks)("does not change candidate ordering when the sink %s", async (_name, recordRunAuditEvent) => { + const rotator = new CredentialInstanceRotator({ + instanceSource: { listInstances: () => refs, getDefaultInstance: () => refs[0] }, + recordRunAuditEvent, + }); + const event = await rotator.beginEvent({ providerId: "anthropic", startingInstanceId: "a", lane: "executor-step", taskId: "FN-1" }); + expect((await event?.next())?.instanceId).toBe("b"); + event?.recordOutcome("rotation-succeeded"); + event?.finishExhausted(); + expect(await event?.next()).toBeUndefined(); + }); + + it.each(sinks)("keeps the real mesh lease-recovery outcome unchanged when sink %s", async (name, recordRunAuditEvent) => { + if (name === "hangs" || name === "late") vi.useFakeTimers(); + try { + const current = { + id: "FN-1", description: "lease", column: "todo", dependencies: [], steps: [], currentStep: 0, log: [], + createdAt: "2026-05-01T00:00:00.000Z", updatedAt: "2026-05-01T00:00:00.000Z", + checkedOutBy: "agent-1", checkoutNodeId: "node-a", checkoutLeaseEpoch: 1, + } as Task; + const writer = recordRunAuditEvent ? vi.fn(recordRunAuditEvent) : undefined; + const store = { + getTask: vi.fn(async () => current), updateTask: vi.fn(async () => current), moveTask: vi.fn(async () => current), + logEntry: vi.fn(async () => undefined), recordRunAuditEvent: writer, + } as unknown as TaskStore; + const manager = new MeshLeaseManager({ taskStore: store, nodeHealthMonitor: { getNodeHealth: () => "offline" } as any, getHandoffPolicy: async () => "reassign-any-healthy", localNodeId: "local" }); + const pending = manager.recoverAbandonedLease("FN-1", "stale heartbeat"); + if (name === "hangs" || name === "late") { + for (let turn = 0; turn < 4; turn += 1) await vi.advanceTimersByTimeAsync(RUN_AUDIT_EMIT_TIMEOUT_MS + 1); + } + await expect(pending).resolves.toBe(true); + if (writer) expect(writer).toHaveBeenCalledWith(expect.objectContaining({ mutationType: "node:lease:recovered", taskId: "FN-1", agentId: "mesh-lease-manager" })); + } finally { vi.useRealTimers(); } + }); + + it.each(sinks)("keeps the runtime-owned rotation adapter bounded when sink %s", async (name, sink) => { + if (name === "hangs" || name === "late") vi.useFakeTimers(); + try { + const recordRunAuditEvent = sink ? vi.fn(sink) : undefined; + const adapter = createRuntimeCredentialRotationAuditAdapter({ recordRunAuditEvent } as unknown as TaskStore); + const pending = adapter("credential:instance-rotation-attempt", { taskId: "FN-1", providerId: "anthropic", toInstanceId: "b", attempt: 1 }); + if (name === "hangs" || name === "late") await vi.advanceTimersByTimeAsync(RUN_AUDIT_EMIT_TIMEOUT_MS); + await expect(pending).resolves.toBeUndefined(); + if (recordRunAuditEvent) expect(recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + taskId: "FN-1", agentId: "runtime", mutationType: "credential:instance-rotation-attempt", target: "anthropic", + })); + } finally { vi.useRealTimers(); } + }); + + it("preserves the healthy attempt payload and observes late rejection", async () => { + let reject!: (error: Error) => void; + const late = new Promise((_resolve, rejectPromise) => { reject = rejectPromise; }); + const audit = vi.fn(() => late); + const unhandled = vi.fn(); + process.on("unhandledRejection", unhandled); + try { + const rotator = new CredentialInstanceRotator({ instanceSource: { listInstances: () => refs, getDefaultInstance: () => refs[0] }, recordRunAuditEvent: audit }); + const event = await rotator.beginEvent({ providerId: "anthropic", startingInstanceId: "a", lane: "executor-agent", taskId: "FN-1" }); + await event?.next(); + expect(audit).toHaveBeenCalledWith("credential:instance-rotation-attempt", expect.objectContaining({ toInstanceId: "b", attempt: 1, taskId: "FN-1" })); + reject(new Error("late")); + await Promise.resolve(); + expect(unhandled).not.toHaveBeenCalled(); + } finally { process.off("unhandledRejection", unhandled); } + }); +}); diff --git a/packages/engine/src/credential-instance-rotation.ts b/packages/engine/src/credential-instance-rotation.ts index 1b62562583..098f2cf259 100644 --- a/packages/engine/src/credential-instance-rotation.ts +++ b/packages/engine/src/credential-instance-rotation.ts @@ -3,6 +3,7 @@ import { parseProviderInstanceKey, type ProviderInstanceRef, } from "@fusion/core"; +import { emitBoundedRunAudit } from "./util/emit-bounded-run-audit.js"; export const CREDENTIAL_INSTANCE_COOLDOWN_MS = 15 * 60_000; @@ -97,7 +98,10 @@ export class CredentialInstanceRotator { let exhausted = false; const attempted: ProviderInstanceRef[] = []; const audit = (type: string, metadata: Record) => { - try { void Promise.resolve(this.options.recordRunAuditEvent?.(type, metadata)).catch(() => undefined); } catch { /* audit is non-fatal */ } + void emitBoundedRunAudit( + { recordRunAuditEvent: () => this.options.recordRunAuditEvent?.(type, metadata) }, + { mutationType: type }, + ); }; const common = { providerId: input.providerId, diff --git a/packages/engine/src/execution/hold-release.ts b/packages/engine/src/execution/hold-release.ts index 9b3e4b0872..0665137d7e 100644 --- a/packages/engine/src/execution/hold-release.ts +++ b/packages/engine/src/execution/hold-release.ts @@ -66,6 +66,7 @@ import { import { createHash } from "node:crypto"; import { readFile } from "node:fs/promises"; import { schedulerLog } from "../logger.js"; +import { emitBoundedRunAudit } from "../util/emit-bounded-run-audit.js"; import { getPromptPath } from "./spec-staleness.js"; import { activeSessionRegistry, executingTaskLock } from "../agents/active-session-registry.js"; import { evaluateStrandedHoldContinuation } from "../plan-review-continuation.js"; @@ -410,8 +411,12 @@ async function dependencySatisfied(ctx: SweepCtx, dep: Task): Promise log.debug("goal retrieval audit emission skipped", message) } }); } diff --git a/packages/engine/src/goals/goal-injection-diagnostics.ts b/packages/engine/src/goals/goal-injection-diagnostics.ts index c420a61e33..1fffbdbba6 100644 --- a/packages/engine/src/goals/goal-injection-diagnostics.ts +++ b/packages/engine/src/goals/goal-injection-diagnostics.ts @@ -6,6 +6,7 @@ import { emitGoalAnchoringAudit } from "./goal-anchoring-audit.js"; import type { EngineRunContext } from "../util/run-audit.js"; import type { RunAuditor } from "../util/run-audit.js"; import { createLogger } from "../logger.js"; +import { emitBoundedRunAudit } from "../util/emit-bounded-run-audit.js"; const diagnosticsLog = createLogger("goal-injection-diagnostics"); @@ -256,7 +257,7 @@ export async function emitGoalInjectionDiagnostic( } try { - await auditStore.recordRunAuditEvent({ + await emitBoundedRunAudit(auditStore, { taskId: input.runContext.taskId, agentId: input.runContext.agentId, runId: input.runContext.runId, @@ -276,7 +277,7 @@ export async function emitGoalInjectionDiagnostic( ...(record.agentId ? { agentId: record.agentId } : {}), ...(record.taskId ? { taskId: record.taskId } : {}), }, - }); + }, { log: diagnosticsLog }); } catch (error) { diagnosticsLog.warn( `failed to append goal-injection run-audit event for lane=${record.lane}: ${error instanceof Error ? error.message : String(error)}`, diff --git a/packages/engine/src/overseer/overseer-advisor-service.ts b/packages/engine/src/overseer/overseer-advisor-service.ts index 941e324bd5..06113cf9e3 100644 --- a/packages/engine/src/overseer/overseer-advisor-service.ts +++ b/packages/engine/src/overseer/overseer-advisor-service.ts @@ -16,6 +16,7 @@ import { type Settings, } from "@fusion/core"; import { createLogger } from "../logger.js"; +import { emitBoundedRunAudit } from "../util/emit-bounded-run-audit.js"; import { evaluateOverseerHumanControl } from "./overseer-human-control-policy.js"; import { OverseerAdvisorRuntime, @@ -408,8 +409,8 @@ export class OverseerAdvisorService { outcome: "pending" | "succeeded", ): void { if (!this.store.recordRunAuditEvent || !this.store.getRunAuditEvents) return; - try { - emitOverseerSteering({ + void emitBoundedRunAudit({ + recordRunAuditEvent: () => emitOverseerSteering({ store: this.store as Parameters[0]["store"], taskId, stage: "executor", @@ -417,9 +418,7 @@ export class OverseerAdvisorService { outcome, severity, source: "session-advisor", - }); - } catch (err) { - log.warn(`emitOverseerSteering failed: ${err instanceof Error ? err.message : String(err)}`); - } + }), + }, { mutationType: "overseer:intervention" }, { log }); } } diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index 65ce426c7a..01ce6017f2 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -2069,11 +2069,11 @@ export class ProjectEngine { * handler already follows). */ private async emitOverseerInterventionSafe(fn: () => unknown | Promise): Promise { - try { - await fn(); - } catch (err) { - runtimeLog.warn(`Failed to emit overseer intervention: ${err instanceof Error ? err.message : String(err)}`); - } + await emitBoundedRunAudit( + { recordRunAuditEvent: () => fn() }, + { mutationType: "overseer:intervention" }, + { log: runtimeLog }, + ); } /** diff --git a/packages/engine/src/project/mesh-lease-manager.ts b/packages/engine/src/project/mesh-lease-manager.ts index 53e3c0c588..a2a365ee11 100644 --- a/packages/engine/src/project/mesh-lease-manager.ts +++ b/packages/engine/src/project/mesh-lease-manager.ts @@ -11,6 +11,7 @@ import type { NodeHealthMonitor } from "./node-health-monitor.js"; import { decideOwningNodeHandoff } from "./node-routing-policy.js"; import { createLogger } from "../logger.js"; import { createRunAuditor, generateSyntheticRunId } from "../util/run-audit.js"; +import { emitBoundedRunAudit } from "../util/emit-bounded-run-audit.js"; const meshLeaseManagerLog = createLogger("mesh-lease-manager"); @@ -443,8 +444,7 @@ export class MeshLeaseManager { handoffReason: handoffDecision.reason, }); meshLeaseManagerLog.log(`mesh-lease: handoff parked taskId=${task.id} reason=${handoffDecision.reason}`); - try { - await this.options.taskStore.recordRunAuditEvent?.({ + await emitBoundedRunAudit(this.options.taskStore, { taskId: task.id, agentId: "mesh-lease-manager", runId: generateSyntheticRunId("mesh-lease", task.id), @@ -466,10 +466,7 @@ export class MeshLeaseManager { source: "mesh-lease.recover", recoveryReason: reason, }, - }); - } catch (error) { - meshLeaseManagerLog.warn(`mesh-lease: failed to emit node:handoff:parked for taskId=${task.id}: ${error instanceof Error ? error.message : String(error)}`); - } + }, { log: meshLeaseManagerLog }); return false; } } @@ -521,8 +518,7 @@ export class MeshLeaseManager { }); } - try { - await this.options.taskStore.recordRunAuditEvent?.({ + await emitBoundedRunAudit(this.options.taskStore, { taskId: task.id, agentId: "mesh-lease-manager", runId: generateSyntheticRunId("mesh-lease", task.id), @@ -540,10 +536,7 @@ export class MeshLeaseManager { epoch: nextEpoch, recoveryReason: `${reason} (${stale.reason ?? "stale"})`, }, - }); - } catch (error) { - meshLeaseManagerLog.warn(`mesh-lease: failed to emit node:lease:recovered for taskId=${task.id}: ${error instanceof Error ? error.message : String(error)}`); - } + }, { log: meshLeaseManagerLog }); if (isUnreachableOwnerReason) { await emitNodeUnreachableRecovery({ diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 478ec83777..32c71a4d52 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -66,6 +66,7 @@ import { TriageProcessor } from "../triage.js"; import { validateProjectNodeMapping } from "../project/node-dispatch-validation.js"; import { attachAgentLinkSync } from "../agents/task-agent-sync.js"; import { createRunAuditor, generateSyntheticRunId } from "../util/run-audit.js"; +import { emitBoundedRunAudit } from "../util/emit-bounded-run-audit.js"; import { setImmediate as setImmediateCb } from "node:timers"; import { seedPreReleasePlanReviewContinuation, type PlanReviewSeedBailReason } from "../plan-review-continuation.js"; import { @@ -848,6 +849,25 @@ function formatRuntimeGitDetectionWarning(workingDirectory: string, detection: E `Task execution will fail until the Git error is resolved. Git reported: ${stderr}.${remedy}`; } +/** + * FNXC:RunAudit 2026-08-20-06:06: + * Credential rotation is runtime-owned recovery plumbing. Its optional audit adapter must use the + * bounded seam so an unavailable telemetry sink cannot delay a production rotation candidate. + */ +export function createRuntimeCredentialRotationAuditAdapter(taskStore: TaskStore) { + return async (mutationType: string, metadata: Record): Promise => { + await emitBoundedRunAudit(taskStore, { + taskId: typeof metadata.taskId === "string" ? metadata.taskId : undefined, + agentId: typeof metadata.agentId === "string" ? metadata.agentId : "runtime", + runId: generateSyntheticRunId("credential-instance-rotation", typeof metadata.taskId === "string" ? metadata.taskId : String(metadata.providerId ?? "unknown")), + domain: "database", + mutationType, + target: String(metadata.providerId ?? "unknown"), + metadata, + }); + }; +} + export class InProcessRuntime extends EventEmitter implements ProjectRuntime @@ -1030,17 +1050,7 @@ export class InProcessRuntime // Rotation evidence is emitted through the runtime-owned audit seam. Metadata // is supplied by the rotator as ids/counts/outcomes only; audit failures stay // non-fatal so an observability outage cannot prevent rate-limit recovery. - recordRunAuditEvent: async (mutationType, metadata) => { - await this.taskStore.recordRunAuditEvent?.({ - taskId: typeof metadata.taskId === "string" ? metadata.taskId : undefined, - agentId: typeof metadata.agentId === "string" ? metadata.agentId : "runtime", - runId: generateSyntheticRunId("credential-instance-rotation", typeof metadata.taskId === "string" ? metadata.taskId : String(metadata.providerId ?? "unknown")), - domain: "database", - mutationType, - target: String(metadata.providerId ?? "unknown"), - metadata, - }); - }, + recordRunAuditEvent: createRuntimeCredentialRotationAuditAdapter(this.taskStore), }); this.usageLimitPauser ??= new UsageLimitPauser(this.taskStore, { credentialRotator: this.credentialRotator, diff --git a/packages/engine/src/workflow-column-boundary-hooks.ts b/packages/engine/src/workflow-column-boundary-hooks.ts index 28707e84ad..b71faaa9e5 100644 --- a/packages/engine/src/workflow-column-boundary-hooks.ts +++ b/packages/engine/src/workflow-column-boundary-hooks.ts @@ -21,6 +21,7 @@ import { ACTIVE_WORKFLOW_WORK_ITEM_STATES } from "@fusion/core"; import { createStoreIrPinPersistence, type WorkflowIrPinStoreSurface } from "./workflows/workflow-column-boundary.js"; import type { WorkflowColumnBoundaryHooks } from "./workflows/workflow-graph-task-runner.js"; import { generateSyntheticRunId } from "./util/run-audit.js"; +import { emitBoundedRunAudit } from "./util/emit-bounded-run-audit.js"; export interface ExecutorColumnBoundaryHooksDeps { store: TaskStore; @@ -108,7 +109,7 @@ export function createExecutorColumnBoundaryHooks( } }, emitAudit: async (event) => { - await store.recordRunAuditEvent?.({ + await emitBoundedRunAudit(store, { taskId: event.taskId, agentId: "executor", runId: generateSyntheticRunId("workflow-column-boundary", event.taskId),