fix(FN-4811): process-wide executingTaskLock blocks parallel execute() across instances

Investigating FN-4814 + FN-4811 re-failures after commit 8bef30655 (which
added per-instance synchronous this.executing.add) revealed the per-instance
guard was insufficient. FN-4809 log at 02:48:17-18 UTC:

  02:48:17  [-]                Resuming execution after unpause
  02:48:17  [-]                Step 4 (Testing & Verification) -> pending
  02:48:17  [-]                Step 4 (Testing & Verification) -> pending
  02:48:17  [6097725-y2nb]     Executor detected stale merge state ...
  02:48:18  [6097816-9gde]     Executor detected stale merge state ...

Both runs y2nb and 9gde reached executor.ts:2661 (which is INSIDE execute(),
past the synchronous this.executing.add claim). The only viable explanation
is that there is more than one TaskExecutor instance in the same Node
process (engine restart race, multi-project hybrid runtime, or similar code
path). Each instance has its own executing Set, so the per-instance guard
doesn't help.

Fix: module-level singleton executingTaskLock in active-session-registry.ts,
shared across all TaskExecutor instances. execute() synchronously tryClaim()s
the lock; if false, bails. Every existing this.executing.delete() site also
calls executingTaskLock.release(). Per-instance this.executing kept because
many other call sites use it (this.executing.has at handler gates,
stuck-detector, resumeTaskForAgent, etc.).

Test setup (resetExecutorMocks in executor-test-helpers.ts) clears the lock
between tests so process-wide state doesn't leak (executor-pause and
executor-prompt tests would otherwise show 'expected 2 createFnAgent calls
but got 0' / 'expected not called but called 3 times' flakes).

Tests:
  - executing-task-lock.test.ts: 2 cases. Key case creates TWO TaskExecutor
    instances and races them on the same task ID, asserts only ONE actually
    runs. Verified FAILS on prior code (8bef30655) and PASSES on fix.

Verification:
  - Targeted suite (4 files, 170 tests): pass.
  - pnpm --filter @fusion/engine build: clean.
  - pnpm lint: clean.

Fusion-Task-Id: FN-4811

.changeset/FN-4811-process-wide-execute-lock.md

@@ -0,0 +1,11 @@
+---
+"@runfusion/fusion": patch
+---
+
+fix(FN-4811): use process-wide executingTaskLock to block parallel execute() across instances
+
+After commit 82f80e72f added a per-instance `this.executing.add()` synchronous claim, production STILL produced two `execute()` invocations for the same task ID that both reached "Executor detected stale merge state" and both generated runIds within 1 second of each other (FN-4809: y2nb + 9gde at 02:48:17–18 UTC; FN-4814 / FN-4811 cascade). The only viable explanation is that there is more than one `TaskExecutor` instance in the process (engine restart race, multi-project hybrid runtime, etc.).
+
+Adds a module-level singleton `executingTaskLock` in `active-session-registry.ts` shared across all `TaskExecutor` instances. `TaskExecutor.execute()` synchronously claims the lock immediately after the `executorLog.log` entry; if `tryClaim()` returns false (someone else owns the lock), the call bails. Every existing `this.executing.delete()` site also releases the lock. Per-instance `this.executing` is kept for back-compat with the many `this.executing.has()` checks throughout `executor.ts`.
+
+Test setup in `executor-test-helpers.ts` clears the process-wide lock in `resetExecutorMocks()` so it doesn't leak across tests.

packages/engine/src/__tests__/executor-test-helpers.ts

@@ -248,6 +248,7 @@ import { existsSync, realpathSync } from "node:fs";
 import { hydrateWorktreeDb } from "../worktree-db-hydrate.js";
 import { isUsableTaskWorktree } from "../worktree-pool.js";
 import { classifyStaleLock, tryRemoveStaleLock } from "../worktree-stale-lock.js";
+import { executingTaskLock } from "../active-session-registry.js";
 
 export const mockedCreateFnAgent = vi.mocked(createFnAgent);
 export const mockedSessionManager = vi.mocked(SessionManager);
@@ -343,4 +344,9 @@ export function resetExecutorMocks() {
   mockExecuteAll.mockResolvedValue([]);
   mockTerminateAllSessions.mockResolvedValue(undefined);
   mockCleanup.mockResolvedValue(undefined);
+  // FN-4811 follow-up: the executingTaskLock is process-wide module state, so it must
+  // be cleared between tests or earlier tests' claims will block later tests' execute()
+  // calls ("expected at least 2 createFnAgent calls but got 0" / "expected not called
+  // but called 3 times" symptoms in executor-pause / executor-prompt tests).
+  executingTaskLock._clearForTest();
 }

