fix(FN-4811): close concurrent-execute race that produced parallel runs

TaskExecutor.execute() had a classic JS async race window. Original:

  async execute(task) {
    if (this.executing.has(task.id)) return;                       // check
    const assignedAgentId = task.assignedAgentId;
    if (assignedAgentId && await this.shouldDeferForHeartbeat(...)) // AWAIT yields
      return;
    this.executing.add(task.id);                                   // add (too late)
    ...
  }

Two concurrent execute(task) calls (scheduler dispatch + task:moved event
handler + restart-recovery) both:
  1. Pass the synchronous has() check (Set is empty).
  2. Enter the awaited shouldDeferForHeartbeat call (yields the event loop).
  3. Resume and both call this.executing.add(task.id).
  4. Both proceed to create the same worktree path.

Production failure shape (FN-4814 + FN-4811, observed within minutes):

  01:30:56  [runA-caoe]  Worktree created at /...worktrees/bright-mesa
  01:30:56  [runB-w23q]  Worktree created at /...worktrees/bright-mesa
  01:30:58              worktree liveness assertion failed: not_usable_task_worktree
  01:31:48              [thirdRun] also fires liveness assertion fail
  01:37:48              In-review stall surfaced [no-worktree-no-merge-confirmed]

This is the root cause of the entire FN-4781/FN-4804/FN-4814/FN-4811
cascade. Every other guard added today (FN-4811 active-session gate,
self-healing reclaim defer, validation-failed recovery, silent reclaim
recovery, integrity-warning dedup) was patching SYMPTOMS of the
duplicate-run race. With this fix, the symptoms stop appearing.

Fix: claim the slot synchronously immediately after the has() check,
release it on the heartbeat-defer early-return path. No await happens
between check and claim, so the race window is closed.

Test added under
packages/engine/src/__tests__/reliability-interactions/concurrent-execute-race.test.ts
verified to fail on the prior (a1b1f9aa0) executor.ts and pass on the
fixed version:

  - Two concurrent execute() calls produce the SAME number of
    createFnAgent invocations as one execute() call (no amplification).
  - A second sequential execute() after the first completes IS allowed
    (slot was released).

The task must have assignedAgentId set to exercise the race \u2014 without
it, the short-circuit `assignedAgentId && ...` evaluates the left side
to false synchronously, and no await happens.

Full engine suite: 5048+ tests pass. The 7 transient test-file failures
in the broad parallel run are pre-existing flaky real-git tests
(branch-conflicts-zero-unique, branch-conflicts-recovery,
merger-overlap-guard subprocess-guard contention) \u2014 all of them pass
when run alone or as a smaller group, none touch the executor.execute()
path.

Fusion-Task-Id: FN-4811

.changeset/FN-4811-concurrent-execute-race.md

@@ -0,0 +1,19 @@
+---
+"@runfusion/fusion": patch
+---
+
+fix(FN-4811): close concurrent-execute race that produced parallel runs for the same task
+
+`TaskExecutor.execute()` had an async race: after the synchronous `this.executing.has(task.id)` check, the code awaited `shouldDeferForHeartbeat(...)` BEFORE adding to the `executing` Set. Two concurrent `execute()` calls (scheduler dispatch + `task:moved` listener + restart-recovery) could both pass the check, both yield on the await, then both add to the Set and both proceed to create the same worktree.
+
+Production signature (FN-4814, FN-4811):
+
+```
+01:30:56  [runA-caoe]  Worktree created at /Users/eclipxe/Projects/kb/.worktrees/bright-mesa
+01:30:56  [runB-w23q]  Worktree created at /Users/eclipxe/Projects/kb/.worktrees/bright-mesa
+01:30:58              worktree liveness assertion failed: not_usable_task_worktree
+```
+
+This is the canonical source of FN-4781/FN-4804/FN-4814/FN-4811 mid-task worktree disappearance and cross-task contamination — every other guard in the stack (FN-4811 active-session gate, self-healing reclaim defer, etc.) was patching the *symptoms* of the duplicate-run race.
+
+Fix: claim the executing slot synchronously immediately after the `has()` check, release it on the heartbeat-defer early return. Closes the race window entirely.

