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

feat(FN-4823): complete lease recovery central-claim reconciliation

Browse Source 
Fusion-Task-Id: FN-4823
Fusion-Task-Lineage: 0034c04f-df82-4a1b-9f25-b93643a2c157
This commit is contained in:
Fusion (runfusion.ai)
2026-05-16 21:15:54 -07:00
committed by gsxdsm
parent 0b28388876
commit 6fa9aef630
12 changed files with 554 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
.changeset/fn-4823-lease-recovery-central-claim.md Normal file
View File

@@ -0,0 +1,10 @@
---
"@runfusion/fusion": patch
---
Lease recovery is now central-claim-aware: `MeshLeaseManager.recoverAbandonedLease`
releases the central claim before clearing local task-row lease fields, and
reconciles split-brain state via `reconcileLeaseRow` on the next scheduler /
self-healing tick. Owner-offline handoff policy and progress-preserving handoff
semantics are unchanged. Single-node deployments (no central claim store) keep
the existing local-only behavior. (FN-4823, FN-4819 §2.5 / §3.3 / §3.6)

2
AGENTS.md
View File

@@ -485,7 +485,7 @@ Task ownership supports explicit checkout leases. Agents should be aware of:
- If `task.checkedOutBy` is set to another agent, the run exits with `reason: "checkout_conflict"`
- Heartbeat execution does not auto-checkout — callers are responsible for obtaining checkout before starting work
When a `CentralClaimStore` is wired, the authoritative lease owner is the central `taskClaims` row in `~/.fusion/fusion-central.db`; per-project task lease fields are treated as a synchronization mirror of that central result. Without a claim store configured, checkout behavior remains the existing single-node per-project lease flow.
When a `CentralClaimStore` is wired, the authoritative lease owner is the central `taskClaims` row in `~/.fusion/fusion-central.db`; per-project task lease fields are treated as a synchronization mirror of that central result. Lease recovery follows FN-4819 §2.5: `MeshLeaseManager.recoverAbandonedLease()` releases central claim ownership first, then clears local lease fields and bumps the local epoch. If one write succeeds and the other fails, `reconcileLeaseRow(taskId)` converges local vs central state on the next scheduler/self-healing tick and emits `task:auto-recover-lease-*` run-audit telemetry. Without a claim store configured, checkout behavior remains the existing single-node per-project lease flow.
## Per-Agent Heartbeat Configuration

6
docs/multi-project.md
View File

@@ -53,9 +53,11 @@ Peer/mesh coordination spans core + engine, with startup ownership in CLI proces
Task ownership is shared as persisted lease metadata (`checkedOutBy`, `checkedOutAt`, `checkoutNodeId`, `checkoutRunId`, `checkoutLeaseRenewedAt`, `checkoutLeaseEpoch`) through the canonical mesh sync payloads.
When a node disappears or stops renewing ownership, recovery is routed only through `MeshLeaseManager.recoverAbandonedLease(...)`. The manager releases ownership only after staleness checks pass and no active local executor session exists for the task. Recovery then bumps `checkoutLeaseEpoch`, clears owner fields, logs the abandonment reason, and returns the task to scheduler-visible work.
When a node disappears or stops renewing ownership, recovery is routed only through `MeshLeaseManager.recoverAbandonedLease(...)`. The manager now performs a two-write release: it releases the authoritative central `taskClaims` row first, then clears per-project owner fields (`checkedOutBy`, `checkoutNodeId`, `checkoutRunId`, `checkoutLeaseRenewedAt`, `checkedOutAt`) and bumps `checkoutLeaseEpoch` locally.
This fencing prevents double-claims: a restarted or delayed stale owner cannot reclaim work using older epoch state once recovery has advanced the lease generation.
If one side succeeds and the other fails, the next scheduler/self-healing tick runs `reconcileLeaseRow(taskId)` to deterministically converge local and central lease state without a side queue. Recovery/reconciliation paths emit `task:auto-recover-lease-*` run-audit events (`...-released`, `...-already-healed`, `...-foreign-owner`, `...-central-unavailable`, `...-partial-write`, `...-reconciled`) for traceability.
This fencing prevents double-claims: a restarted or delayed stale owner cannot reclaim work once central ownership has been released and lease generation has advanced.
## Registering and Managing Projects