packages/engine/src/__tests__/reliability-interactions/executing-task-lock.test.ts

@@ -0,0 +1,112 @@
+/**
+ * FN-4811 follow-up (FN-4809 production reproduction):
+ *
+ * After commit 82f80e72f added a per-instance `this.executing.add()` synchronous
+ * claim, production STILL produced two execute() invocations for the same task
+ * ID that both reached "Executor detected stale merge state" (executor.ts:2661)
+ * and both generated runIds within 1 second of each other (y2nb + 9gde for
+ * FN-4809 at 02:48:17–18 UTC). The only viable explanation is that there is
+ * more than one `TaskExecutor` instance in the process (engine restart race,
+ * multi-project hybrid runtime, or test-helper-style code creating a second
+ * instance).
+ *
+ * The fix is a process-wide singleton `executingTaskLock` in
+ * `active-session-registry.ts`. This test covers the contract directly.
+ */
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import "../executor-test-helpers.js";
+import { TaskExecutor } from "../../executor.js";
+import { executingTaskLock } from "../../active-session-registry.js";
+import { mockedCreateFnAgent, createMockStore, resetExecutorMocks } from "../executor-test-helpers.js";
+
+function makeTask(overrides: Record<string, unknown> = {}) {
+  return {
+    id: "FN-4809",
+    title: "Process-wide execute lock",
+    description: "test",
+    column: "in-progress",
+    paused: false,
+    worktree: "/tmp/test/.worktrees/rapid-fern",
+    branch: "fusion/fn-4809",
+    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-4809): process-wide executingTaskLock", () => {
+  beforeEach(() => {
+    resetExecutorMocks();
+    executingTaskLock._clearForTest();
+  });
+
+  it("two TaskExecutor instances racing execute() for the same task produce only one run", async () => {
+    // Two stores, two executors — simulates engine restart race, multi-project
+    // hybrid runtime, or any code path that creates a second TaskExecutor.
+    const storeA = createMockStore();
+    const storeB = createMockStore();
+
+    mockedCreateFnAgent.mockImplementation(async () => {
+      await new Promise((r) => setTimeout(r, 20));
+      return {
+        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 executorA = new TaskExecutor(storeA as any, "/tmp/test");
+    const executorB = new TaskExecutor(storeB as any, "/tmp/test");
+    const task = makeTask();
+
+    const [resultA, resultB] = await Promise.allSettled([
+      executorA.execute(task),
+      executorB.execute(task),
+    ]);
+
+    expect(resultA.status).toBe("fulfilled");
+    expect(resultB.status).toBe("fulfilled");
+
+    // Exactly one store should have received work-related log entries — the
+    // losing instance bailed at the process-wide claim before any work began.
+    const aLogCount = (storeA.logEntry as any).mock.calls.length;
+    const bLogCount = (storeB.logEntry as any).mock.calls.length;
+    expect((aLogCount > 0) !== (bLogCount > 0)).toBe(true);
+  });
+
+  it("releases the lock after execute() finishes so subsequent calls proceed", 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());
+    expect(executingTaskLock.has("FN-4809")).toBe(false);
+
+    // Second sequential call must be allowed.
+    await executor.execute(makeTask());
+    expect(executingTaskLock.has("FN-4809")).toBe(false);
+  });
+});

packages/engine/src/active-session-registry.ts

@@ -51,3 +51,41 @@ class ActiveSessionRegistry {
 }
 
 export const activeSessionRegistry = new ActiveSessionRegistry();
+
+/**
+ * FN-4811 follow-up: process-wide "executing" lock for `TaskExecutor.execute()`.
+ *
+ * Per-instance `executing: Set<string>` is insufficient when there can be more than
+ * one TaskExecutor instance in the same Node process (e.g., multi-project setups,
+ * engine restarts that race with old instance teardown, hybrid-executor path).
+ * Production failure shape: two execute() invocations for the same task ID both
+ * generated runIds (y2nb + 9gde for FN-4809), both reached "Executor detected stale
+ * merge state" (executor.ts:2661), both attempted worktree creation — producing
+ * duplicate "Worktree created at /..." log entries within the same second
+ * (FN-4809, FN-4814, FN-4781, FN-4804, FN-4811).
+ *
+ * This module-level Set is shared across all TaskExecutor instances in the process,
+ * providing a process-wide claim. Values are taskId strings; presence means
+ * "someone is actively executing this task". Callers MUST claim synchronously
+ * via `tryClaim()` and MUST release on every exit path.
+ */
+const executingTasks = new Set<string>();
+
+export const executingTaskLock = {
+  has(taskId: string): boolean {
+    return executingTasks.has(taskId);
+  },
+  /** Synchronously claim the lock. Returns true if claimed, false if already held. */
+  tryClaim(taskId: string): boolean {
+    if (executingTasks.has(taskId)) return false;
+    executingTasks.add(taskId);
+    return true;
+  },
+  release(taskId: string): void {
+    executingTasks.delete(taskId);
+  },
+  /** Test-only: clear all entries. */
+  _clearForTest(): void {
+    executingTasks.clear();
+  },
+};