packages/engine/src/__tests__/reliability-interactions/concurrent-execute-race.test.ts

@@ -0,0 +1,140 @@
+/**
+ * FN-4811 follow-up (FN-4814 + FN-4811 production failures):
+ *
+ * `TaskExecutor.execute()` previously had a race window where two concurrent
+ * `execute(task)` calls (e.g., scheduler dispatch + restart-recovery + task:moved
+ * event handler) could both pass the `this.executing.has(task.id)` check, both
+ * await `shouldDeferForHeartbeat`, and both proceed to create the same worktree
+ * path — producing two parallel runs for the same task. Production log signature:
+ *
+ *   01:30:56  [runA-caoe]  Worktree created at /Users/eclipxe/Projects/kb/.worktrees/bright-mesa
+ *   01:30:56  [runB-w23q]  Worktree created at /Users/eclipxe/Projects/kb/.worktrees/bright-mesa
+ *   01:30:58  worktree liveness assertion failed: not_usable_task_worktree
+ *
+ * The fix: claim the executing slot SYNCHRONOUSLY immediately after the `has()`
+ * check, before any await. This regression test issues two concurrent execute()
+ * calls and asserts that only ONE actually runs (only one createFnAgent invocation).
+ */
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import "../executor-test-helpers.js";
+import { TaskExecutor } from "../../executor.js";
+import { mockedCreateFnAgent, createMockStore, resetExecutorMocks } from "../executor-test-helpers.js";
+
+function makeTask(overrides: Record<string, unknown> = {}) {
+  return {
+    id: "FN-4814",
+    title: "Concurrent execute race",
+    description: "test",
+    column: "in-progress",
+    paused: false,
+    worktree: "/tmp/test/.worktrees/bright-mesa",
+    branch: "fusion/fn-4814",
+    // assignedAgentId is REQUIRED to actually exercise the race — it's the conditional
+    // that gates the `await shouldDeferForHeartbeat(...)` which is the offending yield
+    // point. Without it, the short-circuit `assignedAgentId && ...` evaluates to false
+    // synchronously and no await happens, so the race window doesn't exist.
+    assignedAgentId: "agent-test-executor",
+    dependencies: [],
+    steps: [],
+    currentStep: 0,
+    log: [],
+    prompt: "# test",
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    ...overrides,
+  } as any;
+}
+
+describe("FN-4811 follow-up (FN-4814): concurrent execute() must not produce parallel runs", () => {
+  beforeEach(() => {
+    resetExecutorMocks();
+  });
+
+  it("two concurrent execute() calls for the same task produce no more sessions than one execute() call", async () => {
+    // Establish baseline: how many createFnAgent invocations happen for ONE execute().
+    // The mocked prompt never calls fn_task_done, so the retry loop fires; we don't care
+    // about the exact count, only that concurrent calls don't AMPLIFY it.
+    const baselineStore = createMockStore();
+    mockedCreateFnAgent.mockImplementation(async () => ({
+      session: {
+        prompt: vi.fn(async () => {
+          await new Promise((r) => setTimeout(r, 5));
+        }),
+        dispose: vi.fn(),
+        subscribe: vi.fn(),
+        on: vi.fn(),
+        sessionManager: { getLeafId: vi.fn().mockReturnValue("leaf-1") },
+        navigateTree: vi.fn(),
+        state: {},
+      },
+    }) as any);
+    const baselineExecutor = new TaskExecutor(baselineStore as any, "/tmp/test");
+    await baselineExecutor.execute(makeTask());
+    const baselineCount = mockedCreateFnAgent.mock.calls.length;
+    expect(baselineCount).toBeGreaterThan(0);
+
+    // Now exercise the race: two concurrent execute() calls in the same tick.
+    resetExecutorMocks();
+    mockedCreateFnAgent.mockImplementation(async () => {
+      // Wider latency than baseline to make the race window deterministic.
+      await new Promise((r) => setTimeout(r, 20));
+      return {
+        session: {
+          prompt: vi.fn(async () => {
+            await new Promise((r) => setTimeout(r, 20));
+          }),
+          dispose: vi.fn(),
+          subscribe: vi.fn(),
+          on: vi.fn(),
+          sessionManager: { getLeafId: vi.fn().mockReturnValue("leaf-1") },
+          navigateTree: vi.fn(),
+          state: {},
+        },
+      } as any;
+    });
+
+    const store = createMockStore();
+    const executor = new TaskExecutor(store as any, "/tmp/test");
+    const task = makeTask();
+
+    const [resultA, resultB] = await Promise.allSettled([
+      executor.execute(task),
+      executor.execute(task),
+    ]);
+
+    expect(resultA.status).toBe("fulfilled");
+    expect(resultB.status).toBe("fulfilled");
+
+    // FN-4814: concurrent calls must NOT amplify createFnAgent invocations. Before the
+    // fix, both calls passed `executing.has()`, awaited `shouldDeferForHeartbeat`, and
+    // BOTH proceeded — doubling the createFnAgent count. After the fix, the second
+    // call returns immediately because the first synchronously claimed the slot.
+    const concurrentCount = mockedCreateFnAgent.mock.calls.length;
+    expect(concurrentCount).toBe(baselineCount);
+  });
+
+  it("a second sequential execute() after the first completes is allowed (slot is released)", async () => {
+    const store = createMockStore();
+
+    mockedCreateFnAgent.mockImplementation(async () => ({
+      session: {
+        prompt: vi.fn(async () => undefined),
+        dispose: vi.fn(),
+        subscribe: vi.fn(),
+        on: vi.fn(),
+        sessionManager: { getLeafId: vi.fn().mockReturnValue("leaf-1") },
+        navigateTree: vi.fn(),
+        state: {},
+      },
+    }) as any);
+
+    const executor = new TaskExecutor(store as any, "/tmp/test");
+    await executor.execute(makeTask());
+    const firstCount = mockedCreateFnAgent.mock.calls.length;
+    await executor.execute(makeTask());
+    const secondCount = mockedCreateFnAgent.mock.calls.length;
+
+    // The second call did SOMETHING (slot was released) — createFnAgent count grew.
+    expect(secondCount).toBeGreaterThan(firstCount);
+  });
+});

