feat(FN-4813): complete Step 2 — add distributed claim mutex tests

Fusion-Task-Id: FN-4813
Fusion-Task-Lineage: 846893a5-2afa-4817-8f64-8c444d2fd713

packages/core/src/__tests__/checkout-claim-mutex.test.ts

@@ -0,0 +1,74 @@
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { mkdtempSync } from "node:fs";
+import { rm } from "node:fs/promises";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+import { AgentStore } from "../agent-store.js";
+import { TaskStore } from "../store.js";
+import { CheckoutConflictError } from "../types.js";
+
+function makeTmpDir(): string {
+  return mkdtempSync(join(tmpdir(), "fn-checkout-claim-test-"));
+}
+
+describe("checkout claim mutex", () => {
+  let rootDir: string;
+  let taskStore: TaskStore;
+  let agentStore: AgentStore;
+  let globalDir: string;
+  let taskId: string;
+  let agentA: string;
+  let agentB: string;
+
+  beforeEach(async () => {
+    rootDir = makeTmpDir();
+    globalDir = join(rootDir, ".fusion-global");
+    taskStore = new TaskStore(rootDir, globalDir, { inMemoryDb: true });
+    await taskStore.init();
+    agentStore = new AgentStore({ rootDir, inMemoryDb: true, taskStore });
+    await agentStore.init();
+
+    agentA = (await agentStore.createAgent({ name: "A", role: "executor" })).id;
+    agentB = (await agentStore.createAgent({ name: "B", role: "executor" })).id;
+    taskId = (await taskStore.createTask({ description: "claim me" })).id;
+  });
+
+  afterEach(async () => {
+    agentStore?.close();
+    taskStore?.close();
+    await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
+  });
+
+  it("first claimant wins and epoch becomes 1", async () => {
+    const claimed = await agentStore.checkoutTask(agentA, taskId, { nodeId: "node-a", runId: "run-1" });
+    expect(claimed.checkedOutBy).toBe(agentA);
+    expect(claimed.checkoutNodeId).toBe("node-a");
+    expect(claimed.checkoutLeaseEpoch).toBe(1);
+  });
+
+  it("different agent claim conflicts and preserves owner", async () => {
+    await agentStore.checkoutTask(agentA, taskId, { nodeId: "node-a", runId: "run-1" });
+    await expect(agentStore.checkoutTask(agentB, taskId, { nodeId: "node-b", runId: "run-2" })).rejects.toBeInstanceOf(CheckoutConflictError);
+    const current = await taskStore.getTask(taskId);
+    expect(current?.checkedOutBy).toBe(agentA);
+    expect(current?.checkoutNodeId).toBe("node-a");
+  });
+
+  it("same agent on different node conflicts", async () => {
+    await agentStore.checkoutTask(agentA, taskId, { nodeId: "node-a", runId: "run-1" });
+    await expect(agentStore.checkoutTask(agentA, taskId, { nodeId: "node-b", runId: "run-2", leaseEpoch: 1 })).rejects.toBeInstanceOf(CheckoutConflictError);
+  });
+
+  it("renewal with matching epoch succeeds and does not bump epoch", async () => {
+    await agentStore.checkoutTask(agentA, taskId, { nodeId: "node-a", runId: "run-1" });
+    const renewed = await agentStore.checkoutTask(agentA, taskId, { nodeId: "node-a", runId: "run-2", leaseEpoch: 1, renewedAt: "2026-05-16T00:00:00.000Z" });
+    expect(renewed.checkoutLeaseEpoch).toBe(1);
+    expect(renewed.checkoutRunId).toBe("run-2");
+    expect(renewed.checkoutLeaseRenewedAt).toBe("2026-05-16T00:00:00.000Z");
+  });
+
+  it("renewal with stale epoch conflicts", async () => {
+    await agentStore.checkoutTask(agentA, taskId, { nodeId: "node-a", runId: "run-1" });
+    await expect(agentStore.checkoutTask(agentA, taskId, { nodeId: "node-a", runId: "run-2", leaseEpoch: 0 })).rejects.toBeInstanceOf(CheckoutConflictError);
+  });
+});

packages/core/src/agent-store.ts

