feat(FN-4132): sync task ownership on agent reassignment

Adds a task reassignment sync helper to the core store, wires it into the system, and ships regression tests; includes a patch changeset for `@runfusion/fusion`.

Fusion-Task-Id: FN-4132

Fusion-Task-Lineage: 80ec3c62-7c09-49dd-9e49-74b203e48bf4

.changeset/fn-4132-agent-task-reassignment-sync.md

@@ -0,0 +1,5 @@
+---
+"@runfusion/fusion": patch
+---
+
+Recover agent/task reassignment sync from upstream PR #58 (author: HarryCordewener). TaskStore.updateTask now keeps agents.taskId aligned with task.assignedAgentId, clears stale checkout leases held by the outgoing agent, and protects against races where the outgoing agent has already moved on.

packages/core/src/__tests__/store-persistence.test.ts

@@ -1,4 +1,6 @@
 import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { AgentStore } from "../agent-store.js";
+import { TaskStore } from "../store.js";
 import { createTaskStoreTestHarness } from "./store-test-helpers.js";
 
 describe("TaskStore", () => {
@@ -57,6 +59,133 @@ describe("TaskStore", () => {
     });
   });
 
+  describe("agent taskId sync on reassignment", () => {
+    it("reassignment clears the old agent taskId and sets the new agent taskId", async () => {
+      harness.store().close();
+      const store = new TaskStore(harness.rootDir(), harness.globalDir());
+      await store.init();
+      const agentStore = new AgentStore({ rootDir: store.getFusionDir() });
+      await agentStore.init();
+
+      try {
+        const task = await store.createTask({ description: "Reassignment target" });
+        const agentA = await agentStore.createAgent({ name: "Agent A", role: "executor" });
+        const agentB = await agentStore.createAgent({ name: "Agent B", role: "executor" });
+
+        await store.updateTask(task.id, { assignedAgentId: agentA.id });
+        expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task.id);
+
+        await store.updateTask(task.id, { assignedAgentId: agentB.id });
+
+        expect((await agentStore.getAgent(agentA.id))?.taskId).toBeUndefined();
+        expect((await agentStore.getAgent(agentB.id))?.taskId).toBe(task.id);
+      } finally {
+        agentStore.close();
+        store.close();
+      }
+    });
+
+    it("reassignment clears stale checkedOutBy when the outgoing agent held the lease", async () => {
+      harness.store().close();
+      const store = new TaskStore(harness.rootDir(), harness.globalDir());
+      await store.init();
+      const agentStore = new AgentStore({ rootDir: store.getFusionDir(), taskStore: store });
+      await agentStore.init();
+
+      try {
+        const task = await store.createTask({ description: "Checkout cleanup target" });
+        const agentA = await agentStore.createAgent({ name: "Agent A Checkout", role: "executor" });
+        const agentB = await agentStore.createAgent({ name: "Agent B Checkout", role: "executor" });
+
+        await store.updateTask(task.id, { assignedAgentId: agentA.id });
+        await agentStore.checkoutTask(agentA.id, task.id);
+        expect((await store.getTask(task.id)).checkedOutBy).toBe(agentA.id);
+
+        const updated = await store.updateTask(task.id, { assignedAgentId: agentB.id });
+        expect(updated.checkedOutBy).toBeUndefined();
+      } finally {
+        agentStore.close();
+        store.close();
+      }
+    });
+
+    it("unassignment clears the agent taskId", async () => {
+      harness.store().close();
+      const store = new TaskStore(harness.rootDir(), harness.globalDir());
+      await store.init();
+      const agentStore = new AgentStore({ rootDir: store.getFusionDir() });
+      await agentStore.init();
+
+      try {
+        const task = await store.createTask({ description: "Unassign target" });
+        const agent = await agentStore.createAgent({ name: "Sole Agent", role: "executor" });
+
+        await store.updateTask(task.id, { assignedAgentId: agent.id });
+        expect((await agentStore.getAgent(agent.id))?.taskId).toBe(task.id);
+
+        await store.updateTask(task.id, { assignedAgentId: null });
+
+        expect((await agentStore.getAgent(agent.id))?.taskId).toBeUndefined();
+      } finally {
+        agentStore.close();
+        store.close();
+      }
+    });
+
+    it("re-setting the same agent id is a no-op for agent task links", async () => {
+      harness.store().close();
+      const store = new TaskStore(harness.rootDir(), harness.globalDir());
+      await store.init();
+      const agentStore = new AgentStore({ rootDir: store.getFusionDir() });
+      await agentStore.init();
+
+      try {
+        const task = await store.createTask({ description: "Idempotent target" });
+        const agentA = await agentStore.createAgent({ name: "Agent Same", role: "executor" });
+
+        await store.updateTask(task.id, { assignedAgentId: agentA.id });
+        expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task.id);
+
+        await store.updateTask(task.id, { assignedAgentId: agentA.id });
+
+        expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task.id);
+      } finally {
+        agentStore.close();
+        store.close();
+      }
+    });
+
+    it("does not clear the outgoing agent taskId when it already moved to another task", async () => {
+      harness.store().close();
+      const store = new TaskStore(harness.rootDir(), harness.globalDir());
+      await store.init();
+      const agentStore = new AgentStore({ rootDir: store.getFusionDir() });
+      await agentStore.init();
+
+      try {
+        const task1 = await store.createTask({ description: "Race guard task 1" });
+        const task2 = await store.createTask({ description: "Race guard task 2" });
+        const agentA = await agentStore.createAgent({ name: "Agent Race A", role: "executor" });
+        const agentB = await agentStore.createAgent({ name: "Agent Race B", role: "executor" });
+
+        await store.updateTask(task1.id, { assignedAgentId: agentA.id });
+        expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task1.id);
+        expect((await agentStore.getAgent(agentB.id))?.taskId).toBeUndefined();
+
+        await agentStore.syncExecutionTaskLink(agentA.id, task2.id);
+        expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task2.id);
+
+        await store.updateTask(task1.id, { assignedAgentId: agentB.id });
+
+        expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task2.id);
+        expect((await agentStore.getAgent(agentB.id))?.taskId).toBe(task1.id);
+      } finally {
+        agentStore.close();
+        store.close();
+      }
+    });
+  });
+
   describe("pausedByAgentId persistence", () => {
     it("creates and lists a task with pausedByAgentId", async () => {
       const task = await harness.store().createTask({ description: "Agent paused task" });

packages/core/src/__tests__/store-test-helpers.ts

@@ -45,6 +45,7 @@ export function createTaskStoreTestHarness() {
     globalDir: () => globalDir,
     store: () => store,
     beforeEach: async () => {
+      vi.useRealTimers();
       rootDir = makeTmpDir();
       globalDir = makeTmpDir();
       store = new TaskStore(rootDir, globalDir, { inMemoryDb: true });

packages/core/src/store.ts

@@ -3677,6 +3677,20 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
           ...(runContext ? { runContext } : {}),
         });
       }
+      if (assignmentChanged) {
+        this.syncAgentTaskLinkOnReassignment(id, previousAssignedAgentId, task.assignedAgentId);
+
+        if (task.checkedOutBy === previousAssignedAgentId) {
+          task.checkedOutBy = undefined;
+          task.checkedOutAt = undefined;
+        }
+
+        task.log.push({
+          timestamp: new Date().toISOString(),
+          action: `Agent task link synced: ${previousAssignedAgentId ?? "none"} → ${task.assignedAgentId ?? "none"}`,
+          ...(runContext ? { runContext } : {}),
+        });
+      }
       if (updates.pausedByAgentId === null) {
         task.pausedByAgentId = undefined;
       } else if (updates.pausedByAgentId !== undefined) {
@@ -4595,6 +4609,51 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
     `).run(updatedAt, updatedAt, taskId);
   }
 
+  /**
+   * Sync `agents.taskId` when {@link updateTask} reassigns a task.
+   *
+   * Uses direct SQL against the shared `agents` table instead of AgentStore to
+   * avoid a circular dependency while keeping the column and JSON data blob in
+   * lockstep. Clearing the previous agent is race-guarded with `WHERE id = ?
+   * AND taskId = ?` so we do not clobber an agent that already moved on to a
+   * different task.
+   */
+  private syncAgentTaskLinkOnReassignment(
+    taskId: string,
+    previousAgentId: string | undefined,
+    newAgentId: string | undefined,
+  ): void {
+    const updatedAt = new Date().toISOString();
+
+    if (previousAgentId) {
+      this.db.prepare(`
+        UPDATE agents
+        SET
+          taskId = NULL,
+          updatedAt = ?,
+          data = CASE
+            WHEN json_valid(data) THEN json_set(json_remove(data, '$.taskId'), '$.updatedAt', ?)
+            ELSE data
+          END
+        WHERE id = ? AND taskId = ?
+      `).run(updatedAt, updatedAt, previousAgentId, taskId);
+    }
+
+    if (newAgentId) {
+      this.db.prepare(`
+        UPDATE agents
+        SET
+          taskId = ?,
+          updatedAt = ?,
+          data = CASE
+            WHEN json_valid(data) THEN json_set(data, '$.taskId', ?, '$.updatedAt', ?)
+            ELSE data
+          END
+        WHERE id = ?
+      `).run(taskId, updatedAt, taskId, updatedAt, newAgentId);
+    }
+  }
+
   /**
    * Clean up the git branch associated with a task.
    *

packages/engine/src/__tests__/merger-overlap-guard.test.ts

@@ -1,5 +1,5 @@
-import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { execSync } from "node:child_process";
 import { join, resolve } from "node:path";
 import { tmpdir } from "node:os";
@@ -151,17 +151,42 @@ function assertIsolatedWorkspace(dir: string): void {
   expect(resolve(dir).startsWith(resolve(repoRoot))).toBe(false);
 }
 
+const createdDirs = new Set<string>();
+
+function cleanupTempDir(dir?: string): void {
+  if (!dir) return;
+  createdDirs.delete(dir);
+  for (let attempt = 1; attempt <= 3; attempt += 1) {
+    try {
+      if (!existsSync(dir)) return;
+      rmSync(dir, { recursive: true, force: true });
+      return;
+    } catch (error) {
+      if (attempt === 3) {
+        throw error;
+      }
+    }
+  }
+}
+
+afterAll(() => {
+  for (const dir of Array.from(createdDirs)) {
+    cleanupTempDir(dir);
+  }
+});
+
 describe("merger overlap guard", () => {
   let dir: string;
 
   beforeEach(() => {
     dir = mkdtempSync(join(tmpdir(), "fusion-test-overlap-guard-"));
+    createdDirs.add(dir);
     assertIsolatedWorkspace(dir);
     initRepo(dir);
   });
 
   afterEach(() => {
-    rmSync(dir, { recursive: true, force: true });
+    cleanupTempDir(dir);
   });
 
   it("detects overlap when branch and recent main commits touch the same file", async () => {
@@ -302,12 +327,13 @@ describe("aiMergeTask overlap-aware fallback integration", () => {
 
   beforeEach(() => {
     dir = mkdtempSync(join(tmpdir(), "fusion-test-overlap-merge-"));
+    createdDirs.add(dir);
     assertIsolatedWorkspace(dir);
     initRepo(dir);
   });
 
   afterEach(() => {
-    rmSync(dir, { recursive: true, force: true });
+    cleanupTempDir(dir);
   });
 
   it("defaults to restoring the branch version for overlapping files under smart-prefer-main", async () => {

packages/engine/src/__tests__/merger-staging-allowlist.test.ts

@@ -15,8 +15,8 @@
  *   6. Untracked file pre-existing in working tree (user WIP) — NOT staged
  */
 
-import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
-import { mkdtempSync, mkdirSync, writeFileSync, rmSync, readFileSync } from "node:fs";
+import { describe, it, expect, beforeEach, afterEach, afterAll, vi } from "vitest";
+import { mkdtempSync, mkdirSync, writeFileSync, rmSync, readFileSync, existsSync } from "node:fs";
 import { join, resolve } from "node:path";
 import { tmpdir } from "node:os";
 import { execSync } from "node:child_process";
@@ -85,6 +85,30 @@ const STUB_SETTINGS = {
   commitAuthorEnabled: false, // skip --author flag to avoid user config issues
 };
 
+const createdDirs = new Set<string>();
+
+function cleanupTempDir(dir?: string): void {
+  if (!dir) return;
+  createdDirs.delete(dir);
+  for (let attempt = 1; attempt <= 3; attempt += 1) {
+    try {
+      if (!existsSync(dir)) return;
+      rmSync(dir, { recursive: true, force: true });
+      return;
+    } catch (error) {
+      if (attempt === 3) {
+        throw error;
+      }
+    }
+  }
+}
+
+afterAll(() => {
+  for (const dir of Array.from(createdDirs)) {
+    cleanupTempDir(dir);
+  }
+});
+
 // ---------------------------------------------------------------------------
 // Tests
 // ---------------------------------------------------------------------------
@@ -94,12 +118,13 @@ describe("snapshotDirtyFiles", () => {
 
   beforeEach(() => {
     dir = mkdtempSync(join(tmpdir(), "fusion-test-merger-snapshot-"));
+    createdDirs.add(dir);
     assertIsolatedWorkspace(dir);
     initRepo(dir);
   });
 
   afterEach(() => {
-    rmSync(dir, { recursive: true, force: true });
+    cleanupTempDir(dir);
   });
 
   it("returns empty set when working tree is clean", async () => {
@@ -159,14 +184,18 @@ describe("commitOrAmendMergeWithFixes — staging allowlist", () => {
 
   beforeEach(() => {
     dir = mkdtempSync(join(tmpdir(), "fusion-test-merger-allowlist-"));
+    createdDirs.add(dir);
     assertIsolatedWorkspace(dir);
     initRepo(dir);
     warnSpy = vi.spyOn(mergerLog, "warn");
   });
 
   afterEach(() => {
-    warnSpy.mockRestore();
-    rmSync(dir, { recursive: true, force: true });
+    try {
+      warnSpy.mockRestore();
+    } finally {
+      cleanupTempDir(dir);
+    }
   });
 
   // ── Scenario 1: Unrelated dirty file is excluded ───────────────────────
@@ -543,14 +572,18 @@ describe("commitOrAmendMergeWithFixes — embedded-space paths round-trip", () =
 
   beforeEach(() => {
     dir = mkdtempSync(join(tmpdir(), "fusion-test-merger-allowlist-spaces-"));
+    createdDirs.add(dir);
     assertIsolatedWorkspace(dir);
     initRepo(dir);
     warnSpy = vi.spyOn(mergerLog, "warn");
   });
 
   afterEach(() => {
-    warnSpy.mockRestore();
-    rmSync(dir, { recursive: true, force: true });
+    try {
+      warnSpy.mockRestore();
+    } finally {
+      cleanupTempDir(dir);
+    }
   });
 
   it("stages and commits a tracked file edited by the fix agent whose path contains spaces", async () => {