test(FN-4822): complete Step 4 — add AgentStore central-claim wiring tests

Fusion-Task-Id: FN-4822
Fusion-Task-Lineage: 08cc29e8-114a-48dc-80de-8d7fd2ce0e69

packages/core/src/__tests__/agent-store-central-claim.test.ts

@@ -0,0 +1,105 @@
+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 { createCentralDatabase, type CentralDatabase } from "../central-db.js";
+import { TaskStore } from "../store.js";
+import { CheckoutConflictError } from "../types.js";
+
+function makeTmpDir(): string {
+  return mkdtempSync(join(tmpdir(), "fn-agent-central-claim-test-"));
+}
+
+describe("AgentStore central claim wiring", () => {
+  let rootDir: string;
+  let globalDir: string;
+  let taskStore: TaskStore;
+  let centralDb: CentralDatabase;
+  let agentStoreA: AgentStore;
+  let agentStoreB: AgentStore;
+  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();
+    centralDb = createCentralDatabase(globalDir);
+    centralDb.init();
+
+    agentStoreA = new AgentStore({ rootDir, inMemoryDb: true, taskStore, claimStore: centralDb, projectId: "P-1", nodeId: "node-a" });
+    agentStoreB = new AgentStore({ rootDir, inMemoryDb: true, taskStore, claimStore: centralDb, projectId: "P-1", nodeId: "node-b" });
+    await agentStoreA.init();
+    await agentStoreB.init();
+
+    agentA = (await agentStoreA.createAgent({ name: "A", role: "executor" })).id;
+    agentB = (await agentStoreB.createAgent({ name: "B", role: "executor" })).id;
+    taskId = (await taskStore.createTask({ description: "claim me" })).id;
+  });
+
+  afterEach(async () => {
+    agentStoreA?.close();
+    agentStoreB?.close();
+    taskStore?.close();
+    centralDb?.close();
+    await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
+  });
+
+  it("successful claim writes central row and per-project mirror", async () => {
+    const claimed = await agentStoreA.checkoutTask(agentA, taskId, { runId: "run-1" });
+    const central = centralDb.getTaskClaim("P-1", taskId);
+    expect(central).toBeTruthy();
+    expect(central?.ownerAgentId).toBe(agentA);
+    expect(central?.ownerNodeId).toBe("node-a");
+    expect(central?.leaseEpoch).toBe(claimed.checkoutLeaseEpoch);
+    expect(claimed.checkoutNodeId).toBe(central?.ownerNodeId);
+  });
+
+  it("conflict uses central holder even when project row is stale", async () => {
+    await agentStoreA.checkoutTask(agentA, taskId, { runId: "run-1" });
+    await taskStore.updateTask(taskId, {
+      checkedOutBy: null,
+      checkedOutAt: null,
+      checkoutNodeId: null,
+      checkoutRunId: null,
+      checkoutLeaseRenewedAt: null,
+      checkoutLeaseEpoch: null,
+    });
+
+    await expect(agentStoreB.checkoutTask(agentB, taskId, { runId: "run-2" })).rejects.toMatchObject({
+      name: "CheckoutConflictError",
+      currentHolderId: agentA,
+    } satisfies Partial<CheckoutConflictError>);
+  });
+
+  it("renewal by same owner does not bump epoch", async () => {
+    await agentStoreA.checkoutTask(agentA, taskId, { runId: "run-1" });
+    const before = centralDb.getTaskClaim("P-1", taskId);
+    const renewed = await agentStoreA.checkoutTask(agentA, taskId, { runId: "run-2", leaseEpoch: before?.leaseEpoch, renewedAt: "2026-05-16T00:00:00.000Z" });
+    const after = centralDb.getTaskClaim("P-1", taskId);
+    expect(before?.leaseEpoch).toBe(1);
+    expect(after?.leaseEpoch).toBe(before?.leaseEpoch);
+    expect(renewed.checkoutLeaseEpoch).toBe(before?.leaseEpoch);
+  });
+
+  it("owner release clears central row and next owner reclaims at epoch 1", async () => {
+    await agentStoreA.checkoutTask(agentA, taskId, { runId: "run-1" });
+    await agentStoreA.releaseTask(agentA, taskId);
+    expect(centralDb.getTaskClaim("P-1", taskId)).toBeNull();
+
+    const claimedByB = await agentStoreB.checkoutTask(agentB, taskId, { runId: "run-2" });
+    expect(claimedByB.checkedOutBy).toBe(agentB);
+    expect(claimedByB.checkoutLeaseEpoch).toBe(1);
+    expect(centralDb.getTaskClaim("P-1", taskId)?.leaseEpoch).toBe(1);
+  });
+
+  it("constructor throws when claimStore is provided without projectId", () => {
+    expect(() => new AgentStore({ rootDir, inMemoryDb: true, taskStore, claimStore: centralDb })).toThrow(
+      "AgentStore requires projectId when claimStore is configured",
+    );
+  });
+});

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