packages/engine/src/executor.ts

@@ -44,7 +44,7 @@ import type { SandboxBackend } from "./sandbox/types.js";
 import { ModelRegistry, SessionManager, type ToolDefinition, type AgentSession } from "@mariozechner/pi-coding-agent";
 import { PRIORITY_EXECUTE, type AgentSemaphore } from "./concurrency.js";
 import { RemovalReason, getRegisteredWorktreePaths, isGitRepository, isInsideWorktreesDir, isRegisteredGitWorktree, isUsableTaskWorktree, removeWorktree, type WorktreePool } from "./worktree-pool.js";
-import { activeSessionRegistry } from "./active-session-registry.js";
+import { activeSessionRegistry, executingTaskLock } from "./active-session-registry.js";
 import {
   StaleWorktreeIndexLockError,
   classifyStaleLock,
@@ -2573,17 +2573,21 @@ export class TaskExecutor {
    * as-is. Branches remain task-scoped (`fusion/{task-id}`).
    */
   async execute(task: Task): Promise<void> {
-    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-4809/FN-4814/FN-4811 production failure): claim a
+    // PROCESS-WIDE lock synchronously before any other work. Per-instance
+    // `this.executing` was insufficient in production because two execute()
+    // invocations for the same task ID still both reached "Executor detected
+    // stale merge state" (executor.ts:2661) and both generated runIds — the only
+    // viable explanation is multiple TaskExecutor instances in the same process
+    // (engine restart race, multi-project hybrid runtime, etc.). The only
+    // fully-reliable guard is a singleton lock shared across all instances.
+    const claimed = executingTaskLock.tryClaim(task.id);
+    executorLog.log(`execute() called for ${task.id} (claimed=${claimed}, perInstanceExecuting=${this.executing.has(task.id)})`);
+    if (!claimed) 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.
+    // Maintain the per-instance Set too, for back-compat with all the existing
+    // `this.executing.has()` checks throughout the file (handler gates,
+    // stuck-detector, resumeTaskForAgent, etc.).
     this.executing.add(task.id);
 
     const assignedAgentId = task.assignedAgentId;
@@ -2591,6 +2595,7 @@ export class TaskExecutor {
       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);
+      executingTaskLock.release(task.id);
       return;
     }
 
@@ -3306,6 +3311,7 @@ export class TaskExecutor {
           }
         } finally {
           this.executing.delete(task.id);
+          executingTaskLock.release(task.id);
           this.loopRecoveryState.delete(task.id);
           // Wrap cleanup in try/catch so activeStepExecutors.delete() always runs.
           // If cleanup() throws, the executor continues to clean up the in-memory map
@@ -4652,6 +4658,7 @@ export class TaskExecutor {
       }
 
       this.executing.delete(task.id);
+      executingTaskLock.release(task.id);
       // Clear run context at end of execute() lifecycle
       this.currentRunContext = undefined;
 
@@ -8990,6 +8997,7 @@ Backward compat fallback: if JSON is unavailable, you may still begin output wit
             `${taskId} force-requeue skipped — task is now in '${latestColumn}' (recovered concurrently)`,
           );
           this.executing.delete(taskId);
+          executingTaskLock.release(taskId);
           this.stuckAborted.delete(taskId);
           return;
         }
@@ -9010,6 +9018,7 @@ Backward compat fallback: if JSON is unavailable, you may still begin output wit
           // The old Promise is still running but the executing guard is cleared so
           // a fresh execute() call won't be blocked.
           this.executing.delete(taskId);
+          executingTaskLock.release(taskId);
           this.stuckAborted.delete(taskId);
           executorLog.log(`${taskId} force-requeued to todo`);
         } catch (err: unknown) {