feat(workspace): Phase C U3 — per-repo land lease (serialize same-sub-repo lands)

landWorkspaceTask now holds a per-repo land lease around each landOneRepo call:
a new activeSessionRegistry kind "workspace-repo-land" keyed on the sub-repo
absolute path, registered synchronously before the per-repo try and released in
a finally (on success and failure, only yanking our own taskId+ownerKey entry —
never a foreign/different-kind entry). Two workspace tasks landing the same
sub-repo serialize; the loser throws the retryable WorkspaceRepoLandBusyError,
which reuses the U2 partial-land retry/park machinery (consume a mergeRetry,
backoff re-enqueue up to MAX skipping landed repos, then operator-park). Disjoint
sub-repos never falsely serialize.

The lease is for serialization / clean-room-collision avoidance, not ref
correctness — advanceIntegrationBranchRef's CAS already makes interleaved
update-ref safe. Distinct from the execution-phase "workspace-repo-acquire" lease
(different kind, different lifecycle phase, each ignores the other's entry).

3 new tests (serialize, independence, release-on-failure); oracle (56) + U1/U2
(idempotency) stay green. Gate: build, typecheck, lint, test:gate (649+58).

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

.changeset/workspace-per-repo-land-lease.md

@@ -0,0 +1,5 @@
+---
+"@runfusion/fusion": minor
+---
+
+Workspace mode (Phase C U3): serialize concurrent same-sub-repo lands with a per-repo file-scope lease. When two workspace tasks try to land onto the SAME sub-repo's local integration ref at the same time, the merge phase now registers the sub-repo's absolute path in the path-keyed active-session registry under a distinct `workspace-repo-land` kind before each land and releases it in a `finally` (on land success or failure — no stuck lock). A second task contending for the same sub-repo fast-fails with a retryable `WorkspaceRepoLandBusyError`, which the existing partial-land auto-retry-then-park dispatch handles (consume a `mergeRetry`, re-enqueue with backoff, then operator-park). Disjoint sub-repos lease different paths and never serialize against each other. The lease prevents clean-room ai-merge worktree collisions; ref correctness is already guaranteed by `advanceIntegrationBranchRef`'s CAS (concurrent-advance → rebuild).

packages/engine/src/__tests__/workspace-merger-lease.test.ts

@@ -0,0 +1,272 @@
+/*
+FNXC:Workspace 2026-06-22-02:10 (Phase C U3, KTD4):
+Per-repo LAND lease tests. They drive the REAL `landWorkspaceTask` against a REAL
+two-repo git fixture (createWorkspaceFixture) and assert the lease seam directly on
+the REAL module-level `activeSessionRegistry` singleton (FN-5048: narrow seam — we
+assert registry state + a merge-agent spy, NO real concurrent processes, NO
+mock-the-world; the AI merge/review agents are injected so no real AI calls happen
+and the squash is a plain `git merge --squash`).
+
+The lease is keyed by the sub-repo ABSOLUTE path under kind "workspace-repo-land".
+It is for SERIALIZATION / clean-room-collision avoidance only — `advanceIntegration
+BranchRef`'s CAS already makes the interleaved `update-ref` correct — so we assert
+serialization behavior (one wins, the other fast-fails) and that the lease never leaks.
+
+Coverage (FN-5893 surfaces):
+- concurrency: two tasks landing the SAME sub-repo → one acquires the land lease,
+  the other FAST-FAILS with WorkspaceRepoLandBusyError; no interleaved update-ref on
+  that repo's ref (the loser advances nothing). Lease kind/path asserted while held.
+- independence: disjoint sub-repos (task1→repo-a, task2→repo-b) → both proceed, no
+  false serialization (neither sees the other's lease path).
+- cleanup: a repo land that THROWS → the lease for that path is released (not stuck),
+  so a subsequent land of the same repo can acquire it.
+*/
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { EventEmitter } from "node:events";
+import { execSync } from "node:child_process";
+import { writeFileSync } from "node:fs";
+import path from "node:path";
+import type { Task, TaskStore } from "@fusion/core";
+import { landWorkspaceTask, WorkspaceRepoLandBusyError } from "../merger-ai.js";
+import { activeSessionRegistry } from "../active-session-registry.js";
+import { createWorkspaceFixture, hasGit, type WorkspaceFixture } from "./_workspace-fixture.js";
+
+const describeIfGit = hasGit ? describe : describe.skip;
+
+const BRANCH = "fusion/fn-3003";
+const LAND_KIND = "workspace-repo-land";
+
+function configureIdentity(dir: string): void {
+  execSync('git config user.email "test@example.com"', { cwd: dir, stdio: "pipe" });
+  execSync('git config user.name "Test"', { cwd: dir, stdio: "pipe" });
+}
+
+interface RecordingStore extends EventEmitter {
+  task: Task;
+  moveTaskCalls: Array<{ id: string; column: string }>;
+}
+
+/** A store that persists workspaceWorktrees/mergeDetails on one in-memory task. */
+function createStore(task: Task): TaskStore & RecordingStore {
+  const emitter = new EventEmitter();
+  const moveTaskCalls: Array<{ id: string; column: string }> = [];
+  const store = Object.assign(emitter, {
+    task,
+    moveTaskCalls,
+    getSettings: vi.fn().mockResolvedValue({ autoMerge: false }),
+    updateTask: vi.fn(async (_id: string, patch: Partial<Task>) => {
+      Object.assign(store.task, patch);
+      return undefined;
+    }),
+    logEntry: vi.fn().mockResolvedValue(undefined),
+    appendAgentLog: vi.fn().mockResolvedValue(undefined),
+    getTask: vi.fn(async () => store.task),
+    moveTask: vi.fn((id: string, column: string) => {
+      moveTaskCalls.push({ id, column });
+      store.task.column = column as Task["column"];
+      return Promise.resolve(store.task);
+    }),
+    upsertTaskCommitAssociation: vi.fn().mockResolvedValue(undefined),
+    accumulateTokenUsage: vi.fn().mockResolvedValue(undefined),
+  }) as unknown as TaskStore & RecordingStore;
+  return store;
+}
+
+/** Add a real `fusion/<id>` branch to a sub-repo with one own non-conflicting commit. */
+function addRepoBranchWithEdit(fx: WorkspaceFixture, repoRel: string, taskId: string, content: string): void {
+  const repoDir = fx.repoPath(repoRel);
+  const worktreePath = path.join(repoDir, `.wt-${taskId}`);
+  fx.git(repoRel, `git worktree add -b ${BRANCH} ${worktreePath} HEAD`);
+  configureIdentity(worktreePath);
+  writeFileSync(path.join(worktreePath, "feature.txt"), content, "utf-8");
+  execSync("git add feature.txt", { cwd: worktreePath, stdio: "pipe" });
+  execSync(`git commit -m "feat(${taskId}): add feature in ${repoRel}"`, { cwd: worktreePath, stdio: "pipe" });
+  fx.git(repoRel, `git worktree remove --force ${worktreePath}`);
+}
+
+/** A merge agent that performs the real squash in the clean room (no AI). */
+function squashMergeAgent(branch: string, onEnter?: (cwd: string) => void | Promise<void>) {
+  return async (cwd: string): Promise<void> => {
+    if (onEnter) await onEnter(cwd);
+    configureIdentity(cwd);
+    try {
+      execSync(`git merge --squash ${branch}`, { cwd, stdio: "pipe" });
+    } catch {
+      // squash reported conflicts — fall through to the unmerged check.
+    }
+    const unmerged = execSync("git ls-files -u", { cwd, encoding: "utf-8" }).trim();
+    if (unmerged.length > 0) throw new Error("merge conflict: unresolved paths in clean room");
+    const staged = execSync("git diff --cached --name-only", { cwd, encoding: "utf-8" }).trim();
+    if (staged.length === 0) return;
+    execSync(`git commit -m "${branch}: squashed"`, { cwd, stdio: "pipe" });
+  };
+}
+
+const approveReviewAgent = async (): Promise<string> => "REVIEW_VERDICT: approve";
+
+function makeTask(id: string, workspaceWorktrees: Task["workspaceWorktrees"]): Task {
+  return {
+    id,
+    title: "Workspace merge task",
+    description: "",
+    column: "in-review",
+    branch: BRANCH,
+    dependencies: [],
+    steps: [],
+    currentStep: 0,
+    log: [],
+    workspaceWorktrees,
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+  } as Task;
+}
+
+describeIfGit("landWorkspaceTask — per-repo land lease (Phase C U3, KTD4)", () => {
+  let fx: WorkspaceFixture;
+  afterEach(() => {
+    fx?.cleanup();
+    activeSessionRegistry.clear();
+    vi.restoreAllMocks();
+  });
+  beforeEach(() => activeSessionRegistry.clear());
+
+  it("concurrency: two tasks landing the SAME sub-repo serialize — one acquires the land lease, the other fast-fails (no interleaved update-ref)", async () => {
+    fx = await createWorkspaceFixture(["repo-a"]);
+    addRepoBranchWithEdit(fx, "repo-a", "FN-3001", "a feature\n");
+    const repoAbs = fx.repoPath("repo-a");
+
+    const task1 = makeTask("FN-3001", { "repo-a": { worktreePath: repoAbs, branch: BRANCH } });
+    const task2 = makeTask("FN-3001", { "repo-a": { worktreePath: repoAbs, branch: BRANCH } });
+    // Distinct task IDs so the lease owner check (taskId !== holder) triggers.
+    task2.id = "FN-3002";
+    const store1 = createStore(task1);
+    const store2 = createStore(task2);
+
+    let loserError: unknown;
+    const tipBefore = fx.git("repo-a", "git rev-parse refs/heads/main");
+
+    // task1's merge agent blocks until task2 has tried (and failed) to acquire the
+    // land lease for the SAME sub-repo path. While task1 holds the lease we assert it
+    // is registered under the right kind + path; task2 fast-fails with the busy error.
+    const winner = landWorkspaceTask(store1, store1.task, fx.rootDir, {}, {
+      mergeAgent: squashMergeAgent(BRANCH, async () => {
+        // task1 now holds the land lease for repo-a.
+        const held = activeSessionRegistry.lookupByPath(repoAbs);
+        expect(held?.kind).toBe(LAND_KIND);
+        expect(held?.taskId).toBe("FN-3001");
+
+        // task2 attempts the same sub-repo concurrently → must fast-fail.
+        try {
+          await landWorkspaceTask(store2, store2.task, fx.rootDir, {}, {
+            mergeAgent: squashMergeAgent(BRANCH),
+            reviewAgent: approveReviewAgent,
+          });
+        } catch (err) {
+          loserError = err;
+        }
+        // The loser advanced NOTHING: the ref is still at the pre-land tip.
+        expect(fx.git("repo-a", "git rev-parse refs/heads/main")).toBe(tipBefore);
+      }),
+      reviewAgent: approveReviewAgent,
+    });
+
+    const result = await winner;
+
+    // Winner landed.
+    expect(result.allLanded).toBe(true);
+    expect(result.repos[0].status).toBe("landed");
+    expect(fx.git("repo-a", "git rev-parse refs/heads/main")).not.toBe(tipBefore);
+
+    // Loser fast-failed with the retryable busy error (serialized, not broken).
+    expect(loserError).toBeInstanceOf(WorkspaceRepoLandBusyError);
+    expect((loserError as WorkspaceRepoLandBusyError).retryable).toBe(true);
+    expect((loserError as WorkspaceRepoLandBusyError).holderTaskId).toBe("FN-3001");
+
+    // Lease released after the winner finished — no leak.
+    expect(activeSessionRegistry.lookupByPath(repoAbs)).toBeNull();
+  });
+
+  it("independence: disjoint sub-repos land without contention (no false serialization)", async () => {
+    fx = await createWorkspaceFixture(["repo-a", "repo-b"]);
+    addRepoBranchWithEdit(fx, "repo-a", "FN-3001", "a feature\n");
+    addRepoBranchWithEdit(fx, "repo-b", "FN-3002", "b feature\n");
+    const repoAAbs = fx.repoPath("repo-a");
+    const repoBAbs = fx.repoPath("repo-b");
+
+    const task1 = makeTask("FN-3001", { "repo-a": { worktreePath: repoAAbs, branch: BRANCH } });
+    const task2 = makeTask("FN-3002", { "repo-b": { worktreePath: repoBAbs, branch: BRANCH } });
+    const store1 = createStore(task1);
+    const store2 = createStore(task2);
+
+    let task2Error: unknown;
+    let task2Landed = false;
+
+    // task1 lands repo-a; mid-land it kicks off task2 landing the DISJOINT repo-b.
+    // task2 leases a DIFFERENT path, so it must NOT serialize against task1.
+    const t1 = landWorkspaceTask(store1, store1.task, fx.rootDir, {}, {
+      mergeAgent: squashMergeAgent(BRANCH, async () => {
+        // While task1 holds repo-a's lease, repo-b's lease is unheld.
+        expect(activeSessionRegistry.lookupByPath(repoAAbs)?.kind).toBe(LAND_KIND);
+        expect(activeSessionRegistry.lookupByPath(repoBAbs)).toBeNull();
+        try {
+          const r2 = await landWorkspaceTask(store2, store2.task, fx.rootDir, {}, {
+            mergeAgent: squashMergeAgent(BRANCH),
+            reviewAgent: approveReviewAgent,
+          });
+          task2Landed = r2.allLanded;
+        } catch (err) {
+          task2Error = err;
+        }
+      }),
+      reviewAgent: approveReviewAgent,
+    });
+
+    const r1 = await t1;
+
+    // Both proceeded — no false serialization.
+    expect(task2Error).toBeUndefined();
+    expect(task2Landed).toBe(true);
+    expect(r1.allLanded).toBe(true);
+    expect(fx.git("repo-a", "git rev-parse refs/heads/main")).not.toBe(
+      fx.git("repo-a", "git rev-parse fusion/fn-3003^"),
+    );
+    // Both leases released.
+    expect(activeSessionRegistry.lookupByPath(repoAAbs)).toBeNull();
+    expect(activeSessionRegistry.lookupByPath(repoBAbs)).toBeNull();
+  });
+
+  it("cleanup: a land failure releases the lease (not stuck) so a subsequent land can acquire", async () => {
+    fx = await createWorkspaceFixture(["repo-a"]);
+    addRepoBranchWithEdit(fx, "repo-a", "FN-3001", "a feature\n");
+    const repoAbs = fx.repoPath("repo-a");
+
+    const task = makeTask("FN-3001", { "repo-a": { worktreePath: repoAbs, branch: BRANCH } });
+    const store = createStore(task);
+
+    // A merge agent that throws → landOneRepo fails → the per-repo land lease finally
+    // must release the lease even on failure.
+    const throwingAgent = async (): Promise<void> => {
+      // Lease is held at this point.
+      expect(activeSessionRegistry.lookupByPath(repoAbs)?.kind).toBe(LAND_KIND);
+      throw new Error("synthetic clean-room failure");
+    };
+
+    const failed = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
+      mergeAgent: throwingAgent,
+      reviewAgent: approveReviewAgent,
+    });
+    expect(failed.allLanded).toBe(false);
+    expect(failed.repos[0].status).toBe("failed");
+    // Lease was released despite the failure — NOT stuck.
+    expect(activeSessionRegistry.lookupByPath(repoAbs)).toBeNull();
+
+    // A subsequent land of the SAME repo can acquire (real squash this time).
+    const retry = await landWorkspaceTask(store, store.task, fx.rootDir, {}, {
+      mergeAgent: squashMergeAgent(BRANCH),
+      reviewAgent: approveReviewAgent,
+    });
+    expect(retry.allLanded).toBe(true);
+    expect(retry.repos[0].status).toBe("landed");
+    expect(activeSessionRegistry.lookupByPath(repoAbs)).toBeNull();
+  });
+});

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

