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

feat(FN-4825): complete Step 2 — emit mesh lease unreachable-owner audits

Browse Source 
Fusion-Task-Id: FN-4825
Fusion-Task-Lineage: dc4e633e-fe46-47a8-9ecc-f032073caee9
This commit is contained in:
Fusion (runfusion.ai)
2026-05-16 19:38:29 -07:00
committed by gsxdsm
parent 2db5c10f07
commit ee121a9f79
2 changed files with 235 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

147
packages/engine/src/__tests__/mesh-lease-manager.test.ts
View File

@@ -1,5 +1,5 @@
import { describe, expect, it, vi } from "vitest";
import type { AgentStore, Task, TaskStore } from "@fusion/core";
import type { AgentStore, RunAuditEventInput, Task, TaskStore } from "@fusion/core";
import { MeshLeaseManager } from "../mesh-lease-manager.js";
function task(overrides: Partial<Task> = {}): Task {
@@ -44,8 +44,9 @@ describe("MeshLeaseManager", () => {
expect(result).toEqual({ recoverable: true, reason: "owner_node_offline" });
});
it("recovers stale lease by bumping epoch and clearing owner fields", async () => {
const currentTask = task({ checkoutLeaseRenewedAt: "2026-05-01T00:00:00.000Z" });
it("emits unreachable audit for recovered-to-todo path", async () => {
const currentTask = task({ column: "in-progress" });
const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
const updateTask = vi.fn().mockResolvedValue(currentTask);
const moveTask = vi.fn().mockResolvedValue(currentTask);
const logEntry = vi.fn().mockResolvedValue(undefined);
@@ -54,32 +55,142 @@ describe("MeshLeaseManager", () => {
updateTask,
moveTask,
logEntry,
recordRunAuditEvent,
} as unknown as TaskStore;
const manager = new MeshLeaseManager({
taskStore,
nodeHealthMonitor: { getNodeHealth: () => "offline" } as any,
getHandoffPolicy: vi.fn().mockResolvedValue("reassign-any-healthy"),
localNodeId: "local",
});
const ok = await manager.recoverAbandonedLease("FN-1", "scheduler detected stale todo lease");
expect(ok).toBe(true);
expect(moveTask).toHaveBeenCalledWith("FN-1", "todo", expect.any(Object));
const event = recordRunAuditEvent.mock.calls[0]?.[0] as RunAuditEventInput;
expect(event.mutationType).toBe("task:auto-recover-node-unreachable");
expect(event.metadata).toMatchObject({
ownerNodeId: "node-a",
ownerNodeHealth: "offline",
previousOwnerAgentId: "agent-1",
previousColumn: "in-progress",
newColumn: "todo",
leaseEpoch: 2,
recoveryReason: "scheduler detected stale todo lease",
handoffPolicy: "reassign-any-healthy",
handoffAction: "reassign-any",
handoffReason: expect.any(String),
decisionPath: "lease-recovered-to-todo",
});
});
it("emits ownerNodeHealth=error for owner_node_error recoveries", async () => {
const currentTask = task({ column: "todo" });
const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
const taskStore = {
getTask: vi.fn().mockResolvedValue(currentTask),
updateTask: vi.fn().mockResolvedValue(currentTask),
moveTask: vi.fn().mockResolvedValue(currentTask),
logEntry: vi.fn().mockResolvedValue(undefined),
recordRunAuditEvent,
} as unknown as TaskStore;
const manager = new MeshLeaseManager({
taskStore,
nodeHealthMonitor: { getNodeHealth: () => "error" } as any,
getHandoffPolicy: vi.fn().mockResolvedValue("reassign-any-healthy"),
localNodeId: "local",
});
const ok = await manager.recoverAbandonedLease("FN-1", "scheduler detected stale todo lease");
expect(ok).toBe(true);
const event = recordRunAuditEvent.mock.calls[0]?.[0] as RunAuditEventInput;
expect(event.metadata).toMatchObject({
ownerNodeHealth: "error",
decisionPath: "lease-recovered-in-place",
newColumn: "todo",
});
});
it("emits parked-by-handoff-policy when handoff action parks", async () => {
const currentTask = task({ column: "in-progress" });
const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
const taskStore = {
getTask: vi.fn().mockResolvedValue(currentTask),
updateTask: vi.fn().mockResolvedValue(currentTask),
moveTask: vi.fn().mockResolvedValue(currentTask),
logEntry: vi.fn().mockResolvedValue(undefined),
recordRunAuditEvent,
} as unknown as TaskStore;
const manager = new MeshLeaseManager({
taskStore,
nodeHealthMonitor: { getNodeHealth: () => "offline" } as any,
getHandoffPolicy: vi.fn().mockResolvedValue("block"),
localNodeId: "local",
});
const ok = await manager.recoverAbandonedLease("FN-1", "scheduler detected stale todo lease");
expect(ok).toBe(false);
const event = recordRunAuditEvent.mock.calls[0]?.[0] as RunAuditEventInput;
expect(event.mutationType).toBe("task:auto-recover-node-unreachable");
expect(event.metadata).toMatchObject({
decisionPath: "lease-parked-by-handoff-policy",
recoveryReason: "handoff-policy-park",
handoffPolicy: "block",
handoffAction: "park",
newColumn: "in-progress",
leaseEpoch: 1,
});
});
it("does not emit node-unreachable event for non-unreachable recoveries", async () => {
const currentTask = task({ checkoutNodeId: undefined });
const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
const taskStore = {
getTask: vi.fn().mockResolvedValue(currentTask),
updateTask: vi.fn().mockResolvedValue(currentTask),
moveTask: vi.fn().mockResolvedValue(currentTask),
logEntry: vi.fn().mockResolvedValue(undefined),
recordRunAuditEvent,
} as unknown as TaskStore;
const agentStore = {
getAgent: vi.fn().mockResolvedValue({
id: "agent-1",
runtimeConfig: { heartbeatTimeoutMs: 60_000 },
lastHeartbeatAt: "2026-05-01T00:00:00.000Z",
lastHeartbeatAt: "2026-04-30T00:00:00.000Z",
}),
} as unknown as AgentStore;
const manager = new MeshLeaseManager({ taskStore, agentStore });
const ok = await manager.recoverAbandonedLease("FN-1", "stale-heartbeat");
expect(ok).toBe(true);
expect(updateTask).toHaveBeenCalledWith(
"FN-1",
expect.objectContaining({
checkedOutBy: null,
checkedOutAt: null,
checkoutNodeId: null,
checkoutRunId: null,
checkoutLeaseRenewedAt: null,
checkoutLeaseEpoch: 2,
}),
undefined,
);
expect(moveTask).toHaveBeenCalledWith("FN-1", "todo", expect.any(Object));
expect(recordRunAuditEvent).not.toHaveBeenCalled();
});
it("swallows audit emission failures and still returns expected result", async () => {
const currentTask = task({ column: "todo" });
const recordRunAuditEvent = vi.fn().mockRejectedValue(new Error("boom"));
const taskStore = {
getTask: vi.fn().mockResolvedValue(currentTask),
updateTask: vi.fn().mockResolvedValue(currentTask),
moveTask: vi.fn().mockResolvedValue(currentTask),
logEntry: vi.fn().mockResolvedValue(undefined),
recordRunAuditEvent,
} as unknown as TaskStore;
const manager = new MeshLeaseManager({
taskStore,
nodeHealthMonitor: { getNodeHealth: () => "offline" } as any,
getHandoffPolicy: vi.fn().mockResolvedValue("reassign-any-healthy"),
localNodeId: "local",
});
await expect(manager.recoverAbandonedLease("FN-1", "scheduler detected stale todo lease")).resolves.toBe(true);
});
});

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

@@ -8,6 +8,7 @@ import type {
import type { NodeHealthMonitor } from "./node-health-monitor.js";
import { decideOwningNodeHandoff } from "./node-routing-policy.js";
import { createLogger } from "./logger.js";
import { createRunAuditor, generateSyntheticRunId } from "./run-audit.js";
const meshLeaseManagerLog = createLogger("mesh-lease-manager");
@@ -90,25 +91,125 @@ export class MeshLeaseManager {
return false;
}
if ((stale.reason === "owner_node_offline" || stale.reason === "owner_node_error")
&& task.checkoutNodeId
&& this.options.nodeHealthMonitor) {
const ownerNodeHealth = this.options.nodeHealthMonitor.getNodeHealth(task.checkoutNodeId);
const isUnreachableOwnerReason = stale.reason === "owner_node_offline" || stale.reason === "owner_node_error";
const ownerNodeId = task.checkoutNodeId;
const ownerNodeHealth = stale.reason === "owner_node_error" ? "error" : "offline";
const previousOwnerAgentId = task.checkedOutBy;
const previousColumn = task.column;
const auditor = createRunAuditor(this.options.taskStore, {
runId: generateSyntheticRunId("mesh-lease", taskId),
agentId: "mesh-lease-manager",
taskId,
taskLineageId: task.lineageId,
phase: "recover-unreachable-owner-lease",
});
const emitNodeUnreachableRecovery = async ({
decisionPath,
newColumn,
leaseEpoch,
recoveryReason,
handoffPolicy,
handoffAction,
handoffReason,
}: {
decisionPath: "lease-parked-by-handoff-policy" | "lease-recovered-in-place" | "lease-recovered-to-todo";
newColumn: string;
leaseEpoch: number;
recoveryReason: string;
handoffPolicy: OwningNodeHandoffPolicy | undefined;
handoffAction: string;
handoffReason: string;
}): Promise<void> => {
if (!isUnreachableOwnerReason || !ownerNodeId) {
return;
}
try {
await auditor.database({
type: "task:auto-recover-node-unreachable",
target: taskId,
metadata: {
ownerNodeId,
ownerNodeHealth,
previousOwnerAgentId,
previousColumn,
newColumn,
leaseEpoch,
recoveryReason,
handoffPolicy,
handoffAction,
handoffReason,
decisionPath,
},
});
} catch (error) {
meshLeaseManagerLog.warn(
`mesh-lease: failed to emit node-unreachable auto-recovery audit for taskId=${task.id}: ${error instanceof Error ? error.message : String(error)}`,
);
}
};
if (isUnreachableOwnerReason && task.checkoutNodeId && this.options.nodeHealthMonitor) {
const currentOwnerNodeHealth = this.options.nodeHealthMonitor.getNodeHealth(task.checkoutNodeId);
const handoffPolicy = await this.options.getHandoffPolicy?.();
const handoffDecision = decideOwningNodeHandoff({
task,
ownerNodeId: task.checkoutNodeId,
ownerNodeHealth,
ownerNodeHealth: currentOwnerNodeHealth,
localNodeId: this.options.localNodeId ?? "local",
handoffPolicy,
});
if (handoffDecision.action === "park") {
await emitNodeUnreachableRecovery({
decisionPath: "lease-parked-by-handoff-policy",
newColumn: task.column,
leaseEpoch: task.checkoutLeaseEpoch ?? 0,
recoveryReason: "handoff-policy-park",
handoffPolicy,
handoffAction: handoffDecision.action,
handoffReason: handoffDecision.reason,
});
meshLeaseManagerLog.log(
`mesh-lease: handoff parked taskId=${task.id} reason=${handoffDecision.reason}`,
);
return false;
}
const nextEpoch = (task.checkoutLeaseEpoch ?? 0) + 1;
await this.options.taskStore.updateTask(
taskId,
{
checkedOutBy: null,
checkedOutAt: null,
checkoutNodeId: null,
checkoutRunId: null,
checkoutLeaseRenewedAt: null,
checkoutLeaseEpoch: nextEpoch,
},
context.runContext,
);
await this.options.taskStore.logEntry(
taskId,
"Recovered abandoned lease",
`${reason} (${stale.reason ?? "stale"}); epoch=${nextEpoch}`,
context.runContext,
);
if (task.column !== "todo") {
await this.options.taskStore.moveTask(taskId, "todo", {
preserveProgress: context.preserveProgress ?? (task.currentStep > 0 || task.steps.some((step) => step.status !== "pending")),
});
}
await emitNodeUnreachableRecovery({
decisionPath: task.column === "todo" ? "lease-recovered-in-place" : "lease-recovered-to-todo",
newColumn: task.column === "todo" ? task.column : "todo",
leaseEpoch: nextEpoch,
recoveryReason: reason,
handoffPolicy,
handoffAction: handoffDecision.action,
handoffReason: handoffDecision.reason,
});
return true;
}
const nextEpoch = (task.checkoutLeaseEpoch ?? 0) + 1;