@@ -52,15 +52,8 @@ import {
   getLegacyAgentInstructionsBundleDirName,
   getSafeAgentAssetIdSegment,
 } from "./types.js";
-import type { RunMutationContext } from "./types.js";
+import type { CheckoutClaimContext, 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, canAgentTakeImplementationTaskForExplicitRouting, formatRoleMismatchReason } from "./agent-role-policy.js";
 import { normalizeAgentPermissionPolicy, resolveEffectiveAgentPermissionPolicy } from "./agent-permission-policy.js";
@@ -1378,7 +1371,7 @@ export class AgentStore extends EventEmitter {
   async checkoutTask(
     agentId: string,
     taskId: string,
-    leaseContext?: CheckoutLeaseContext,
+    leaseContext?: CheckoutClaimContext,
     runContext?: RunMutationContext,
   ): Promise<Task> {
     if (!this.taskStore) {
@@ -1399,11 +1392,51 @@ export class AgentStore extends EventEmitter {
       throw new CheckoutConflictError(taskId, task.checkedOutBy, agentId);
     }
 
-    const nextEpoch = leaseContext?.leaseEpoch ?? task.checkoutLeaseEpoch ?? 0;
     const nextRenewedAt = leaseContext?.renewedAt ?? new Date().toISOString();
-    const existingNodeId = task.checkoutNodeId;
+    const existingNodeId = task.checkoutNodeId ?? null;
     const existingEpoch = task.checkoutLeaseEpoch ?? 0;
+    const tryClaimCheckout = "tryClaimCheckout" in this.taskStore
+      ? (this.taskStore as TaskStore & {
+        tryClaimCheckout: NonNullable<TaskStore["tryClaimCheckout"]>;
+      }).tryClaimCheckout
+      : undefined;
 
+    if (tryClaimCheckout) {
+      const isSameAgentHolder = task.checkedOutBy === agentId;
+      const isRenewal = isSameAgentHolder
+        && existingNodeId !== null
+        && leaseContext?.nodeId === existingNodeId
+        && leaseContext?.leaseEpoch === existingEpoch;
+
+      if (isSameAgentHolder && !isRenewal) {
+        throw new CheckoutConflictError(taskId, agentId, agentId);
+      }
+
+      const result = await tryClaimCheckout.call(this.taskStore, taskId, {
+        agentId,
+        nodeId: leaseContext?.nodeId ?? existingNodeId ?? "",
+        runId: leaseContext?.runId ?? task.checkoutRunId ?? null,
+        renewedAt: nextRenewedAt,
+        leaseEpoch: isRenewal ? existingEpoch : existingEpoch + 1,
+      }, {
+        expectedCheckedOutBy: task.checkedOutBy ?? null,
+        expectedNodeId: existingNodeId,
+        expectedLeaseEpoch: existingEpoch,
+      });
+
+      if (!result.ok) {
+        if (result.reason === "row_not_found") {
+          throw new Error(`Task ${taskId} not found`);
+        }
+        const currentHolder = result.current?.checkedOutBy;
+        throw new CheckoutConflictError(taskId, currentHolder ?? "unknown", agentId);
+      }
+
+      await this.taskStore.logEntry(taskId, `Checked out by agent ${agentId}`, undefined, runContext);
+      return result.task;
+    }
+
+    const nextEpoch = leaseContext?.leaseEpoch ?? task.checkoutLeaseEpoch ?? 0;
     if (
       task.checkedOutBy === agentId
       && existingNodeId === (leaseContext?.nodeId ?? existingNodeId)

packages/core/src/store.ts

@@ -3,7 +3,7 @@ import { randomUUID } from "node:crypto";
 import { mkdir, readdir, readFile, writeFile, rename, unlink } from "node:fs/promises";
 import { join } from "node:path";
 import { existsSync, watch, type FSWatcher } from "node:fs";
-import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, BoardConfig, Column, MergeResult, Settings, GlobalSettings, ProjectSettings, ActivityLogEntry, ActivityEventType, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, InboxTask, TaskLogEntry, RunMutationContext, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, ArchivedTaskEntry, ArchiveAgentLogMode, TaskPriority, SourceType, WorkflowStepTemplate, Agent, AutostashOrphanRecord, TaskCommitAssociation, TaskCommitAssociationMatchSource, TaskCommitAssociationConfidence, GithubIssueAction } from "./types.js";
+import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, BoardConfig, Column, CheckoutClaimPrecondition, MergeResult, Settings, GlobalSettings, ProjectSettings, ActivityLogEntry, ActivityEventType, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, InboxTask, TaskLogEntry, RunMutationContext, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, ArchivedTaskEntry, ArchiveAgentLogMode, TaskPriority, SourceType, WorkflowStepTemplate, Agent, AutostashOrphanRecord, TaskCommitAssociation, TaskCommitAssociationMatchSource, TaskCommitAssociationConfidence, GithubIssueAction } from "./types.js";
 import { createActivityLogSnapshot, createRunAuditSnapshot, createTaskMetadataSnapshot, toTaskMetadataRecord, validateSnapshotEnvelope, type ActivityLogSnapshot, type RunAuditSnapshot, type TaskMetadataSnapshot } from "./shared-mesh-state.js";
 import { VALID_TRANSITIONS, DEFAULT_SETTINGS, isGlobalOnlySettingsKey, WORKFLOW_STEP_TEMPLATES, validateDocumentKey } from "./types.js";
 import { DEFAULT_PROJECT_SETTINGS } from "./settings-schema.js";
@@ -3800,6 +3800,60 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
     return rows.map((row) => this.rowToTask(row));
   }
 
+  async tryClaimCheckout(
+    taskId: string,
+    claim: {
+      agentId: string;
+      nodeId: string;
+      runId: string | null;
+      leaseEpoch: number;
+      renewedAt: string;
+    },
+    precondition: CheckoutClaimPrecondition,
+  ): Promise<{ ok: true; task: Task } | { ok: false; reason: "row_not_found" | "precondition_failed"; current: Task | null }> {
+    const current = await this.getTask(taskId);
+    if (!current) {
+      return { ok: false, reason: "row_not_found", current: null };
+    }
+
+    const updateResult = this.db.prepare(`
+      UPDATE tasks
+      SET
+        checkedOutBy = ?,
+        checkedOutAt = COALESCE(checkedOutAt, ?),
+        checkoutNodeId = ?,
+        checkoutRunId = ?,
+        checkoutLeaseRenewedAt = ?,
+        checkoutLeaseEpoch = ?
+      WHERE id = ?
+        AND COALESCE(checkedOutBy, '') = COALESCE(?, '')
+        AND COALESCE(checkoutNodeId, '') = COALESCE(?, '')
+        AND COALESCE(checkoutLeaseEpoch, 0) = COALESCE(?, 0)
+    `).run(
+      claim.agentId,
+      new Date().toISOString(),
+      claim.nodeId,
+      claim.runId,
+      claim.renewedAt,
+      claim.leaseEpoch,
+      taskId,
+      precondition.expectedCheckedOutBy ?? null,
+      precondition.expectedNodeId ?? null,
+      precondition.expectedLeaseEpoch ?? 0,
+    ) as { changes: number };
+
+    const post = await this.getTask(taskId);
+    if (updateResult.changes === 0) {
+      return { ok: false, reason: "precondition_failed", current: post };
+    }
+
+    if (!post) {
+      return { ok: false, reason: "row_not_found", current: null };
+    }
+
+    return { ok: true, task: post };
+  }
+
   async selectNextTaskForAgent(
     agentId: string,
     agent?: Pick<Agent, "id" | "role">,

packages/core/src/types.ts

@@ -1242,6 +1242,24 @@ export interface CheckoutLease {
   checkedOutAt: string;
 }
 
+export interface CheckoutClaimContext {
+  /** Node identity for the claimant. */
+  nodeId: string;
+  /** Owning run/session ID when known. */
+  runId?: string;
+  /** Expected current lease epoch for renewal operations. */
+  leaseEpoch?: number;
+  /** ISO-8601 timestamp for lease-renewed heartbeat updates. */
+  renewedAt?: string;
+}
+
+export interface CheckoutClaimPrecondition {
+  /** Null/undefined means expecting an unclaimed row. */
+  expectedCheckedOutBy?: string | null;
+  expectedNodeId?: string | null;
+  expectedLeaseEpoch?: number | null;
+}
+
 /**
  * Durable task-level aggregate token usage totals persisted on the task row.
  *