@@ -6,8 +6,21 @@ sub-repo absolute path (NOT the worktree path) so two concurrent workspace tasks
 contending for the SAME sub-repo are serialized. Keeping it distinct from
 "executor"/"step-session" means it does not collide with the executor's later
 session registration on the produced worktree path.
+
+FNXC:Workspace 2026-06-22-02:10 (Phase C U3, KTD4):
+"workspace-repo-land" is a DISTINCT registry kind for the LAND-time (merge phase)
+same-sub-repo lease. Like the acquire kind it is keyed by the sub-repo ABSOLUTE
+path, but it guards a different lifecycle scope: two workspace tasks landing the
+SAME sub-repo onto its local integration ref are serialized so their clean-room
+ai-merge worktrees do not collide. This lease is for SERIALIZATION / clean-room-
+collision avoidance only — it is NOT what makes the interleaved `update-ref`
+correct. `advanceIntegrationBranchRef`'s CAS already makes a concurrent advance
+safe by construction (concurrent-advance → rebuild). The acquire lease (execution
+phase) and the land lease (merge phase) never overlap in time on the same path, so
+keeping them distinct kinds (each released in its own `finally`) means a stale
+entry of one kind can never be mistaken for a live hold of the other.
 */
-export type ActiveSessionKind = "executor" | "step-session" | "workflow-step" | "step-session-parallel" | "ai-merge" | "workspace-repo-acquire";
+export type ActiveSessionKind = "executor" | "step-session" | "workflow-step" | "step-session-parallel" | "ai-merge" | "workspace-repo-acquire" | "workspace-repo-land";
 
 export interface ActiveSessionRegistration {
   taskId: string;

packages/engine/src/merger-ai.ts

@@ -1403,7 +1403,48 @@ The partial-land retry/park policy (consume a mergeRetry, auto-retry skipping la
 repos up to MAX, then operator-park) is wired at the engine dispatch (project-engine.ts),
 NOT here: this function reports the partial via `allLanded:false` and the dispatch drives
 the retry seam.
+
+FNXC:Workspace 2026-06-22-02:10 (Phase C U3, KTD4):
+Per-repo LAND lease. Before each `landOneRepo` we register the sub-repo ABSOLUTE
+path in the path-keyed activeSessionRegistry under kind "workspace-repo-land" and
+release it in a per-repo `finally` (so the lease is freed on land success OR land
+failure — no stuck lock). If another task already holds the land lease for that
+sub-repo path we FAST-FAIL the whole `landWorkspaceTask` with a retryable
+`WorkspaceRepoLandBusyError`, which the U2 partial-land retry/park machinery
+(project-engine dispatch) already handles — reusing that path instead of
+reimplementing a waiting lock. The lease serializes same-sub-repo lands so two
+tasks' clean-room ai-merge worktrees do not collide; it is NOT what makes the
+interleaved `update-ref` correct — `advanceIntegrationBranchRef`'s CAS already
+guarantees ref correctness (concurrent-advance → rebuild). Disjoint sub-repos lease
+DIFFERENT paths, so they never serialize against each other (no false contention).
+This lease is a DIFFERENT scope/kind from the execution-phase
+"workspace-repo-acquire" lease and from `landOneRepo`'s own inner "ai-merge"
+clean-room registration on the temp worktree path — none of the three collide.
 */
+
+/** FNXC:Workspace 2026-06-22-02:10 (Phase C U3, KTD4): ownerKey for the land-time lease. */
+const WORKSPACE_REPO_LAND_OWNER_KEY = "workspace-repo-land";
+
+/*
+FNXC:Workspace 2026-06-22-02:10 (Phase C U3, KTD4):
+Thrown when a second workspace task tries to land a sub-repo already inside another
+task's land critical section. Distinct from a generic land failure so the engine
+dispatch (and tests) can tell "serialized, retry later" apart from "this land is
+broken". Carries `retryable = true` so the existing partial-land auto-retry/park
+path treats it as a transient contention, not a terminal failure.
+*/
+export class WorkspaceRepoLandBusyError extends Error {
+  public readonly retryable = true;
+  constructor(
+    public readonly repoRel: string,
+    public readonly holderTaskId: string,
+    public readonly requestingTaskId: string,
+  ) {
+    super(`workspace sub-repo ${repoRel} land is in progress for task ${holderTaskId}`);
+    this.name = "WorkspaceRepoLandBusyError";
+  }
+}
+
 export async function landWorkspaceTask(
   store: TaskStore,
   task: Task,
@@ -1476,6 +1517,32 @@ export async function landWorkspaceTask(
       continue;
     }
 
+    /*
+    FNXC:Workspace 2026-06-22-02:10 (Phase C U3, KTD4):
+    Same-sub-repo LAND lease. Register the sub-repo absolute path BEFORE landing so
+    two tasks landing the SAME sub-repo are serialized (their clean-room ai-merge
+    worktrees would otherwise collide). The lookupByPath → registerPath pair stays in
+    ONE synchronous slice (no `await` between them) so the claim is atomic — an
+    interleaved await would let a second task pass the gate before we register. If
+    another task holds the land lease we FAST-FAIL with a retryable busy error; the
+    U2 dispatch auto-retry/park path handles it (no waiting lock reimplemented here).
+    We only treat a HELD entry of OUR OWN land ownerKey as contention, so a stale
+    entry of a different kind on this path (e.g. a leftover acquire entry) is ignored.
+    */
+    const landLeaseHolder = activeSessionRegistry.lookupByPath(repoRootDir);
+    if (
+      landLeaseHolder &&
+      landLeaseHolder.ownerKey === WORKSPACE_REPO_LAND_OWNER_KEY &&
+      landLeaseHolder.taskId !== taskId
+    ) {
+      throw new WorkspaceRepoLandBusyError(repoRel, landLeaseHolder.taskId, taskId);
+    }
+    activeSessionRegistry.registerPath(repoRootDir, {
+      taskId,
+      kind: "workspace-repo-land",
+      ownerKey: WORKSPACE_REPO_LAND_OWNER_KEY,
+    });
+
     try {
       const landResult = await landOneRepo(store, repoRootDir, entry.branch, integrationBranch, {
         taskId, settings, audit, log, setStatus, maxPasses,
@@ -1505,6 +1572,18 @@ export async function landWorkspaceTask(
       // `landedSha` is persisted, so the engine dispatch's auto-retry re-runs this
       // loop and the landed predicate above skips them (only the failed repo retries).
       break;
+    } finally {
+      /*
+      FNXC:Workspace 2026-06-22-02:10 (Phase C U3, KTD4):
+      Release the land lease — on land SUCCESS or land FAILURE — but ONLY when WE hold
+      it (own taskId + own ownerKey), so a future-acquire path's entry on this path is
+      never yanked. The fast-fail busy throw above happens BEFORE registerPath, so a
+      serialized loser never unregisters the winner's lease.
+      */
+      const held = activeSessionRegistry.lookupByPath(repoRootDir);
+      if (held && held.taskId === taskId && held.ownerKey === WORKSPACE_REPO_LAND_OWNER_KEY) {
+        activeSessionRegistry.unregisterPath(repoRootDir);
+      }
     }
   }
 

packages/engine/src/project-engine.ts

@@ -2471,8 +2471,18 @@ export class ProjectEngine {
           // — mirroring the conflict-retry seam below. Detect by err.name (robust across
           // the package boundary). Manual merges fall through to rejectMergeResolvers at
           // the hasManualResolver early-return below (no auto-retry for manual).
+          /*
+          FNXC:Workspace 2026-06-22-02:10 (Phase C U3, KTD4):
+          A `WorkspaceRepoLandBusyError` (a second task holds the same sub-repo's
+          land lease) is ALSO retryable here — it is transient contention, not a
+          terminal failure. Route it through the SAME auto-retry-then-park seam (it
+          consumes a mergeRetry and re-enqueues with backoff; a re-run skips
+          already-landed repos and finds the lease freed). Detect by err.name across
+          the package boundary, same as the partial-land error.
+          */
           const isWorkspacePartialLand =
-            err instanceof Error && err.name === "WorkspacePartialLandError";
+            err instanceof Error &&
+            (err.name === "WorkspacePartialLandError" || err.name === "WorkspaceRepoLandBusyError");
           if (isWorkspacePartialLand && !hasManualResolver) {
             const wsSettings = await store.getSettings().catch(() => ({ maxAutoMergeRetries: undefined }));
             const wsTask = await store.getTask(taskId).catch(() => null);