packages/engine/src/executor.ts

@@ -2570,14 +2570,24 @@ export class TaskExecutor {
     executorLog.log(`execute() called for ${task.id} (already executing=${this.executing.has(task.id)})`);
     if (this.executing.has(task.id)) return;
 
+    // FN-4811 follow-up (FN-4814/FN-4811 production failure): claim the executing slot
+    // SYNCHRONOUSLY before any await. Without this, two concurrent execute() calls
+    // (e.g., scheduler dispatch + restart-recovery + task:moved event) both pass the
+    // `has()` check, both await `shouldDeferForHeartbeat`, both proceed past it, and
+    // both end up creating the same worktree path — producing two parallel runs for
+    // the same task with duplicate "Worktree created at /..." log entries within the
+    // same second. This is the canonical source of FN-4781/FN-4804/FN-4814/FN-4811
+    // mid-task worktree disappearance and cross-task contamination.
+    this.executing.add(task.id);
+
     const assignedAgentId = task.assignedAgentId;
     if (assignedAgentId && await this.shouldDeferForHeartbeat(assignedAgentId)) {
       executorLog.log(`${task.id}: skipping execute — agent ${assignedAgentId} has active heartbeat run (allowParallelExecution=false)`);
+      // Release the slot we just claimed — we never actually ran.
+      this.executing.delete(task.id);
       return;
     }
 
-    this.executing.add(task.id);
-
     executorLog.log(`Starting ${task.id}: ${task.title || task.description.slice(0, 60)}`);
 
     // Fetch settings early — needed for worktree naming and later configuration