root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FN-9176: Route remaining run-audit emitters through bounded seam

Browse Source 
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>
This commit is contained in:
gsxdsm
2026-08-19 23:20:27 -07:00
parent a426e28781
commit 6d51ae84c3
18 changed files with 524 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/fn-9176-remaining-run-audit-emitter-isolation.md Normal file
View File

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

5
AGENTS.md
View File

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

2
docs/run-audit.md
View File

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

8
packages/engine/src/__tests__/non-executor-run-audit-emitter-isolation.test.ts
View File

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

116
packages/engine/src/__tests__/run-audit-sink-health-boundary-heartbeat.test.ts Normal file
View File

@@ -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); }
});
});

71
packages/engine/src/__tests__/run-audit-sink-health-goals.test.ts Normal file
View File

@@ -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(); }
});
});

102
packages/engine/src/__tests__/run-audit-sink-health-hold-release.test.ts Normal file
View File

@@ -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); }
});
});

71
packages/engine/src/__tests__/run-audit-sink-health-overseer.test.ts Normal file
View File

@@ -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); }
});
});

82
packages/engine/src/__tests__/run-audit-sink-health-runtime-leases.test.ts Normal file
View File

@@ -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); }
});
});

6
packages/engine/src/credential-instance-rotation.ts
View File

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

32
packages/engine/src/execution/hold-release.ts
View File

@@ -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",
},
});
} catch {
// Audit is best-effort.
}
}, { log: schedulerLog });
}
// 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") {
try {
await store.recordRunAuditEvent({
await emitBoundedRunAudit(store, {
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 },
});
} catch {
// Audit is best-effort — never block the operator's override on it.
}
}
}, { log: schedulerLog });
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 store.recordRunAuditEvent?.({
void emitBoundedRunAudit(store, {
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 },
});
} catch {
// best-effort
}
}, { log: schedulerLog });
const target = resolveReleaseTarget(ir, task.column, true);
if (!target) return { released: false, rejection: "no-release-target" };

9
packages/engine/src/goals/goal-anchoring-audit.ts
View File

@@ -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 store.recordRunAuditEvent({
void emitBoundedRunAudit(store, {
runId: ctx.runId,
agentId: ctx.agentId,
taskId: ctx.taskId,
@@ -88,8 +88,5 @@ export function emitGoalRetrievalAudit(
goalIds: input.goalIds ?? [],
notFound: input.notFound ?? false,
},
});
} catch (error) {
log.debug("goal retrieval audit emission skipped", error);
}
}, { log: { warn: (message: string) => log.debug("goal retrieval audit emission skipped", message) } });
}

5
packages/engine/src/goals/goal-injection-diagnostics.ts
View File

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

11
packages/engine/src/overseer/overseer-advisor-service.ts
View File

@@ -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<typeof emitOverseerSteering>[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 });
}
}

10
packages/engine/src/project-engine.ts
View File

@@ -2069,11 +2069,11 @@ export class ProjectEngine {
* handler already follows).
*/
private async emitOverseerInterventionSafe(fn: () => unknown | Promise<unknown>): Promise<void> {
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 },
);
}
/**

17
packages/engine/src/project/mesh-lease-manager.ts
View File

@@ -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({

32
packages/engine/src/runtimes/in-process-runtime.ts
View File

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

3
packages/engine/src/workflow-column-boundary-hooks.ts
View File

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