99
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, RunAuditEventInput, Task, TaskStore } from "@fusion/core";
import type { AgentStore, CentralClaimStore, RunAuditEventInput, Task, TaskStore } from "@fusion/core";
import { MeshLeaseManager } from "../mesh-lease-manager.js";
function task(overrides: Partial<Task> = {}): Task {
@@ -173,6 +173,103 @@ describe("MeshLeaseManager", () => {
expect(recordRunAuditEvent).not.toHaveBeenCalled();
});
it("emits lease-released when central + local recovery succeeds", async () => {
const currentTask = task({ column: "in-progress" });
const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
const centralClaimStore: CentralClaimStore = {
tryClaimTask: vi.fn() as any,
renewTaskClaim: vi.fn() as any,
getTaskClaim: vi.fn().mockReturnValue(null),
releaseTaskClaim: vi.fn().mockReturnValue({ ok: true }),
};
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,
centralClaimStore,
projectId: "project-1",
agentStore: {
getAgent: vi.fn().mockResolvedValue({ lastHeartbeatAt: "2026-04-30T00:00:00.000Z" }),
} as any,
});
const ok = await manager.recoverAbandonedLease("FN-1", "stale-heartbeat");
expect(ok).toBe(true);
expect(centralClaimStore.releaseTaskClaim).toHaveBeenCalledWith({
projectId: "project-1",
taskId: "FN-1",
nodeId: "node-a",
agentId: "agent-1",
});
expect(recordRunAuditEvent.mock.calls.some((call) => call[0].mutationType === "task:auto-recover-lease-released")).toBe(true);
});
it("returns false and emits foreign-owner when central release rejects ownership", async () => {
const currentTask = task();
const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
const centralClaimStore: CentralClaimStore = {
tryClaimTask: vi.fn() as any,
renewTaskClaim: vi.fn() as any,
getTaskClaim: vi.fn().mockReturnValue(null),
releaseTaskClaim: vi.fn().mockReturnValue({
ok: false,
reason: "not_owner",
current: {
projectId: "project-1",
taskId: "FN-1",
ownerNodeId: "node-b",
ownerAgentId: "agent-2",
ownerRunId: null,
leaseEpoch: 3,
leaseRenewedAt: "2026-05-01T00:00:00.000Z",
createdAt: "2026-05-01T00:00:00.000Z",
updatedAt: "2026-05-01T00:00:00.000Z",
},
}),
};
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, centralClaimStore, projectId: "project-1" });
const ok = await manager.recoverAbandonedLease("FN-1", "stale-heartbeat");
expect(ok).toBe(false);
expect(taskStore.updateTask).not.toHaveBeenCalled();
expect(recordRunAuditEvent.mock.calls.some((call) => call[0].mutationType === "task:auto-recover-lease-foreign-owner")).toBe(true);
});
it("reconcileLeaseRow clears local owner when central claim is already gone", async () => {
const currentTask = task();
const centralClaimStore: CentralClaimStore = {
tryClaimTask: vi.fn() as any,
renewTaskClaim: vi.fn() as any,
releaseTaskClaim: vi.fn() as any,
getTaskClaim: vi.fn().mockReturnValue(null),
};
const taskStore = {
getTask: vi.fn().mockResolvedValue(currentTask),
updateTask: vi.fn().mockResolvedValue(currentTask),
moveTask: vi.fn().mockResolvedValue(currentTask),
logEntry: vi.fn().mockResolvedValue(undefined),
recordRunAuditEvent: vi.fn().mockResolvedValue(undefined),
} as unknown as TaskStore;
const manager = new MeshLeaseManager({ taskStore, centralClaimStore, projectId: "project-1" });
const ok = await manager.reconcileLeaseRow("FN-1");
expect(ok).toBe(true);
expect(taskStore.updateTask).toHaveBeenCalledWith("FN-1", expect.objectContaining({ checkedOutBy: null }));
});
it("swallows audit emission failures and still returns expected result", async () => {
const currentTask = task({ column: "todo" });
const recordRunAuditEvent = vi.fn().mockRejectedValue(new Error("boom"));

116
packages/engine/src/__tests__/reliability-interactions/lease-recovery-central-claim.test.ts Normal file
View File

@@ -0,0 +1,116 @@
import { describe, expect, it, vi } from "vitest";
import type { CentralClaimStore, Task, TaskStore } from "@fusion/core";
import { Scheduler } from "../../scheduler.js";
import { SelfHealingManager } from "../../self-healing.js";
import { MeshLeaseManager } from "../../mesh-lease-manager.js";
function makeTask(overrides: Partial<Task> = {}): Task {
return {
id: "FN-X",
description: "x",
column: "todo",
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("reliability interactions: lease recovery central claim", () => {
it("scheduler invokes reconcile once when lease recovery returns false", async () => {
const task = makeTask();
const store = {
listTasks: vi.fn().mockResolvedValue([task]),
getTask: vi.fn().mockResolvedValue(task),
updateTask: vi.fn().mockResolvedValue(task),
updateTaskStatus: vi.fn().mockResolvedValue(undefined),
logEntry: vi.fn().mockResolvedValue(undefined),
moveTask: vi.fn().mockResolvedValue(task),
getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 1, maxWorktrees: 1 }),
parseFileScopeFromPrompt: vi.fn().mockResolvedValue([]),
on: vi.fn(),
off: vi.fn(),
} as unknown as TaskStore;
const reconcileLeaseRow = vi.fn().mockResolvedValue(true);
const scheduler = new Scheduler(store, {
leaseManager: {
recoverAbandonedLease: vi.fn().mockResolvedValue(false),
reconcileLeaseRow,
} as any,
});
(scheduler as any).running = true;
await scheduler.schedule();
expect(reconcileLeaseRow).toHaveBeenCalledTimes(1);
expect(reconcileLeaseRow).toHaveBeenCalledWith("FN-X");
});
it("self-healing orphan recovery invokes reconcile once when recovery returns false", async () => {
const task = makeTask({ column: "in-progress", worktree: undefined, updatedAt: "2026-01-01T00:00:00.000Z" });
const store = {
listTasks: vi.fn().mockResolvedValue([task]),
updateTask: vi.fn().mockResolvedValue(task),
logEntry: vi.fn().mockResolvedValue(undefined),
moveTask: vi.fn().mockResolvedValue(task),
} as unknown as TaskStore;
vi.setSystemTime(new Date("2026-01-01T00:05:00.000Z"));
const reconcileLeaseRow = vi.fn().mockResolvedValue(true);
const manager = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",
getExecutingTaskIds: () => new Set<string>(),
leaseManager: {
recoverAbandonedLease: vi.fn().mockResolvedValue(false),
reconcileLeaseRow,
} as any,
});
const recovered = await manager.recoverOrphanedExecutions();
expect(recovered).toBe(1);
expect(reconcileLeaseRow).toHaveBeenCalledWith("FN-X");
manager.stop();
vi.useRealTimers();
});
it("reconciles split-brain state after central release succeeds but local update initially fails", async () => {
const current = makeTask({ column: "in-progress" });
const updateTask = vi
.fn()
.mockRejectedValueOnce(new Error("write failed"))
.mockRejectedValueOnce(new Error("write failed"))
.mockResolvedValue(current);
const taskStore = {
getTask: vi.fn().mockResolvedValue(current),
updateTask,
moveTask: vi.fn().mockResolvedValue(current),
logEntry: vi.fn().mockResolvedValue(undefined),
recordRunAuditEvent: vi.fn().mockResolvedValue(undefined),
} as unknown as TaskStore;
const centralClaimStore: CentralClaimStore = {
tryClaimTask: vi.fn() as any,
renewTaskClaim: vi.fn() as any,
getTaskClaim: vi.fn().mockReturnValue(null),
releaseTaskClaim: vi.fn().mockReturnValue({ ok: true }),
};
const manager = new MeshLeaseManager({ taskStore, centralClaimStore, projectId: "project-1" });
const recovered = await manager.recoverAbandonedLease("FN-X", "stale-heartbeat");
expect(recovered).toBe(false);
updateTask.mockResolvedValueOnce(current);
const reconciled = await manager.reconcileLeaseRow("FN-X");
expect(reconciled).toBe(true);
expect(updateTask).toHaveBeenCalled();
});
});

