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) <noreply@runfusion.ai>

.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.

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.

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.
+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.

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;
 
 /**

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<void> }, 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<void>(() => {})],
+  ["late", () => new Promise<void>(() => {})],
+];
+
+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<string, AgentHeartbeatRun>();
+  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<void>((_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); }
+  });
+});

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<void>(() => {})],
+  ["late", () => new Promise<void>(() => {})],
+];
+
+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<void>((_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(); }
+  });
+});

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<void>(() => {})],
+];
+
+function task(overrides: Partial<Task>): 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<unknown>, 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<void>((_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); }
+  });
+});

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<void>(() => {})],
+    ["late", () => new Promise<void>(() => {})],
+  ])("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<void>(() => {})],
+    ["late", () => new Promise<void>(() => {})],
+  ])("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<void>((_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); }
+  });
+});

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<string, unknown>) => unknown);
+const sinks: [string, Sink][] = [
+  ["absent", undefined], ["throws", () => { throw new Error("down"); }],
+  ["rejects", () => Promise.reject(new Error("down"))], ["hangs", () => new Promise<void>(() => {})],
+  ["late", () => new Promise<void>(() => {})],
+];
+
+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<void>((_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); }
+  });
+});

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<string, unknown>) => {
-      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,

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<Dependency
   const legacy = legacyDependencySatisfied(dep) || markerAccepted === true;
 
   if (completeFlag !== legacy) {
-    try {
+    /*
-      void ctx.store.recordRunAuditEvent?.({
+    FNXC:RunAudit 2026-08-20-05:39:
+    Hold-release telemetry is best-effort: a hostile audit sink must never gate a sweep,
+    force-promote, or event release.
+    */
+    void emitBoundedRunAudit(ctx.store, {
         taskId: dep.id,
         agentId: "scheduler",
         runId: `hold-release:${dep.id}`,
@@ -424,10 +429,7 @@ async function dependencySatisfied(ctx: SweepCtx, dep: Task): Promise<Dependency
           legacyResult: legacy,
           source: "hold-release.dependency",
         },
-      });
+      }, { log: schedulerLog });
-    } catch {
-      // Audit is best-effort.
-    }
   }
   // Dual-accept: satisfied if EITHER signal says so (the dual-accept window
   // closes at graduation per U12; until then both are accepted).
@@ -987,9 +989,7 @@ export async function promoteHeldTask(
       }
     }
     // ids/outcomes-only audit of the override (no prompt or reason prose).
-    if (typeof store.recordRunAuditEvent === "function") {
+    await emitBoundedRunAudit(store, {
-      try {
-        await store.recordRunAuditEvent({
           domain: "database",
           mutationType: "task:promote-forced-unplanned",
           target: task.id,
@@ -997,11 +997,7 @@ export async function promoteHeldTask(
           agentId: "system",
           runId: `promote-force-${task.id}`,
           metadata: { fromColumn: task.column, toColumn: target, priorStatus: task.status ?? null },
-        });
+        }, { log: schedulerLog });
-      } catch {
-        // Audit is best-effort — never block the operator's override on it.
-      }
-    }
     schedulerLog.log(`Force-promote for ${task.id} bypassing unplanned gate into ${target} (operator override)`);
   }
 