@@ -0,0 +1,101 @@
+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 { createCentralDatabase, type CentralDatabase } from "../central-db.js";
+
+function makeTmpDir(): string {
+  return mkdtempSync(join(tmpdir(), "fn-central-claim-test-"));
+}
+
+describe("central claim mutex", () => {
+  let globalDir: string;
+  let db: CentralDatabase;
+
+  beforeEach(() => {
+    globalDir = makeTmpDir();
+    db = createCentralDatabase(globalDir);
+    db.init();
+  });
+
+  afterEach(async () => {
+    db.close();
+    await rm(globalDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
+  });
+
+  it("first claim creates epoch 1 row", () => {
+    const result = db.tryClaimTask({
+      projectId: "P-1",
+      taskId: "FN-1",
+      nodeId: "node-a",
+      agentId: "agent-a",
+      runId: "run-1",
+      renewedAt: "2026-05-16T00:00:00.000Z",
+    });
+    expect(result.ok).toBe(true);
+    if (!result.ok) return;
+    expect(result.claim.leaseEpoch).toBe(1);
+    expect(result.claim.ownerAgentId).toBe("agent-a");
+    expect(result.claim.ownerNodeId).toBe("node-a");
+  });
+
+  it("different owner without expectedEpoch conflicts and does not mutate", () => {
+    db.tryClaimTask({ projectId: "P-1", taskId: "FN-1", nodeId: "node-a", agentId: "agent-a", runId: "run-1", renewedAt: "2026-05-16T00:00:00.000Z" });
+    const conflict = db.tryClaimTask({ projectId: "P-1", taskId: "FN-1", nodeId: "node-b", agentId: "agent-b", runId: "run-2", renewedAt: "2026-05-16T00:01:00.000Z" });
+    expect(conflict.ok).toBe(false);
+    if (conflict.ok) return;
+    expect(conflict.reason).toBe("conflict");
+    expect(conflict.current.ownerAgentId).toBe("agent-a");
+    const row = db.getTaskClaim("P-1", "FN-1");
+    expect(row?.ownerAgentId).toBe("agent-a");
+    expect(row?.leaseEpoch).toBe(1);
+  });
+
+  it("owner-change with matching expectedEpoch increments exactly by one", () => {
+    db.tryClaimTask({ projectId: "P-1", taskId: "FN-1", nodeId: "node-a", agentId: "agent-a", runId: "run-1", renewedAt: "2026-05-16T00:00:00.000Z" });
+    const changed = db.tryClaimTask({ projectId: "P-1", taskId: "FN-1", nodeId: "node-b", agentId: "agent-b", runId: "run-2", renewedAt: "2026-05-16T00:01:00.000Z", expectedEpoch: 1 });
+    expect(changed.ok).toBe(true);
+    if (!changed.ok) return;
+    expect(changed.claim.leaseEpoch).toBe(2);
+    expect(changed.claim.ownerAgentId).toBe("agent-b");
+  });
+
+  it("renew with matching expectedEpoch preserves epoch and updates renewedAt", () => {
+    db.tryClaimTask({ projectId: "P-1", taskId: "FN-1", nodeId: "node-a", agentId: "agent-a", runId: "run-1", renewedAt: "2026-05-16T00:00:00.000Z" });
+    const renewed = db.renewTaskClaim({ projectId: "P-1", taskId: "FN-1", nodeId: "node-a", agentId: "agent-a", runId: "run-2", renewedAt: "2026-05-16T00:02:00.000Z", expectedEpoch: 1 });
+    expect(renewed.ok).toBe(true);
+    if (!renewed.ok) return;
+    expect(renewed.claim.leaseEpoch).toBe(1);
+    expect(renewed.claim.ownerRunId).toBe("run-2");
+    expect(renewed.claim.leaseRenewedAt).toBe("2026-05-16T00:02:00.000Z");
+  });
+
+  it("renew with stale expectedEpoch conflicts", () => {
+    db.tryClaimTask({ projectId: "P-1", taskId: "FN-1", nodeId: "node-a", agentId: "agent-a", runId: "run-1", renewedAt: "2026-05-16T00:00:00.000Z" });
+    const renewed = db.renewTaskClaim({ projectId: "P-1", taskId: "FN-1", nodeId: "node-a", agentId: "agent-a", runId: "run-2", renewedAt: "2026-05-16T00:02:00.000Z", expectedEpoch: 0 });
+    expect(renewed.ok).toBe(false);
+    if (renewed.ok) return;
+    expect(renewed.reason).toBe("conflict");
+  });
+
+  it("release succeeds for owner and not_owner for other agent", () => {
+    db.tryClaimTask({ projectId: "P-1", taskId: "FN-1", nodeId: "node-a", agentId: "agent-a", runId: "run-1", renewedAt: "2026-05-16T00:00:00.000Z" });
+    const notOwner = db.releaseTaskClaim({ projectId: "P-1", taskId: "FN-1", nodeId: "node-b", agentId: "agent-b" });
+    expect(notOwner.ok).toBe(false);
+    if (!notOwner.ok) {
+      expect(notOwner.reason).toBe("not_owner");
+      expect(notOwner.current?.ownerAgentId).toBe("agent-a");
+    }
+    const released = db.releaseTaskClaim({ projectId: "P-1", taskId: "FN-1", nodeId: "node-a", agentId: "agent-a" });
+    expect(released).toEqual({ ok: true });
+  });
+
+  it("getTaskClaim returns full row before release and null after", () => {
+    db.tryClaimTask({ projectId: "P-1", taskId: "FN-1", nodeId: "node-a", agentId: "agent-a", runId: "run-1", renewedAt: "2026-05-16T00:00:00.000Z" });
+    const before = db.getTaskClaim("P-1", "FN-1");
+    expect(before).toMatchObject({ projectId: "P-1", taskId: "FN-1", ownerAgentId: "agent-a", ownerNodeId: "node-a", leaseEpoch: 1 });
+    db.releaseTaskClaim({ projectId: "P-1", taskId: "FN-1", nodeId: "node-a", agentId: "agent-a" });
+    expect(db.getTaskClaim("P-1", "FN-1")).toBeNull();
+  });
+});

packages/core/src/agent-store.ts

@@ -52,7 +52,7 @@ import {
   getLegacyAgentInstructionsBundleDirName,
   getSafeAgentAssetIdSegment,
 } from "./types.js";
-import type { CheckoutClaimContext, RunMutationContext } from "./types.js";
+import type { CentralClaimStore, CheckoutClaimContext, RunMutationContext } from "./types.js";
 import type { TaskStore } from "./store.js";
 import { computeAccessState } from "./agent-permissions.js";
 import { canAgentTakeImplementationTask, canAgentTakeImplementationTaskForExplicitRouting, formatRoleMismatchReason } from "./agent-role-policy.js";
@@ -102,6 +102,12 @@ export interface AgentStoreOptions {
   rootDir?: string;
   /** Optional TaskStore for checkout/release operations */
   taskStore?: TaskStore;
+  /** Optional authoritative central claim store for cross-node checkout mutex. */
+  claimStore?: CentralClaimStore;
+  /** Project ID for central claim ownership rows (required when claimStore is set). */
+  projectId?: string;
+  /** Optional default nodeId when checkout leaseContext omits one. */
+  nodeId?: string;
   /**
    * Test-only: open the underlying SQLite DB as `:memory:` instead of a
    * disk-backed file. Skips per-test fsync and WAL setup; mirrors the
@@ -220,6 +226,9 @@ export class AgentStore extends EventEmitter {
   private locks: Map<string, AgentLock> = new Map();
   private _db: Database | null = null;
   private taskStore?: TaskStore;
+  private readonly claimStore?: CentralClaimStore;
+  private readonly claimProjectId?: string;
+  private readonly defaultNodeId?: string;
   private readonly inMemoryDb: boolean;
 
   constructor(options: AgentStoreOptions = {}) {
@@ -234,6 +243,12 @@ export class AgentStore extends EventEmitter {
     this.rootDir = options.rootDir ?? resolve(".fusion");
     this.agentsDir = join(this.rootDir, "agents");
     this.taskStore = options.taskStore;
+    this.claimStore = options.claimStore;
+    this.claimProjectId = options.projectId;
+    this.defaultNodeId = options.nodeId;
+    if (this.claimStore && !this.claimProjectId) {
+      throw new Error("AgentStore requires projectId when claimStore is configured");
+    }
     this.inMemoryDb = options.inMemoryDb === true;
   }
 
@@ -1402,6 +1417,75 @@ export class AgentStore extends EventEmitter {
       : undefined;
 
     if (tryClaimCheckout) {
+      const requestNodeId = this.claimStore
+        ? (leaseContext?.nodeId ?? this.defaultNodeId)
+        : (leaseContext?.nodeId ?? existingNodeId ?? "");
+      if (this.claimStore && !requestNodeId) {
+        throw new Error("checkoutTask requires leaseContext.nodeId or AgentStore nodeId when claimStore is configured");
+      }
+
+      if (this.claimStore && this.claimProjectId) {
+        const isSameAgentHolder = task.checkedOutBy === agentId;
+        const isSameNodeHolder = existingNodeId !== null && requestNodeId === existingNodeId;
+        const expectedEpoch = isSameAgentHolder && isSameNodeHolder ? existingEpoch : undefined;
+
+        const centralResult = this.claimStore.tryClaimTask({
+          projectId: this.claimProjectId,
+          taskId,
+          nodeId: requestNodeId,
+          agentId,
+          runId: leaseContext?.runId ?? task.checkoutRunId ?? null,
+          renewedAt: nextRenewedAt,
+          expectedEpoch,
+        });
+
+        if (!centralResult.ok) {
+          throw new CheckoutConflictError(taskId, centralResult.current.ownerAgentId, agentId);
+        }
+
+        const claim = centralResult.claim;
+        const mirrorLease = {
+          agentId,
+          nodeId: claim.ownerNodeId,
+          runId: claim.ownerRunId,
+          renewedAt: claim.leaseRenewedAt,
+          leaseEpoch: claim.leaseEpoch,
+        };
+        const mirrorPrecondition = {
+          expectedCheckedOutBy: task.checkedOutBy ?? null,
+          expectedNodeId: existingNodeId,
+          expectedLeaseEpoch: existingEpoch,
+        };
+
+        let result = await tryClaimCheckout.call(this.taskStore, taskId, mirrorLease, mirrorPrecondition);
+        if (!result.ok) {
+          const mirrorTask = await this.taskStore.getTask(taskId);
+          if (mirrorTask) {
+            result = await tryClaimCheckout.call(this.taskStore, taskId, mirrorLease, {
+              expectedCheckedOutBy: mirrorTask.checkedOutBy ?? null,
+              expectedNodeId: mirrorTask.checkoutNodeId ?? null,
+              expectedLeaseEpoch: mirrorTask.checkoutLeaseEpoch ?? 0,
+            });
+          }
+          if (!result.ok) {
+            await this.taskStore.logEntry(
+              taskId,
+              "Warning: central checkout claim succeeded but per-project mirror update failed",
+              undefined,
+              runContext,
+            );
+            const latestTask = await this.taskStore.getTask(taskId);
+            if (!latestTask) {
+              throw new Error(`Task ${taskId} not found`);
+            }
+            return latestTask;
+          }
+        }
+
+        await this.taskStore.logEntry(taskId, `Checked out by agent ${agentId}`, undefined, runContext);
+        return result.task;
+      }
+
       const isSameAgentHolder = task.checkedOutBy === agentId;
       const isRenewal = isSameAgentHolder
         && existingNodeId !== null
@@ -1481,6 +1565,15 @@ export class AgentStore extends EventEmitter {
       return task;
     }
 
+    if (this.claimStore && this.claimProjectId && task.checkoutNodeId) {
+      this.claimStore.releaseTaskClaim({
+        projectId: this.claimProjectId,
+        taskId,
+        nodeId: task.checkoutNodeId,
+        agentId,
+      });
+    }
+
     const updated = await this.taskStore.updateTask(taskId, {
       checkedOutBy: null,
       checkedOutAt: null,