6
packages/engine/src/__tests__/reliability-interactions/owning-node-unavailable-interactions.test.ts
View File

@@ -220,9 +220,11 @@ describe("reliability interactions: owning-node unavailable handoff", () => {
nodeId: "node-b",
});
const store = createMockStore(task, { maxConcurrent: 1, maxWorktrees: 1, owningNodeHandoffPolicy: "reassign-to-local" });
const reconcileLeaseRow = vi.fn().mockResolvedValue(false);
const scheduler = new Scheduler(store, {
leaseManager: {
recoverAbandonedLease: vi.fn().mockResolvedValue(false),
reconcileLeaseRow,
} as any,
nodeHealthMonitor: createMockHealthMonitor({ "node-a": "online", "node-b": "online" }),
validateNodeDispatch: vi.fn().mockResolvedValue({ allowed: true }),
@@ -232,6 +234,8 @@ describe("reliability interactions: owning-node unavailable handoff", () => {
await scheduler.schedule();
expect(store.logEntry).not.toHaveBeenCalledWith(task.id, expect.stringContaining("Owning-node handoff applied"));
expect(reconcileLeaseRow).toHaveBeenCalledTimes(1);
expect(reconcileLeaseRow).toHaveBeenCalledWith(task.id);
expect(store.updateTask).toHaveBeenCalledWith(task.id, { status: "queued" });
});
@@ -250,7 +254,7 @@ describe("reliability interactions: owning-node unavailable handoff", () => {
unavailableNodePolicy: "block",
});
const scheduler = new Scheduler(store, {
leaseManager: { recoverAbandonedLease: vi.fn().mockResolvedValue(true) } as any,
leaseManager: { recoverAbandonedLease: vi.fn().mockResolvedValue(true), reconcileLeaseRow: vi.fn() } as any,
nodeHealthMonitor: createMockHealthMonitor({ "node-a": "offline", "node-b": "online" }),
validateNodeDispatch: vi.fn().mockResolvedValue({ allowed: true }),
});

34
packages/engine/src/__tests__/self-healing.test.ts
View File

@@ -5137,6 +5137,40 @@ describe("SelfHealingManager", () => {
managerWithRecovery.stop();
});
it("reconciles lease state once when abandoned-lease recovery returns false", async () => {
const getExecuting = vi.fn().mockReturnValue(new Set<string>());
const reconcileLeaseRow = vi.fn().mockResolvedValue(false);
const managerWithRecovery = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",
getExecutingTaskIds: getExecuting,
leaseManager: {
recoverAbandonedLease: vi.fn().mockResolvedValue(false),
reconcileLeaseRow,
} as any,
});
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
{
id: "FN-212",
column: "in-progress",
paused: false,
checkedOutBy: "agent-1",
worktree: undefined,
steps: [{ status: "in-progress" }],
updatedAt: "2026-01-01T00:00:00.000Z",
},
]);
vi.setSystemTime(new Date("2026-01-01T00:05:00.000Z"));
const result = await managerWithRecovery.recoverOrphanedExecutions();
expect(result).toBe(1);
expect(reconcileLeaseRow).toHaveBeenCalledTimes(1);
expect(reconcileLeaseRow).toHaveBeenCalledWith("FN-212");
managerWithRecovery.stop();
});
});
describe("recoverApprovedTriageTasks", () => {

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

@@ -1,5 +1,6 @@
import type {
AgentStore,
CentralClaimStore,
OwningNodeHandoffPolicy,
RunMutationContext,
Task,
@@ -19,6 +20,8 @@ export interface MeshLeaseManagerOptions {
getExecutingTaskIds?: () => Set<string>;
localNodeId?: string;
getHandoffPolicy?: () => Promise<OwningNodeHandoffPolicy | undefined>;
centralClaimStore?: CentralClaimStore;
projectId?: string;
}
export interface LeaseRecoveryContext {
@@ -82,6 +85,229 @@ export class MeshLeaseManager {
return { recoverable: false, reason: "owner_heartbeat_fresh" };
}
private createAuditor(task: Task) {
return createRunAuditor(this.options.taskStore, {
runId: generateSyntheticRunId("mesh-lease", task.id),
agentId: "mesh-lease-manager",
taskId: task.id,
taskLineageId: task.lineageId,
phase: "recover-unreachable-owner-lease",
});
}
private async emitLeaseAudit(
task: Task,
type:
| "task:auto-recover-lease-released"
| "task:auto-recover-lease-already-healed"
| "task:auto-recover-lease-foreign-owner"
| "task:auto-recover-lease-central-unavailable"
| "task:auto-recover-lease-partial-write"
| "task:auto-recover-lease-reconciled",
metadata: Record<string, unknown>,
): Promise<void> {
try {
await this.createAuditor(task).database({
type,
target: task.id,
metadata: {
taskId: task.id,
projectId: this.options.projectId ?? null,
...metadata,
},
});
} catch (error) {
meshLeaseManagerLog.warn(
`mesh-lease: failed to emit ${type} for taskId=${task.id}: ${error instanceof Error ? error.message : String(error)}`,
);
}
}
private async clearLocalLease(task: Task, reason: string, context: LeaseRecoveryContext, nextEpoch: number): Promise<void> {
await this.options.taskStore.updateTask(
task.id,
{
checkedOutBy: null,
checkedOutAt: null,
checkoutNodeId: null,
checkoutRunId: null,
checkoutLeaseRenewedAt: null,
checkoutLeaseEpoch: nextEpoch,
},
context.runContext,
);
await this.options.taskStore.logEntry(
task.id,
"Recovered abandoned lease",
`${reason}; epoch=${nextEpoch}`,
context.runContext,
);
if (task.column !== "todo") {
await this.options.taskStore.moveTask(task.id, "todo", {
preserveProgress:
context.preserveProgress ??
(task.currentStep > 0 || task.steps.some((step) => step.status !== "pending")),
});
}
}
private async releaseCentralClaim(task: Task, reason: string, nextEpoch: number): Promise<"released" | "already-healed" | "foreign-owner" | "unavailable"> {
const { centralClaimStore, projectId } = this.options;
if (!centralClaimStore || !projectId || !task.checkedOutBy || !task.checkoutNodeId) {
return "released";
}
const tryRelease = () =>
centralClaimStore.releaseTaskClaim({
projectId,
taskId: task.id,
nodeId: task.checkoutNodeId!,
agentId: task.checkedOutBy!,
});
try {
const released = tryRelease();
if (released.ok) {
return "released";
}
if (released.reason === "not_found") {
await this.emitLeaseAudit(task, "task:auto-recover-lease-already-healed", {
priorEpoch: task.checkoutLeaseEpoch ?? 0,
nextEpoch,
reason,
});
return "already-healed";
}
await this.emitLeaseAudit(task, "task:auto-recover-lease-foreign-owner", {
priorEpoch: task.checkoutLeaseEpoch ?? 0,
nextEpoch,
reason,
centralOwnerNodeId: released.current?.ownerNodeId ?? null,
centralOwnerAgentId: released.current?.ownerAgentId ?? null,
centralOwnerRunId: released.current?.ownerRunId ?? null,
centralLeaseEpoch: released.current?.leaseEpoch ?? null,
});
return "foreign-owner";
} catch (_error) {
await new Promise((resolve) => setTimeout(resolve, 120));
try {
const released = tryRelease();
if (released.ok) {
return "released";
}
if (released.reason === "not_found") {
await this.emitLeaseAudit(task, "task:auto-recover-lease-already-healed", {
priorEpoch: task.checkoutLeaseEpoch ?? 0,
nextEpoch,
reason,
});
return "already-healed";
}
await this.emitLeaseAudit(task, "task:auto-recover-lease-foreign-owner", {
priorEpoch: task.checkoutLeaseEpoch ?? 0,
nextEpoch,
reason,
centralOwnerNodeId: released.current?.ownerNodeId ?? null,
centralOwnerAgentId: released.current?.ownerAgentId ?? null,
centralOwnerRunId: released.current?.ownerRunId ?? null,
centralLeaseEpoch: released.current?.leaseEpoch ?? null,
});
return "foreign-owner";
} catch (retryError) {
await this.emitLeaseAudit(task, "task:auto-recover-lease-central-unavailable", {
priorEpoch: task.checkoutLeaseEpoch ?? 0,
nextEpoch,
reason,
error: retryError instanceof Error ? retryError.message : String(retryError),
});
meshLeaseManagerLog.warn(
`mesh-lease: central release unavailable for taskId=${task.id}: ${retryError instanceof Error ? retryError.message : String(retryError)}`,
);
return "unavailable";
}
}
}
async reconcileLeaseRow(taskId: string): Promise<boolean> {
const task = await this.options.taskStore.getTask(taskId);
const { centralClaimStore, projectId } = this.options;
if (!task || !centralClaimStore || !projectId) {
return false;
}
const claim = centralClaimStore.getTaskClaim(projectId, taskId);
const localHasOwner = Boolean(task.checkedOutBy || task.checkoutNodeId);
if (!claim && localHasOwner) {
const nextEpoch = (task.checkoutLeaseEpoch ?? 0) + 1;
await this.options.taskStore.updateTask(task.id, {
checkedOutBy: null,
checkedOutAt: null,
checkoutNodeId: null,
checkoutRunId: null,
checkoutLeaseRenewedAt: null,
checkoutLeaseEpoch: nextEpoch,
});
await this.emitLeaseAudit(task, "task:auto-recover-lease-reconciled", {
direction: "central-cleared->local-cleared",
priorEpoch: task.checkoutLeaseEpoch ?? 0,
nextEpoch,
});
return true;
}
if (claim && !localHasOwner) {
const status = this.options.nodeHealthMonitor?.getNodeHealth(claim.ownerNodeId);
const staleCutoff = this.staleThresholdMs();
const renewedAtMs = Date.parse(claim.leaseRenewedAt);
const staleByTime = Number.isFinite(renewedAtMs) && Date.now() - renewedAtMs > staleCutoff;
if (status === "offline" || status === "error" || staleByTime) {
const released = centralClaimStore.releaseTaskClaim({
projectId,
taskId,
nodeId: claim.ownerNodeId,
agentId: claim.ownerAgentId,
});
if (released.ok || released.reason === "not_found") {
await this.emitLeaseAudit(task, "task:auto-recover-lease-reconciled", {
direction: "local-cleared->central-cleared",
priorEpoch: task.checkoutLeaseEpoch ?? 0,
nextEpoch: task.checkoutLeaseEpoch ?? 0,
staleByTime,
ownerNodeHealth: status ?? null,
});
return true;
}
}
return false;
}
if (!claim && !localHasOwner) {
return true;
}
if (
claim &&
task.checkedOutBy === claim.ownerAgentId &&
task.checkoutNodeId === claim.ownerNodeId &&
(task.checkoutLeaseEpoch ?? 0) === claim.leaseEpoch
) {
return true;
}
await this.emitLeaseAudit(task, "task:auto-recover-lease-foreign-owner", {
priorEpoch: task.checkoutLeaseEpoch ?? 0,
nextEpoch: task.checkoutLeaseEpoch ?? 0,
reason: "split-brain-owner-mismatch",
centralOwnerNodeId: claim?.ownerNodeId ?? null,
centralOwnerAgentId: claim?.ownerAgentId ?? null,
centralLeaseEpoch: claim?.leaseEpoch ?? null,
localOwnerNodeId: task.checkoutNodeId ?? null,
localOwnerAgentId: task.checkedOutBy ?? null,
});
return false;
}
async recoverAbandonedLease(taskId: string, reason: string, context: LeaseRecoveryContext = {}): Promise<boolean> {
const task = await this.options.taskStore.getTask(taskId);
if (!task) return false;
@@ -96,13 +322,7 @@ export class MeshLeaseManager {
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 auditor = this.createAuditor(task);
const emitNodeUnreachableRecovery = async ({
decisionPath,
@@ -149,9 +369,12 @@ export class MeshLeaseManager {
}
};
let handoffPolicy: OwningNodeHandoffPolicy | undefined;
let handoffAction = "reassign-to-local";
let handoffReason = "stale_lease";
if (isUnreachableOwnerReason && task.checkoutNodeId && this.options.nodeHealthMonitor) {
const currentOwnerNodeHealth = this.options.nodeHealthMonitor.getNodeHealth(task.checkoutNodeId);
const handoffPolicy = await this.options.getHandoffPolicy?.();
handoffPolicy = await this.options.getHandoffPolicy?.();
const handoffDecision = decideOwningNodeHandoff({
task,
ownerNodeId: task.checkoutNodeId,
@@ -159,6 +382,8 @@ export class MeshLeaseManager {
localNodeId: this.options.localNodeId ?? "local",
handoffPolicy,
});
handoffAction = handoffDecision.action;
handoffReason = handoffDecision.reason;
if (handoffDecision.action === "park") {
await emitNodeUnreachableRecovery({
@@ -170,70 +395,59 @@ export class MeshLeaseManager {
handoffAction: handoffDecision.action,
handoffReason: handoffDecision.reason,
});
meshLeaseManagerLog.log(
`mesh-lease: handoff parked taskId=${task.id} reason=${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")),
});
// FN-4823/FN-4819 §2.5: without central claim store, retain local-only recovery behavior.
const nextEpoch = (task.checkoutLeaseEpoch ?? 0) + 1;
if (this.options.centralClaimStore && this.options.projectId) {
const centralResult = await this.releaseCentralClaim(task, `${reason} (${stale.reason ?? "stale"})`, nextEpoch);
if (centralResult === "foreign-owner" || centralResult === "unavailable") {
return false;
}
}
try {
await this.clearLocalLease(task, `${reason} (${stale.reason ?? "stale"})`, context, nextEpoch);
} catch (_error) {
try {
await this.clearLocalLease(task, `${reason} (${stale.reason ?? "stale"})`, context, nextEpoch);
} catch (retryError) {
if (this.options.centralClaimStore && this.options.projectId) {
await this.emitLeaseAudit(task, "task:auto-recover-lease-partial-write", {
priorEpoch: task.checkoutLeaseEpoch ?? 0,
nextEpoch,
reason,
error: retryError instanceof Error ? retryError.message : String(retryError),
});
}
return false;
}
}
if (this.options.centralClaimStore && this.options.projectId) {
await this.emitLeaseAudit(task, "task:auto-recover-lease-released", {
priorOwnerNodeId: task.checkoutNodeId ?? null,
priorOwnerAgentId: task.checkedOutBy ?? null,
priorEpoch: task.checkoutLeaseEpoch ?? 0,
nextEpoch,
reason,
handoffAction,
handoffReason,
});
}
if (isUnreachableOwnerReason) {
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;
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")),
handoffAction,
handoffReason,
});
}
return true;

6
packages/engine/src/run-audit.ts
View File

@@ -135,6 +135,12 @@ export type DatabaseMutationType =
| "task:auto-recover-finalize-already-on-main"
| "task:auto-recover-branch-misbound"
| "task:auto-recover-node-unreachable"
| "task:auto-recover-lease-released"
| "task:auto-recover-lease-already-healed"
| "task:auto-recover-lease-foreign-owner"
| "task:auto-recover-lease-central-unavailable"
| "task:auto-recover-lease-partial-write"
| "task:auto-recover-lease-reconciled"
| "task:auto-recover-completion-fanout"
| "task:auto-recover-worktree-session-exhausted"
| "task:auto-recover-starved-refinement"

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

@@ -311,7 +311,9 @@ export class InProcessRuntime
this.leaseManager = new MeshLeaseManager({
taskStore: this.taskStore,
agentStore: this.agentStore,
getHandoffPolicy: () => this.taskStore.getSettings().then((settings) => settings.owningNodeHandoffPolicy),
getExecutingTaskIds: () => this.executor?.getExecutingTaskIds() ?? new Set<string>(),
projectId: this.config.projectId,
});
const autoClaimSnapshotManager = new AutoClaimSnapshotManager({ taskStore: this.taskStore });

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

@@ -870,6 +870,7 @@ export class Scheduler {
{ preserveProgress: true },
);
if (!recovered) {
await this.options.leaseManager.reconcileLeaseRow(task.id);
await this.store.updateTask(task.id, { status: "queued" });
await this.logDispatchQueuedReason(task.id, "queued — checkout lease recovery blocked dispatch");
continue;

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

@@ -3973,6 +3973,7 @@ export class SelfHealingManager {
recovered++;
continue;
}
await this.options.leaseManager.reconcileLeaseRow(task.id);
}
// Reset steps whose work was never committed before clearing the worktree