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

feat(FN-3452): document mesh lease recovery semantics

Browse Source 
Documents mesh lease recovery semantics across the agents, architecture, and multi-project reference files, adding 32 lines of clarifying documentation to explain how mesh leases are recovered in the system.

Fusion-Task-Id: FN-3452
This commit is contained in:
Fusion
2026-05-09 11:24:24 -07:00
committed by gsxdsm
parent 90e9dde569
commit 80e40455aa
16 changed files with 486 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
packages/core/src/__tests__/agent-store.test.ts
View File

@@ -1847,23 +1847,45 @@ describe("AgentStore", () => {
taskStore.close();
});
it("checkoutTask acquires a lease and stamps checkedOutAt", async () => {
const updated = await store.checkoutTask(holderId, taskId);
it("checkoutTask acquires a lease and stamps lease metadata", async () => {
const updated = await store.checkoutTask(holderId, taskId, { nodeId: "node-a", runId: "run-1", leaseEpoch: 2 });
expect(updated.checkedOutBy).toBe(holderId);
expect(updated.checkedOutAt).toBeDefined();
expect(updated.checkoutNodeId).toBe("node-a");
expect(updated.checkoutRunId).toBe("run-1");
expect(updated.checkoutLeaseRenewedAt).toBeDefined();
expect(updated.checkoutLeaseEpoch).toBe(2);
const persisted = await taskStore.getTask(taskId);
expect(persisted?.checkedOutBy).toBe(holderId);
expect(persisted?.checkedOutAt).toBeDefined();
expect(persisted?.checkoutNodeId).toBe("node-a");
expect(persisted?.checkoutRunId).toBe("run-1");
expect(persisted?.checkoutLeaseRenewedAt).toBeDefined();
expect(persisted?.checkoutLeaseEpoch).toBe(2);
});
it("checkoutTask is idempotent when the same agent re-checks out", async () => {
const first = await store.checkoutTask(holderId, taskId);
const second = await store.checkoutTask(holderId, taskId);
it("checkoutTask is idempotent for same agent/node/epoch and renews lease timestamp", async () => {
const first = await store.checkoutTask(holderId, taskId, { nodeId: "node-a", runId: "run-1", leaseEpoch: 2 });
await new Promise((resolve) => setTimeout(resolve, 5));
const second = await store.checkoutTask(holderId, taskId, { nodeId: "node-a", runId: "run-2", leaseEpoch: 2 });
expect(second.checkedOutBy).toBe(holderId);
expect(second.checkedOutAt).toBe(first.checkedOutAt);
expect(second.checkoutNodeId).toBe("node-a");
expect(second.checkoutRunId).toBe("run-2");
expect(second.checkoutLeaseEpoch).toBe(2);
expect(second.checkoutLeaseRenewedAt).not.toBe(first.checkoutLeaseRenewedAt);
});
it("checkoutTask updates epoch for same holder when lease epoch increases", async () => {
await store.checkoutTask(holderId, taskId, { nodeId: "node-a", runId: "run-1", leaseEpoch: 1 });
const bumped = await store.checkoutTask(holderId, taskId, { nodeId: "node-a", runId: "run-2", leaseEpoch: 3 });
expect(bumped.checkedOutBy).toBe(holderId);
expect(bumped.checkoutLeaseEpoch).toBe(3);
expect(bumped.checkoutRunId).toBe("run-2");
});
it("checkoutTask throws CheckoutConflictError when already held by another agent", async () => {
@@ -1915,11 +1937,15 @@ describe("AgentStore", () => {
});
it("forceReleaseTask clears checkout regardless of holder", async () => {
await store.checkoutTask(holderId, taskId);
await store.checkoutTask(holderId, taskId, { nodeId: "node-a", runId: "run-1", leaseEpoch: 9 });
const released = await store.forceReleaseTask(taskId);
expect(released.checkedOutBy).toBeUndefined();
expect(released.checkedOutAt).toBeUndefined();
expect(released.checkoutNodeId).toBeUndefined();
expect(released.checkoutRunId).toBeUndefined();
expect(released.checkoutLeaseRenewedAt).toBeUndefined();
expect(released.checkoutLeaseEpoch).toBeUndefined();
});
it("getCheckedOutBy returns holder ID when checked out and undefined otherwise", async () => {

28
packages/core/src/__tests__/store.test.ts
View File

@@ -11755,6 +11755,34 @@ describe("RunMutationContext", () => {
});
describe("shared mesh snapshots", () => {
it("persists and replicates extended lease metadata", async () => {
const task = await store.createTask({ description: "lease snapshot task" });
await store.updateTask(task.id, {
checkedOutBy: "agent-1",
checkedOutAt: "2026-05-01T00:00:00.000Z",
checkoutNodeId: "node-a",
checkoutRunId: "run-1",
checkoutLeaseRenewedAt: "2026-05-01T00:01:00.000Z",
checkoutLeaseEpoch: 7,
});
const snapshot = await store.getTaskMetadataSnapshot();
const replicated = snapshot.payload.tasks.find((entry) => entry.id === task.id);
expect(replicated).toMatchObject({
checkedOutBy: "agent-1",
checkedOutAt: "2026-05-01T00:00:00.000Z",
checkoutNodeId: "node-a",
checkoutRunId: "run-1",
checkoutLeaseRenewedAt: "2026-05-01T00:01:00.000Z",
checkoutLeaseEpoch: 7,
});
await store.updateTask(task.id, { checkedOutBy: null, checkoutLeaseEpoch: 8 });
const released = await store.getTask(task.id);
expect(released).toMatchObject({ checkedOutBy: undefined, checkoutLeaseEpoch: 8 });
});
it("exports and reapplies task/activity/audit snapshots deterministically", async () => {
const task = await store.createTask({ description: "snapshot task" });
await store.updateTask(task.id, { worktree: "/tmp/fn-worktree", executionStartBranch: "fn/base" });

58
packages/core/src/agent-store.ts
View File

@@ -54,6 +54,13 @@ import {
} from "./types.js";
import type { RunMutationContext } from "./types.js";
import type { TaskStore } from "./store.js";
interface CheckoutLeaseContext {
nodeId?: string;
runId?: string;
leaseEpoch?: number;
renewedAt?: string;
}
import { computeAccessState } from "./agent-permissions.js";
import { canAgentTakeImplementationTask, formatRoleMismatchReason } from "./agent-role-policy.js";
import { resolveEffectiveAgentPermissionPolicy } from "./agent-permission-policy.js";
@@ -1341,7 +1348,7 @@ export class AgentStore extends EventEmitter {
}
try {
await this.checkoutTask(agentId, taskId, runContext);
await this.checkoutTask(agentId, taskId, undefined, runContext);
} catch (error) {
if (error instanceof CheckoutConflictError) {
return { ok: false, reason: "checkout_conflict", task };
@@ -1359,7 +1366,12 @@ export class AgentStore extends EventEmitter {
* Acquire a checkout lease for a task.
* Throws CheckoutConflictError when another agent already holds the lease.
*/
async checkoutTask(agentId: string, taskId: string, runContext?: RunMutationContext): Promise<Task> {
async checkoutTask(
agentId: string,
taskId: string,
leaseContext?: CheckoutLeaseContext,
runContext?: RunMutationContext,
): Promise<Task> {
if (!this.taskStore) {
throw new Error("TaskStore not configured for checkout operations");
}
@@ -1378,11 +1390,30 @@ export class AgentStore extends EventEmitter {
throw new CheckoutConflictError(taskId, task.checkedOutBy, agentId);
}
if (task.checkedOutBy === agentId) {
return task;
const nextEpoch = leaseContext?.leaseEpoch ?? task.checkoutLeaseEpoch ?? 0;
const nextRenewedAt = leaseContext?.renewedAt ?? new Date().toISOString();
const existingNodeId = task.checkoutNodeId;
const existingEpoch = task.checkoutLeaseEpoch ?? 0;
if (
task.checkedOutBy === agentId
&& existingNodeId === (leaseContext?.nodeId ?? existingNodeId)
&& existingEpoch === nextEpoch
) {
return this.taskStore.updateTask(taskId, {
checkoutRunId: leaseContext?.runId ?? task.checkoutRunId ?? null,
checkoutLeaseRenewedAt: nextRenewedAt,
});
}
const updated = await this.taskStore.updateTask(taskId, { checkedOutBy: agentId });
const updated = await this.taskStore.updateTask(taskId, {
checkedOutBy: agentId,
checkedOutAt: task.checkedOutBy === agentId ? task.checkedOutAt : undefined,
checkoutNodeId: leaseContext?.nodeId ?? null,
checkoutRunId: leaseContext?.runId ?? null,
checkoutLeaseRenewedAt: nextRenewedAt,
checkoutLeaseEpoch: nextEpoch,
});
await this.taskStore.logEntry(taskId, `Checked out by agent ${agentId}`, undefined, runContext);
return updated;
}
@@ -1408,7 +1439,13 @@ export class AgentStore extends EventEmitter {
return task;
}
const updated = await this.taskStore.updateTask(taskId, { checkedOutBy: null });
const updated = await this.taskStore.updateTask(taskId, {
checkedOutBy: null,
checkedOutAt: null,
checkoutNodeId: null,
checkoutRunId: null,
checkoutLeaseRenewedAt: null,
});
await this.taskStore.logEntry(taskId, `Released by agent ${agentId}`, undefined, runContext);
return updated;
}
@@ -1421,7 +1458,14 @@ export class AgentStore extends EventEmitter {
throw new Error("TaskStore not configured for checkout operations");
}
const updated = await this.taskStore.updateTask(taskId, { checkedOutBy: null });
const updated = await this.taskStore.updateTask(taskId, {
checkedOutBy: null,
checkedOutAt: null,
checkoutNodeId: null,
checkoutRunId: null,
checkoutLeaseRenewedAt: null,
checkoutLeaseEpoch: null,
});
await this.taskStore.logEntry(taskId, "Checkout force-released", undefined, runContext);
return updated;
}

12
packages/core/src/db.ts
View File

@@ -226,7 +226,13 @@ CREATE TABLE IF NOT EXISTS tasks (
sourceSessionId TEXT,
sourceMessageId TEXT,
sourceParentTaskId TEXT,
sourceMetadata TEXT
sourceMetadata TEXT,
checkedOutBy TEXT,
checkedOutAt TEXT,
checkoutNodeId TEXT,
checkoutRunId TEXT,
checkoutLeaseRenewedAt TEXT,
checkoutLeaseEpoch INTEGER DEFAULT 0
);
-- Config table (single row with project settings)
@@ -1459,6 +1465,10 @@ export class Database {
this.applyMigration(20, () => {
this.addColumnIfMissing("tasks", "checkedOutBy", "TEXT");
this.addColumnIfMissing("tasks", "checkedOutAt", "TEXT");
this.addColumnIfMissing("tasks", "checkoutNodeId", "TEXT");
this.addColumnIfMissing("tasks", "checkoutRunId", "TEXT");
this.addColumnIfMissing("tasks", "checkoutLeaseRenewedAt", "TEXT");
this.addColumnIfMissing("tasks", "checkoutLeaseEpoch", "INTEGER DEFAULT 0");
});
}

57
packages/core/src/store.ts
View File

@@ -121,6 +121,10 @@ interface TaskRow {
sourceMetadata: string | null;
checkedOutBy: string | null;
checkedOutAt: string | null;
checkoutNodeId: string | null;
checkoutRunId: string | null;
checkoutLeaseRenewedAt: string | null;
checkoutLeaseEpoch: number | null;
}
/** Database row shape for the task_documents table. */
@@ -831,6 +835,10 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
sourceMetadata: fromJson<Record<string, unknown>>(row.sourceMetadata) ?? undefined,
checkedOutBy: row.checkedOutBy || undefined,
checkedOutAt: row.checkedOutAt || undefined,
checkoutNodeId: row.checkoutNodeId || undefined,
checkoutRunId: row.checkoutRunId || undefined,
checkoutLeaseRenewedAt: row.checkoutLeaseRenewedAt || undefined,
checkoutLeaseEpoch: row.checkoutLeaseEpoch ?? undefined,
};
}
@@ -1060,7 +1068,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
"breakIntoSubtasks", "enabledWorkflowSteps", "modifiedFiles",
"missionId", "sliceId", "assignedAgentId", "pausedByAgentId", "assigneeUserId", "nodeId", "effectiveNodeId", "effectiveNodeSource",
"sourceType", "sourceAgentId", "sourceRunId", "sourceSessionId", "sourceMessageId", "sourceParentTaskId", "sourceMetadata",
"checkedOutBy", "checkedOutAt",
"checkedOutBy", "checkedOutAt", "checkoutNodeId", "checkoutRunId", "checkoutLeaseRenewedAt", "checkoutLeaseEpoch",
// `log` is fetched in slim mode so the server can aggregate
// `timedExecutionMs` from `[timing] … in <N>ms` entries before
// returning. The log itself is stripped from the response —
@@ -1109,7 +1117,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
"breakIntoSubtasks", "enabledWorkflowSteps", "modifiedFiles",
"missionId", "sliceId", "assignedAgentId", "pausedByAgentId", "assigneeUserId", "nodeId", "effectiveNodeId", "effectiveNodeSource",
"sourceType", "sourceAgentId", "sourceRunId", "sourceSessionId", "sourceMessageId", "sourceParentTaskId", "sourceMetadata",
"checkedOutBy", "checkedOutAt",
"checkedOutBy", "checkedOutAt", "checkoutNodeId", "checkoutRunId", "checkoutLeaseRenewedAt", "checkoutLeaseEpoch",
];
const limitedLog = `
@@ -1150,9 +1158,9 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
dependencies, steps, log, attachments, steeringComments,
comments, review, reviewState, workflowStepResults, prInfo, issueInfo,
sourceIssueProvider, sourceIssueRepository, sourceIssueExternalIssueId, sourceIssueNumber, sourceIssueUrl,
mergeDetails, breakIntoSubtasks, enabledWorkflowSteps, modifiedFiles, missionId, sliceId, assignedAgentId, pausedByAgentId, assigneeUserId, nodeId, effectiveNodeId, effectiveNodeSource, sourceType, sourceAgentId, sourceRunId, sourceSessionId, sourceMessageId, sourceParentTaskId, sourceMetadata, checkedOutBy, checkedOutAt
mergeDetails, breakIntoSubtasks, enabledWorkflowSteps, modifiedFiles, missionId, sliceId, assignedAgentId, pausedByAgentId, assigneeUserId, nodeId, effectiveNodeId, effectiveNodeSource, sourceType, sourceAgentId, sourceRunId, sourceSessionId, sourceMessageId, sourceParentTaskId, sourceMetadata, checkedOutBy, checkedOutAt, checkoutNodeId, checkoutRunId, checkoutLeaseRenewedAt, checkoutLeaseEpoch
) VALUES (
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
)
ON CONFLICT(id) DO UPDATE SET
title = excluded.title,
@@ -1237,7 +1245,11 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
sourceParentTaskId = excluded.sourceParentTaskId,
sourceMetadata = excluded.sourceMetadata,
checkedOutBy = excluded.checkedOutBy,
checkedOutAt = excluded.checkedOutAt
checkedOutAt = excluded.checkedOutAt,
checkoutNodeId = excluded.checkoutNodeId,
checkoutRunId = excluded.checkoutRunId,
checkoutLeaseRenewedAt = excluded.checkoutLeaseRenewedAt,
checkoutLeaseEpoch = excluded.checkoutLeaseEpoch
`).run(
task.id,
task.title ?? null,
@@ -1323,6 +1335,10 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
toJsonNullable(task.sourceMetadata),
task.checkedOutBy ?? null,
task.checkedOutAt ?? null,
task.checkoutNodeId ?? null,
task.checkoutRunId ?? null,
task.checkoutLeaseRenewedAt ?? null,
task.checkoutLeaseEpoch ?? 0,
);
this.db.bumpLastModified();
}
@@ -3287,7 +3303,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
async updateTask(
id: string,
updates: { title?: string; description?: string; priority?: TaskPriority | null; prompt?: string; worktree?: string | null; status?: string | null; dependencies?: string[]; steps?: import("./types.js").TaskStep[]; currentStep?: number; blockedBy?: string | null; assignedAgentId?: string | null; pausedByAgentId?: string | null; assigneeUserId?: string | null; nodeId?: string | null; effectiveNodeId?: string | null; effectiveNodeSource?: string | null; checkedOutBy?: string | null; checkedOutAt?: string | null; paused?: boolean; baseBranch?: string | null; branch?: string | null; executionStartBranch?: string | null; baseCommitSha?: string | null; size?: "S" | "M" | "L"; reviewLevel?: number; executionMode?: import("./types.js").ExecutionMode | null; mergeRetries?: number; workflowStepRetries?: number; stuckKillCount?: number | null; postReviewFixCount?: number | null; recoveryRetryCount?: number | null; taskDoneRetryCount?: number | null; verificationFailureCount?: number | null; mergeConflictBounceCount?: number | null; nextRecoveryAt?: string | null; enabledWorkflowSteps?: string[]; modelProvider?: string | null; modelId?: string | null; validatorModelProvider?: string | null; validatorModelId?: string | null; planningModelProvider?: string | null; planningModelId?: string | null; thinkingLevel?: string | null; error?: string | null; summary?: string | null; sessionFile?: string | null; executionStartedAt?: string | null; executionCompletedAt?: string | null; review?: import("./types.js").TaskReview | null; reviewState?: import("./types.js").TaskReviewState | null; workflowStepResults?: import("./types.js").WorkflowStepResult[] | null; mergeDetails?: import("./types.js").MergeDetails | null; sourceIssue?: import("./types.js").TaskSourceIssue | null; tokenUsage?: import("./types.js").TaskTokenUsage | null; modifiedFiles?: string[] | null; missionId?: string | null; sliceId?: string | null },
updates: { title?: string; description?: string; priority?: TaskPriority | null; prompt?: string; worktree?: string | null; status?: string | null; dependencies?: string[]; steps?: import("./types.js").TaskStep[]; currentStep?: number; blockedBy?: string | null; assignedAgentId?: string | null; pausedByAgentId?: string | null; assigneeUserId?: string | null; nodeId?: string | null; effectiveNodeId?: string | null; effectiveNodeSource?: string | null; checkedOutBy?: string | null; checkedOutAt?: string | null; checkoutNodeId?: string | null; checkoutRunId?: string | null; checkoutLeaseRenewedAt?: string | null; checkoutLeaseEpoch?: number | null; paused?: boolean; baseBranch?: string | null; branch?: string | null; executionStartBranch?: string | null; baseCommitSha?: string | null; size?: "S" | "M" | "L"; reviewLevel?: number; executionMode?: import("./types.js").ExecutionMode | null; mergeRetries?: number; workflowStepRetries?: number; stuckKillCount?: number | null; postReviewFixCount?: number | null; recoveryRetryCount?: number | null; taskDoneRetryCount?: number | null; verificationFailureCount?: number | null; mergeConflictBounceCount?: number | null; nextRecoveryAt?: string | null; enabledWorkflowSteps?: string[]; modelProvider?: string | null; modelId?: string | null; validatorModelProvider?: string | null; validatorModelId?: string | null; planningModelProvider?: string | null; planningModelId?: string | null; thinkingLevel?: string | null; error?: string | null; summary?: string | null; sessionFile?: string | null; executionStartedAt?: string | null; executionCompletedAt?: string | null; review?: import("./types.js").TaskReview | null; reviewState?: import("./types.js").TaskReviewState | null; workflowStepResults?: import("./types.js").WorkflowStepResult[] | null; mergeDetails?: import("./types.js").MergeDetails | null; sourceIssue?: import("./types.js").TaskSourceIssue | null; tokenUsage?: import("./types.js").TaskTokenUsage | null; modifiedFiles?: string[] | null; missionId?: string | null; sliceId?: string | null },
runContext?: RunMutationContext,
): Promise<Task> {
return this.withTaskLock(id, async () => {
@@ -3424,10 +3440,35 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
if (updates.checkedOutBy === null) {
task.checkedOutBy = undefined;
task.checkedOutAt = undefined;
task.checkoutNodeId = undefined;
task.checkoutRunId = undefined;
task.checkoutLeaseRenewedAt = undefined;
} else if (updates.checkedOutBy !== undefined) {
task.checkedOutBy = updates.checkedOutBy;
// Auto-set checkedOutAt when acquiring a lease (use provided value or generate timestamp)
task.checkedOutAt = updates.checkedOutAt ?? new Date().toISOString();
task.checkedOutAt = updates.checkedOutAt ?? task.checkedOutAt ?? new Date().toISOString();
task.checkoutNodeId = updates.checkoutNodeId ?? task.checkoutNodeId;
task.checkoutRunId = updates.checkoutRunId ?? task.checkoutRunId;
task.checkoutLeaseRenewedAt = updates.checkoutLeaseRenewedAt ?? task.checkoutLeaseRenewedAt ?? task.checkedOutAt;
}
if (updates.checkoutNodeId === null) {
task.checkoutNodeId = undefined;
} else if (updates.checkoutNodeId !== undefined && updates.checkedOutBy === undefined) {
task.checkoutNodeId = updates.checkoutNodeId;
}
if (updates.checkoutRunId === null) {
task.checkoutRunId = undefined;
} else if (updates.checkoutRunId !== undefined && updates.checkedOutBy === undefined) {
task.checkoutRunId = updates.checkoutRunId;
}
if (updates.checkoutLeaseRenewedAt === null) {
task.checkoutLeaseRenewedAt = undefined;
} else if (updates.checkoutLeaseRenewedAt !== undefined && updates.checkedOutBy === undefined) {
task.checkoutLeaseRenewedAt = updates.checkoutLeaseRenewedAt;
}
if (updates.checkoutLeaseEpoch === null) {
task.checkoutLeaseEpoch = undefined;
} else if (updates.checkoutLeaseEpoch !== undefined) {
task.checkoutLeaseEpoch = updates.checkoutLeaseEpoch;
}
if (updates.paused !== undefined) task.paused = updates.paused || undefined;
if (updates.baseBranch === null) {

8
packages/core/src/types.ts
View File

@@ -1223,6 +1223,14 @@ export interface Task {
checkedOutBy?: string;
/** ISO-8601 timestamp when the checkout lease was acquired. */
checkedOutAt?: string;
/** Node ID currently owning the checkout lease. */
checkoutNodeId?: string;
/** Owning run/session ID for the checkout lease when known. */
checkoutRunId?: string;
/** ISO-8601 timestamp of the last successful lease renewal heartbeat. */
checkoutLeaseRenewedAt?: string;
/** Monotonically increasing lease generation used to prevent stale reclaim attempts. */
checkoutLeaseEpoch?: number;
/** Path to the persisted agent session file, enabling pause/resume without
* losing conversation context. Set when execution starts; cleared on
* completion or terminal failure. */

85
packages/engine/src/__tests__/mesh-lease-manager.test.ts Normal file
View File

@@ -0,0 +1,85 @@
import { describe, expect, it, vi } from "vitest";
import type { AgentStore, Task, TaskStore } from "@fusion/core";
import { MeshLeaseManager } from "../mesh-lease-manager.js";
function task(overrides: Partial<Task> = {}): Task {
return {
id: "FN-1",
description: "x",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: "2026-05-01T00:00:00.000Z",
updatedAt: "2026-05-01T00:00:00.000Z",
checkedOutBy: "agent-1",
checkedOutAt: "2026-05-01T00:00:00.000Z",
checkoutLeaseRenewedAt: "2026-05-01T00:00:00.000Z",
checkoutLeaseEpoch: 1,
checkoutNodeId: "node-a",
...overrides,
};
}
describe("MeshLeaseManager", () => {
it("prefers active local execution over stale replicated timestamps", async () => {
const getTask = vi.fn().mockResolvedValue(task());
const manager = new MeshLeaseManager({
taskStore: { getTask } as unknown as TaskStore,
getExecutingTaskIds: () => new Set(["FN-1"]),
});
const result = await manager.isLeaseRecoverable(task(), Date.parse("2026-05-01T00:10:00.000Z"));
expect(result).toEqual({ recoverable: false, reason: "active_local_execution" });
});
it("marks lease recoverable when owner node is offline", async () => {
const manager = new MeshLeaseManager({
taskStore: {} as TaskStore,
nodeHealthMonitor: { getNodeHealth: () => "offline" } as any,
});
const result = await manager.isLeaseRecoverable(task(), Date.parse("2026-05-01T00:01:00.000Z"));
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" });
const updateTask = vi.fn().mockResolvedValue(currentTask);
const moveTask = vi.fn().mockResolvedValue(currentTask);
const logEntry = vi.fn().mockResolvedValue(undefined);
const taskStore = {
getTask: vi.fn().mockResolvedValue(currentTask),
updateTask,
moveTask,
logEntry,
} as unknown as TaskStore;
const agentStore = {
getAgent: vi.fn().mockResolvedValue({
id: "agent-1",
runtimeConfig: { heartbeatTimeoutMs: 60_000 },
lastHeartbeatAt: "2026-05-01T00: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));
});
});

42
packages/engine/src/executor.ts
View File

@@ -595,6 +595,34 @@ export class TaskExecutor {
/** Set of ephemeral spawned agent IDs with in-flight cleanup (prevents duplicate deletion attempts). */
private pendingEphemeralDeletions = new Set<string>();
private async renewTaskLease(
taskId: string,
agentId: string,
leaseEpoch: number,
nodeId: string,
runId: string | undefined,
): Promise<void> {
const renewedAt = new Date().toISOString();
if (this.options.agentStore) {
await this.options.agentStore.checkoutTask(
agentId,
taskId,
{
nodeId,
runId,
leaseEpoch,
renewedAt,
},
this.currentRunContext,
);
return;
}
await this.store.updateTask(taskId, {
checkoutRunId: runId ?? null,
checkoutLeaseRenewedAt: renewedAt,
});
}
private async finalizeAlreadyReviewedTask(taskId: string): Promise<"merged" | "blocked" | "missing"> {
const latestTask = await this.store.getTask(taskId);
if (!latestTask || latestTask.column !== "in-review") {
@@ -3136,6 +3164,17 @@ export class TaskExecutor {
lastAssignedAgentId: detail.assignedAgentId ?? null,
});
let leaseRenewalTimer: ReturnType<typeof setInterval> | undefined;
if (detail.assignedAgentId && detail.checkedOutBy === detail.assignedAgentId) {
const leaseEpoch = detail.checkoutLeaseEpoch ?? 0;
const checkoutNodeId = detail.checkoutNodeId ?? detail.effectiveNodeId ?? detail.nodeId ?? "local";
const runId = this.currentRunContext?.runId;
await this.renewTaskLease(task.id, detail.assignedAgentId, leaseEpoch, checkoutNodeId, runId).catch(() => {});
leaseRenewalTimer = setInterval(() => {
void this.renewTaskLease(task.id, detail.assignedAgentId!, leaseEpoch, checkoutNodeId, runId).catch(() => {});
}, 30_000);
}
// Register with stuck task detector for heartbeat monitoring
stuckDetector?.trackTask(task.id, session);
executorLog.log(`${task.id}: session registered (model=${describeModel(session)}, stuckDetector=${!!stuckDetector})`);
@@ -3559,6 +3598,9 @@ export class TaskExecutor {
}
}
} finally {
if (leaseRenewalTimer) {
clearInterval(leaseRenewalTimer);
}
this.activeSessions.delete(task.id);
stuckDetector?.untrackTask(task.id);
await agentLogger.flush();

1
packages/engine/src/index.ts
View File

@@ -17,6 +17,7 @@ export { TriageProcessor, type TriageProcessorOptions } from "./triage.js";
export { TaskExecutor, type TaskExecutorOptions } from "./executor.js";
export { collectTaskEvaluationEvidence } from "./evaluator-evidence.js";
export { Scheduler, type SchedulerOptions } from "./scheduler.js";
export { MeshLeaseManager, type MeshLeaseManagerOptions, type LeaseRecoveryContext } from "./mesh-lease-manager.js";
export { MissionAutopilot, type MissionAutopilotOptions } from "./mission-autopilot.js";
export { MissionExecutionLoop, type MissionExecutionLoopOptions, type ValidationResult, loopLog } from "./mission-execution-loop.js";
export { aiMergeTask, type MergerOptions } from "./merger.js";

107
packages/engine/src/mesh-lease-manager.ts Normal file
View File

@@ -0,0 +1,107 @@
import type { AgentStore, RunMutationContext, Task, TaskStore } from "@fusion/core";
import type { NodeHealthMonitor } from "./node-health-monitor.js";
export interface MeshLeaseManagerOptions {
taskStore: TaskStore;
agentStore?: AgentStore;
nodeHealthMonitor?: NodeHealthMonitor;
getExecutingTaskIds?: () => Set<string>;
}
export interface LeaseRecoveryContext {
runContext?: RunMutationContext;
preserveProgress?: boolean;
}
export class MeshLeaseManager {
constructor(private readonly options: MeshLeaseManagerOptions) {}
private staleThresholdMs(agentHeartbeatTimeoutMs?: number): number {
return Math.max((agentHeartbeatTimeoutMs ?? 60_000) * 2, 120_000);
}
async isLeaseRecoverable(task: Task, now = Date.now()): Promise<{ recoverable: boolean; reason?: string }> {
if (!task.checkedOutBy) {
return { recoverable: false, reason: "no_lease" };
}
if (this.options.getExecutingTaskIds?.().has(task.id)) {
return { recoverable: false, reason: "active_local_execution" };
}
if (task.checkoutNodeId && this.options.nodeHealthMonitor) {
const status = this.options.nodeHealthMonitor.getNodeHealth(task.checkoutNodeId);
if (status === "offline" || status === "error") {
return { recoverable: true, reason: `owner_node_${status}` };
}
}
const renewedAtIso = task.checkoutLeaseRenewedAt ?? task.checkedOutAt;
if (!renewedAtIso) {
return { recoverable: false, reason: "lease_never_renewed" };
}
let heartbeatTimeoutMs = 60_000;
let ownerLastHeartbeatAt: string | undefined;
if (this.options.agentStore && task.checkedOutBy) {
const owner = await this.options.agentStore.getAgent(task.checkedOutBy);
if (owner?.runtimeConfig && typeof owner.runtimeConfig.heartbeatTimeoutMs === "number") {
heartbeatTimeoutMs = owner.runtimeConfig.heartbeatTimeoutMs;
}
ownerLastHeartbeatAt = owner?.lastHeartbeatAt;
}
const staleMs = this.staleThresholdMs(heartbeatTimeoutMs);
const renewedAtMs = Date.parse(renewedAtIso);
if (!Number.isFinite(renewedAtMs) || now - renewedAtMs < staleMs) {
return { recoverable: false, reason: "lease_not_stale" };
}
if (!ownerLastHeartbeatAt) {
return { recoverable: true, reason: "owner_heartbeat_missing" };
}
const ownerHeartbeatMs = Date.parse(ownerLastHeartbeatAt);
if (!Number.isFinite(ownerHeartbeatMs) || now - ownerHeartbeatMs >= staleMs) {
return { recoverable: true, reason: "owner_heartbeat_stale" };
}
return { recoverable: false, reason: "owner_heartbeat_fresh" };
}
async recoverAbandonedLease(taskId: string, reason: string, context: LeaseRecoveryContext = {}): Promise<boolean> {
const task = await this.options.taskStore.getTask(taskId);
if (!task) return false;
const stale = await this.isLeaseRecoverable(task);
if (!stale.recoverable) {
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")),
});
}
return true;
}
}

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

@@ -33,6 +33,7 @@ import { runtimeLog } from "../logger.js";
import { StuckTaskDetector } from "../stuck-task-detector.js";
import type { UsageLimitPauser } from "../usage-limit-detector.js";
import { SelfHealingManager } from "../self-healing.js";
import { MeshLeaseManager } from "../mesh-lease-manager.js";
import { PluginRunner } from "../plugin-runner.js";
import { MissionAutopilot } from "../mission-autopilot.js";
import { MissionExecutionLoop } from "../mission-execution-loop.js";
@@ -91,6 +92,7 @@ export class InProcessRuntime
private stuckTaskDetector?: StuckTaskDetector;
private usageLimitPauser?: UsageLimitPauser;
private selfHealingManager?: SelfHealingManager;
private leaseManager?: MeshLeaseManager;
private agentStore?: AgentStore;
private heartbeatMonitor?: HeartbeatMonitor;
private triggerScheduler?: HeartbeatTriggerScheduler;
@@ -285,6 +287,12 @@ export class InProcessRuntime
})
: undefined;
this.leaseManager = new MeshLeaseManager({
taskStore: this.taskStore,
agentStore: this.agentStore,
getExecutingTaskIds: () => this.executor?.getExecutingTaskIds() ?? new Set<string>(),
});
this.scheduler = new Scheduler(this.taskStore, {
maxConcurrent: this.config.maxConcurrent,
maxWorktrees: this.config.maxWorktrees,
@@ -292,6 +300,7 @@ export class InProcessRuntime
missionStore,
missionAutopilot,
missionExecutionLoop,
leaseManager: this.leaseManager,
onTaskFailed: (taskId) => {
if (missionAutopilot) {
void missionAutopilot.handleTaskFailure(taskId);
@@ -623,6 +632,7 @@ export class InProcessRuntime
evictStaleTriageProcessing: () => this.triageProcessor?.evictStaleProcessing() ?? new Set<string>(),
enqueueMerge: this.mergeEnqueuer ? (taskId: string) => this.mergeEnqueuer?.(taskId) : undefined,
getActiveMergeTaskId: () => this.activeMergeTaskIdProvider?.() ?? null,
leaseManager: this.leaseManager,
});
this.selfHealingManager.start();
this.stuckTaskDetector.start();

15
packages/engine/src/scheduler.ts
View File

@@ -20,6 +20,7 @@ import { evaluateSpecStaleness, getPromptPath } from "./spec-staleness.js";
import { resolveEffectiveNode } from "./effective-node.js";
import { applyUnavailableNodePolicy } from "./node-routing-policy.js";
import type { NodeDispatchValidationResult } from "./node-dispatch-validation.js";
import type { MeshLeaseManager } from "./mesh-lease-manager.js";
/**
* Check whether two sets of file scope paths overlap.
@@ -117,6 +118,8 @@ export interface SchedulerOptions {
prMonitor?: PrMonitor;
/** Optional MissionStore for slice activation and auto-advance */
missionStore?: MissionStore;
/** Optional lease manager used to recover stale checkout leases before scheduling. */
leaseManager?: MeshLeaseManager;
/** Optional MissionAutopilot for autonomous mission progression */
missionAutopilot?: import("./mission-autopilot.js").MissionAutopilot;
/**
@@ -676,6 +679,18 @@ export class Scheduler {
for (const taskId of ordered) {
const task = tasks.find((t) => t.id === taskId)!;
if (task.checkedOutBy && this.options.leaseManager) {
const recovered = await this.options.leaseManager.recoverAbandonedLease(
task.id,
"scheduler detected stale todo lease",
{ preserveProgress: true },
);
if (!recovered) {
await this.store.updateTask(task.id, { status: "queued" });
continue;
}
}
// Check all deps are satisfied (done, in-review, or archived)
const unmetDeps = task.dependencies.filter((depId) => {
const dep = tasks.find((t) => t.id === depId);

15
packages/engine/src/self-healing.ts
View File

@@ -18,6 +18,7 @@ import { promisify } from "node:util";
import { existsSync, readdirSync, rmSync, statSync } from "node:fs";
import { isAbsolute, join, relative, resolve } from "node:path";
import { getTaskMergeBlocker, isEphemeralAgent, type AgentStore, type TaskStore, type Settings, type Task, type MergeDetails } from "@fusion/core";
import type { MeshLeaseManager } from "./mesh-lease-manager.js";
import { createLogger } from "./logger.js";
import { getRegisteredWorktreePaths, scanIdleWorktrees, scanOrphanedBranches } from "./worktree-pool.js";
@@ -29,6 +30,8 @@ export interface SelfHealingOptions {
rootDir: string;
/** Optional AgentStore for agent-level self-healing checks. */
agentStore?: AgentStore;
/** Canonical stale-lease recovery manager. */
leaseManager?: MeshLeaseManager;
/**
* Callback to recover a completed task that is stuck in in-progress.
* Called by the periodic maintenance cycle when it detects a task whose
@@ -1491,6 +1494,18 @@ export class SelfHealingManager {
? "worktree exists but no active session"
: "missing worktree/session";
if (this.options.leaseManager && task.checkedOutBy) {
const leaseRecovered = await this.options.leaseManager.recoverAbandonedLease(
task.id,
`orphaned execution: ${reason}`,
{ preserveProgress: true },
);
if (leaseRecovered) {
recovered++;
continue;
}
}
// Reset steps whose work was never committed before clearing the worktree
await this.resetStepsIfWorkLost(task);