feat(engine): guard one engine per project per machine

Adds a per-machine singleton lock that engages in
ProjectEngineManager.createAndStart() before any engine subsystems
spin up. Two fn dashboard processes can no longer run engines for the
same project on the same host — previously they would share .fusion/
state and corrupt worktrees / task rows for in-process projects.

The guard combines two independent checks:
  - A proper-lockfile file at <project>/.fusion/engine.lock with
    stale-lock recovery (auto-released on process death).
  - A loopback listener on a hashed per-project address — UDS on
    POSIX, named pipe on Windows. Stale UDS files are probed and
    unlinked before a retry bind.

Failures raise EngineAlreadyRunningError. Both guards are released
from stopAll() and pauseProject(); a release on engine.start()
failure lets retries re-acquire cleanly.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

packages/engine/src/__tests__/engine-singleton-lock.test.ts

@@ -0,0 +1,118 @@
+import { mkdtempSync, rmSync, existsSync } from "node:fs";
+import { tmpdir, platform } from "node:os";
+import { join } from "node:path";
+import net from "node:net";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import {
+  acquireEngineSingleton,
+  computeEngineLockFilePath,
+  computeEngineSocketPath,
+  EngineAlreadyRunningError,
+  type EngineSingletonLock,
+} from "../engine-singleton-lock.js";
+
+function uniqueProjectId(label: string): string {
+  return `proj_${label}_${process.pid}_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
+}
+
+describe("engine-singleton-lock", () => {
+  let workDir: string;
+  let acquired: EngineSingletonLock[] = [];
+
+  beforeEach(() => {
+    workDir = mkdtempSync(join(tmpdir(), "fusion-engine-lock-test-"));
+    acquired = [];
+  });
+
+  afterEach(async () => {
+    for (const lock of acquired) {
+      await lock.release().catch(() => {});
+    }
+    rmSync(workDir, { recursive: true, force: true });
+  });
+
+  it("computes a deterministic socket path per projectId", () => {
+    const id = "proj_deterministic";
+    expect(computeEngineSocketPath(id)).toBe(computeEngineSocketPath(id));
+    expect(computeEngineSocketPath(id)).not.toBe(
+      computeEngineSocketPath("proj_other"),
+    );
+  });
+
+  it("places the lockfile under <workingDir>/.fusion/engine.lock", () => {
+    expect(computeEngineLockFilePath(workDir)).toBe(
+      join(workDir, ".fusion", "engine.lock"),
+    );
+  });
+
+  it("acquires successfully on first call and creates .fusion/engine.lock", async () => {
+    const id = uniqueProjectId("first");
+    const lock = await acquireEngineSingleton(id, workDir);
+    acquired.push(lock);
+    expect(existsSync(lock.lockFilePath)).toBe(true);
+  });
+
+  it("rejects a second acquisition for the same project + workingDir", async () => {
+    const id = uniqueProjectId("double");
+    const first = await acquireEngineSingleton(id, workDir);
+    acquired.push(first);
+    await expect(acquireEngineSingleton(id, workDir)).rejects.toBeInstanceOf(
+      EngineAlreadyRunningError,
+    );
+  });
+
+  it("allows re-acquisition after release", async () => {
+    const id = uniqueProjectId("cycle");
+    const first = await acquireEngineSingleton(id, workDir);
+    await first.release();
+    // Double-release must be a no-op.
+    await first.release();
+
+    const second = await acquireEngineSingleton(id, workDir);
+    acquired.push(second);
+    expect(existsSync(second.lockFilePath)).toBe(true);
+  });
+
+  it("recovers from a stale socket file (POSIX only)", async () => {
+    if (platform() === "win32") return;
+
+    const id = uniqueProjectId("stale");
+    const socketPath = computeEngineSocketPath(id);
+
+    // Simulate a stale socket file from a crashed engine.
+    const stale = net.createServer();
+    stale.unref();
+    await new Promise<void>((resolve, reject) => {
+      stale.once("error", reject);
+      stale.once("listening", () => resolve());
+      stale.listen(socketPath);
+    });
+    await new Promise<void>((resolve) => stale.close(() => resolve()));
+
+    const lock = await acquireEngineSingleton(id, workDir);
+    acquired.push(lock);
+    expect(existsSync(lock.lockFilePath)).toBe(true);
+  });
+
+  it("releases the lockfile so a fresh process could re-acquire", async () => {
+    const id = uniqueProjectId("release-lockfile");
+    const lock = await acquireEngineSingleton(id, workDir);
+    await lock.release();
+    // proper-lockfile uses `<path>.lock` as the actual mutex dir.
+    expect(existsSync(`${lock.lockFilePath}.lock`)).toBe(false);
+  });
+
+  it("different projects don't block each other", async () => {
+    const a = await acquireEngineSingleton(uniqueProjectId("a"), workDir);
+    acquired.push(a);
+    const otherWork = mkdtempSync(join(tmpdir(), "fusion-engine-lock-test-b-"));
+    try {
+      const b = await acquireEngineSingleton(uniqueProjectId("b"), otherWork);
+      acquired.push(b);
+      expect(existsSync(a.lockFilePath)).toBe(true);
+      expect(existsSync(b.lockFilePath)).toBe(true);
+    } finally {
+      rmSync(otherWork, { recursive: true, force: true });
+    }
+  });
+});

packages/engine/src/__tests__/project-engine-manager.test.ts

@@ -1,5 +1,16 @@
 import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
 
+// Stub out the per-machine singleton lock so tests with fake working dirs
+// (e.g. /mapped/...) don't try to mkdir or bind real sockets.
+vi.mock("../engine-singleton-lock.js", () => ({
+  acquireEngineSingleton: vi.fn().mockResolvedValue({
+    release: vi.fn().mockResolvedValue(undefined),
+    socketPath: "/tmp/test.sock",
+    lockFilePath: "/tmp/test.lock",
+  }),
+  EngineAlreadyRunningError: class EngineAlreadyRunningError extends Error {},
+}));
+
 // Mock ProjectEngine before importing the manager
 vi.mock("../project-engine.js", () => {
   return {

packages/engine/src/engine-singleton-lock.ts

@@ -0,0 +1,224 @@
+/**
+ * Engine singleton lock — ensures only one engine per project per machine.
+ *
+ * Acquires two independent guards before the engine subsystems start:
+ *
+ *   1. A lockfile under `<workingDir>/.fusion/engine.lock` via `proper-lockfile`.
+ *      Uses OS-level link()/atomic-rename semantics; auto-released on process
+ *      death; stale locks are recovered after `STALE_MS` of no `mtime` updates.
+ *
+ *   2. A loopback listener on a per-project address. On POSIX this is a Unix
+ *      domain socket under `os.tmpdir()`; on Windows it is a named pipe under
+ *      `\\.\pipe\`. Node's `net.Server.listen(path)` abstracts both. If the
+ *      address is already bound, another engine is live.
+ *
+ * Together the two guards cover the failure modes the other one misses:
+ *   - Lockfile alone: file locks can survive an `rm -rf .fusion`.
+ *   - Socket alone: stale UDS files survive crashes (we probe + unlink).
+ *
+ * If acquisition fails, both guards are unwound before throwing.
+ */
+
+import { mkdirSync } from "node:fs";
+import { access, unlink, writeFile } from "node:fs/promises";
+import net from "node:net";
+import { platform, tmpdir } from "node:os";
+import { join } from "node:path";
+import { createHash } from "node:crypto";
+import lockfile from "proper-lockfile";
+
+const STALE_MS = 30_000;
+const UPDATE_MS = 10_000;
+const PROBE_TIMEOUT_MS = 500;
+
+export interface EngineSingletonLock {
+  /** Idempotent release of both the lockfile and the loopback listener. */
+  release(): Promise<void>;
+  /** Address of the loopback listener (UDS path or named pipe). */
+  readonly socketPath: string;
+  /** Path of the lockfile target. */
+  readonly lockFilePath: string;
+}
+
+export class EngineAlreadyRunningError extends Error {
+  constructor(
+    public readonly projectId: string,
+    public readonly reason: "lockfile" | "socket",
+    cause?: unknown,
+  ) {
+    super(
+      `Another engine is already running for project ${projectId} on this machine (blocked by ${reason})`,
+    );
+    this.name = "EngineAlreadyRunningError";
+    if (cause !== undefined) {
+      (this as { cause?: unknown }).cause = cause;
+    }
+  }
+}
+
+function projectHash(projectId: string): string {
+  return createHash("sha1").update(projectId).digest("hex").slice(0, 16);
+}
+
+/**
+ * Cross-platform local address for the loopback listener.
+ *   - POSIX: UDS path under `os.tmpdir()`. macOS caps sun_path at 104 bytes,
+ *     so we hash the project id to keep it short.
+ *   - Windows: named pipe under `\\.\pipe\`. No length issue but we hash
+ *     for symmetry and to avoid leaking project ids into the pipe namespace.
+ */
+export function computeEngineSocketPath(projectId: string): string {
+  const hash = projectHash(projectId);
+  if (platform() === "win32") {
+    return `\\\\.\\pipe\\fusion-engine-${hash}`;
+  }
+  return join(tmpdir(), `fusion-engine-${hash}.sock`);
+}
+
+export function computeEngineLockFilePath(workingDir: string): string {
+  return join(workingDir, ".fusion", "engine.lock");
+}
+
+async function ensureLockTargetExists(lockPath: string): Promise<void> {
+  try {
+    await access(lockPath);
+  } catch {
+    await writeFile(lockPath, "");
+  }
+}
+
+async function acquireLockfile(
+  workingDir: string,
+  onCompromised: (err: Error) => void,
+): Promise<{ release: () => Promise<void>; path: string }> {
+  const dir = join(workingDir, ".fusion");
+  mkdirSync(dir, { recursive: true });
+  const path = computeEngineLockFilePath(workingDir);
+  await ensureLockTargetExists(path);
+  const release = await lockfile.lock(path, {
+    stale: STALE_MS,
+    update: UPDATE_MS,
+    retries: 0,
+    realpath: false,
+    onCompromised,
+  });
+  return { release, path };
+}
+
+function listen(server: net.Server, path: string): Promise<void> {
+  return new Promise((resolve, reject) => {
+    const onError = (err: Error) => {
+      server.removeListener("listening", onListening);
+      reject(err);
+    };
+    const onListening = () => {
+      server.removeListener("error", onError);
+      resolve();
+    };
+    server.once("error", onError);
+    server.once("listening", onListening);
+    server.listen(path);
+  });
+}
+
+/** Probe the loopback address — true if another process accepts connections. */
+async function isAddressLive(path: string): Promise<boolean> {
+  return new Promise((resolve) => {
+    const socket = net.connect(path);
+    const done = (live: boolean) => {
+      socket.removeAllListeners();
+      socket.destroy();
+      resolve(live);
+    };
+    socket.once("connect", () => done(true));
+    socket.once("error", () => done(false));
+    setTimeout(() => done(false), PROBE_TIMEOUT_MS).unref();
+  });
+}
+
+async function bindLoopback(socketPath: string): Promise<net.Server> {
+  const server = net.createServer((socket) => {
+    socket.end();
+  });
+  // Don't let the listener itself keep the process alive — the engine has
+  // its own refs (scheduler timers, db handles) that determine lifetime.
+  server.unref();
+
+  try {
+    await listen(server, socketPath);
+    return server;
+  } catch (err) {
+    const code = (err as NodeJS.ErrnoException).code;
+    if (code === "EADDRINUSE" && platform() !== "win32") {
+      // POSIX: stale UDS file from a crashed engine. Probe first, unlink, retry.
+      const live = await isAddressLive(socketPath);
+      if (!live) {
+        await unlink(socketPath).catch(() => {});
+        const retryServer = net.createServer((s) => s.end());
+        retryServer.unref();
+        await listen(retryServer, socketPath);
+        return retryServer;
+      }
+    }
+    throw err;
+  }
+}
+
+function closeServer(server: net.Server): Promise<void> {
+  return new Promise((resolve) => {
+    server.close(() => resolve());
+  });
+}
+
+/**
+ * Acquire both guards. On any failure throws {@link EngineAlreadyRunningError}
+ * (or the underlying error) and releases whatever was partially acquired.
+ *
+ * @param projectId  Stable per-project id (used to derive the socket address).
+ * @param workingDir Project root — must exist; `.fusion/` is created if missing.
+ * @param onCompromised Called if the lockfile is lost mid-flight. Defaults to a no-op.
+ */
+export async function acquireEngineSingleton(
+  projectId: string,
+  workingDir: string,
+  onCompromised: (err: Error) => void = () => {},
+): Promise<EngineSingletonLock> {
+  let lock: { release: () => Promise<void>; path: string } | undefined;
+  let server: net.Server | undefined;
+  const socketPath = computeEngineSocketPath(projectId);
+  try {
+    try {
+      lock = await acquireLockfile(workingDir, onCompromised);
+    } catch (err) {
+      throw new EngineAlreadyRunningError(projectId, "lockfile", err);
+    }
+    try {
+      server = await bindLoopback(socketPath);
+    } catch (err) {
+      throw new EngineAlreadyRunningError(projectId, "socket", err);
+    }
+  } catch (err) {
+    if (server) {
+      await closeServer(server).catch(() => {});
+    }
+    if (lock) {
+      await lock.release().catch(() => {});
+    }
+    throw err;
+  }
+
+  let released = false;
+  const lockPath = lock.path;
+  const release = lock.release;
+  const boundServer = server;
+  return {
+    socketPath,
+    lockFilePath: lockPath,
+    async release() {
+      if (released) return;
+      released = true;
+      await closeServer(boundServer).catch(() => {});
+      await release().catch(() => {});
+    },
+  };
+}

packages/engine/src/index.ts

@@ -263,6 +263,13 @@ export {
 export { ProjectManager } from "./project-manager.js";
 export { ProjectEngine, type ProjectEngineOptions } from "./project-engine.js";
 export { ProjectEngineManager, type EngineManagerOptions } from "./project-engine-manager.js";
+export {
+  acquireEngineSingleton,
+  computeEngineLockFilePath,
+  computeEngineSocketPath,
+  EngineAlreadyRunningError,
+  type EngineSingletonLock,
+} from "./engine-singleton-lock.js";
 export { NodeHealthMonitor } from "./node-health-monitor.js";
 export {
   HybridExecutor,

packages/engine/src/project-engine-manager.ts

@@ -22,6 +22,11 @@ import { ProjectEngine } from "./project-engine.js";
 import type { ProjectEngineOptions } from "./project-engine.js";
 import type { ProjectRuntimeConfig } from "./project-runtime.js";
 import { AgentSemaphore } from "./concurrency.js";
+import {
+  acquireEngineSingleton,
+  EngineAlreadyRunningError,
+  type EngineSingletonLock,
+} from "./engine-singleton-lock.js";
 import { runtimeLog } from "./logger.js";
 
 /**
@@ -41,6 +46,7 @@ export const DEFAULT_RECONCILIATION_INTERVAL_MS = 30_000;
 export class ProjectEngineManager {
   private engines = new Map<string, ProjectEngine>();
   private starting = new Map<string, Promise<ProjectEngine>>();
+  private singletonLocks = new Map<string, EngineSingletonLock>();
   private stopped = false;
 
   /**
@@ -135,6 +141,8 @@ export class ProjectEngineManager {
 
     // Remove from starting set to prevent a stalled start from completing
     this.starting.delete(projectId);
+
+    await this.releaseSingleton(projectId);
   }
 
   /**
@@ -250,6 +258,30 @@ export class ProjectEngineManager {
     await Promise.all(stops);
     this.engines.clear();
     this.starting.clear();
+
+    // Release all singleton locks so another fusion process can take over.
+    const releases = Array.from(this.singletonLocks.values()).map((lock) =>
+      lock.release().catch((err) => {
+        const message = err instanceof Error ? err.message : String(err);
+        runtimeLog.warn(`Singleton lock release error: ${message}`);
+      }),
+    );
+    await Promise.all(releases);
+    this.singletonLocks.clear();
+  }
+
+  private async releaseSingleton(projectId: string): Promise<void> {
+    const lock = this.singletonLocks.get(projectId);
+    if (!lock) return;
+    this.singletonLocks.delete(projectId);
+    try {
+      await lock.release();
+    } catch (err) {
+      const message = err instanceof Error ? err.message : String(err);
+      runtimeLog.warn(
+        `Singleton lock release error for ${projectId}: ${message}`,
+      );
+    }
   }
 
   /**
@@ -378,13 +410,41 @@ export class ProjectEngineManager {
     const runtimeConfig = await this.buildRuntimeConfig(project);
     const engineOptions = this.buildEngineOptions(project, overrides);
 
+    // Acquire the per-machine singleton guard before spinning up any engine
+    // subsystems. This prevents two fusion processes from running engines for
+    // the same project on one machine.
+    const singleton = await acquireEngineSingleton(
+      projectId,
+      runtimeConfig.workingDirectory,
+      (err) => {
+        runtimeLog.warn(
+          `Engine singleton lock for ${projectId} was compromised: ${err.message}`,
+        );
+      },
+    ).catch((err) => {
+      if (err instanceof EngineAlreadyRunningError) {
+        runtimeLog.warn(
+          `Refusing to start engine for ${projectId}: ${err.message}`,
+        );
+      }
+      throw err;
+    });
+    this.singletonLocks.set(projectId, singleton);
+
     const engine = new ProjectEngine(
       runtimeConfig,
       this.centralCore,
       engineOptions,
     );
 
-    await engine.start();
+    try {
+      await engine.start();
+    } catch (err) {
+      // If engine start fails we must release the singleton so a retry can
+      // re-acquire it.
+      await this.releaseSingleton(projectId);
+      throw err;
+    }
 
     this.engines.set(projectId, engine);
     this.starting.delete(projectId);