@@ -1040,8 +1036,7 @@ export async function releaseHeldTaskByEvent(
   if (!column || !holdConfig || holdConfig.release !== "external-event") {
     return { released: false, rejection: "not-external-event-hold" };
   }
-  try {
+  void emitBoundedRunAudit(store, {
-    void store.recordRunAuditEvent?.({
       taskId,
       agentId: "scheduler",
       runId: `hold-release:event:${taskId}`,
@@ -1049,10 +1044,7 @@ export async function releaseHeldTaskByEvent(
       mutationType: "task:hold-release-event",
       target: taskId,
       metadata: { eventTag, fromColumn: task.column },
-    });
+    }, { log: schedulerLog });
-  } catch {
-    // best-effort
-  }
   const target = resolveReleaseTarget(ir, task.column, true);
   if (!target) return { released: false, rejection: "no-release-target" };
 

packages/engine/src/goals/goal-anchoring-audit.ts

@@ -2,6 +2,7 @@ import type { TaskStore } from "@fusion/core";
 
 import type { RunAuditor } from "../util/run-audit.js";
 import { createLogger } from "../logger.js";
+import { emitBoundedRunAudit } from "../util/emit-bounded-run-audit.js";
 const log = createLogger("goal-anchoring-audit");
 
 /** Goal context was injected into heartbeat/executor prompts for the Slice 2 cite-rate experiment. */
@@ -74,8 +75,7 @@ export function emitGoalRetrievalAudit(
 ): void {
   if (!ctx.runId || !ctx.agentId) return;
 
-  try {
+  void emitBoundedRunAudit(store, {
-    void store.recordRunAuditEvent({
       runId: ctx.runId,
       agentId: ctx.agentId,
       taskId: ctx.taskId,
@@ -88,8 +88,5 @@ export function emitGoalRetrievalAudit(
         goalIds: input.goalIds ?? [],
         notFound: input.notFound ?? false,
       },
-    });
+  }, { log: { warn: (message: string) => log.debug("goal retrieval audit emission skipped", message) } });
-  } catch (error) {
-    log.debug("goal retrieval audit emission skipped", error);
-  }
 }

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)}`,

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 {
+    void emitBoundedRunAudit({
-      emitOverseerSteering({
+      recordRunAuditEvent: () => emitOverseerSteering({
         store: this.store as Parameters<typeof emitOverseerSteering>[0]["store"],
         taskId,
         stage: "executor",
@@ -417,9 +418,7 @@ export class OverseerAdvisorService {
         outcome,
         severity,
         source: "session-advisor",
-      });
+      }),
-    } catch (err) {
+    }, { mutationType: "overseer:intervention" }, { log });
-      log.warn(`emitOverseerSteering failed: ${err instanceof Error ? err.message : String(err)}`);
-    }
   }
 }

packages/engine/src/project-engine.ts

@@ -2069,11 +2069,11 @@ export class ProjectEngine {
    * handler already follows).
    */
   private async emitOverseerInterventionSafe(fn: () => unknown | Promise<unknown>): Promise<void> {
-    try {
+    await emitBoundedRunAudit(
-      await fn();
+      { recordRunAuditEvent: () => fn() },
-    } catch (err) {
+      { mutationType: "overseer:intervention" },
-      runtimeLog.warn(`Failed to emit overseer intervention: ${err instanceof Error ? err.message : String(err)}`);
+      { log: runtimeLog },
-    }
+    );
   }
 
   /**

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 emitBoundedRunAudit(this.options.taskStore, {
-          await this.options.taskStore.recordRunAuditEvent?.({
             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,
             },
-          });
+          }, { log: meshLeaseManagerLog });
-        } catch (error) {
-          meshLeaseManagerLog.warn(`mesh-lease: failed to emit node:handoff:parked for taskId=${task.id}: ${error instanceof Error ? error.message : String(error)}`);
-        }
         return false;
       }
     }
@@ -521,8 +518,7 @@ export class MeshLeaseManager {
       });
     }
 
-    try {
+    await emitBoundedRunAudit(this.options.taskStore, {
-      await this.options.taskStore.recordRunAuditEvent?.({
         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"})`,
         },
-      });
+      }, { log: meshLeaseManagerLog });
-    } catch (error) {
-      meshLeaseManagerLog.warn(`mesh-lease: failed to emit node:lease:recovered for taskId=${task.id}: ${error instanceof Error ? error.message : String(error)}`);
-    }
 
     if (isUnreachableOwnerReason) {
       await emitNodeUnreachableRecovery({

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<string, unknown>): Promise<void> => {
+    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<ProjectRuntimeEvents>
   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) => {
+        recordRunAuditEvent: createRuntimeCredentialRotationAuditAdapter(this.taskStore),
-          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,
-          });
-        },
       });
       this.usageLimitPauser ??= new UsageLimitPauser(this.taskStore, {
         credentialRotator: this.credentialRotator,

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),