fix: isolate repository publication locks by checkout

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/adapter-utils/src/workspace-restore-merge.test.ts

@@ -17,6 +17,7 @@ import {
   parseDirectorySnapshot,
   serializeDirectorySnapshot,
   withDirectoryMergeLock,
+  withDirectoryPublicationLock,
   WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE,
 } from "./workspace-restore-merge.js";
 
@@ -31,6 +32,42 @@ describe("workspace restore merge", () => {
     }
   });
 
+  it("locks absent publications independently while sharing the published target's merge lock", async () => {
+    const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-publication-lock-"));
+    cleanupDirs.push(root);
+    const env = { PAPERCLIP_HOME: path.join(root, "home"), PAPERCLIP_INSTANCE_ID: "publication" };
+    const first = path.join(root, "first"), second = path.join(root, "second");
+    await withDirectoryPublicationLock(first, async canonical => {
+      expect(await lstat(first).catch(() => null)).toBeNull();
+      await mkdir(canonical);
+      await withDirectoryPublicationLock(second, async other => {
+        await mkdir(other);
+        expect(other).not.toBe(canonical);
+      }, env);
+      const merge = vi.fn();
+      await expect(withDirectoryMergeLock(first, merge, env, undefined, 30)).rejects.toMatchObject({
+        code: WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE,
+      });
+      expect(merge).not.toHaveBeenCalled();
+    }, env);
+    await withDirectoryMergeLock(first, async canonical => expect(canonical).toBe(await realpath(first)), env);
+  });
+
+  it.skipIf(process.platform === "win32")("canonicalizes publication parents and rejects unsafe existing leaves", async () => {
+    const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-publication-alias-"));
+    cleanupDirs.push(root);
+    const env = { PAPERCLIP_HOME: path.join(root, "home"), PAPERCLIP_INSTANCE_ID: "publication" };
+    const parent = path.join(root, "parent"), alias = path.join(root, "alias");
+    await mkdir(parent); await symlink(parent, alias);
+    await withDirectoryPublicationLock(path.join(alias, "child"), async canonical => {
+      expect(canonical).toBe(path.join(await realpath(parent), "child"));
+    }, env);
+    await symlink(parent, path.join(parent, "child"));
+    await expect(withDirectoryPublicationLock(path.join(alias, "child"), async () => undefined, env)).rejects.toThrow("not a plain directory");
+    await rm(path.join(parent, "child")); await writeFile(path.join(parent, "child"), "occupied");
+    await expect(withDirectoryPublicationLock(path.join(alias, "child"), async () => undefined, env)).rejects.toThrow("not a plain directory");
+  });
+
   it("round-trips a deterministic durable snapshot and rejects traversal", async () => {
     const rootDir = await mkdtemp(
       path.join(os.tmpdir(), "paperclip-snapshot-"),

packages/adapter-utils/src/workspace-restore-merge.ts

@@ -504,6 +504,42 @@ export async function withDirectoryMergeLock<T>(
   // Canonicalize before we hash or lock: a retargeted symlink must not let the
   // lock protect one directory while the caller mutates another.
   const canonicalTargetDir = await fs.realpath(targetDir);
+  return withCanonicalDirectoryMergeLock(canonicalTargetDir, fn, env, diagnosticOperation, waitMs);
+}
+
+/** Lock an atomic directory publication before its target exists. The parent
+ * must already exist; existing leaf symlinks and non-directories are refused. */
+export async function withDirectoryPublicationLock<T>(
+  targetDir: string,
+  fn: (canonicalTargetDir: string) => Promise<T>,
+  env: NodeJS.ProcessEnv = process.env,
+): Promise<T> {
+  const absoluteTarget = path.resolve(targetDir);
+  const parent = await fs.realpath(path.dirname(absoluteTarget));
+  const canonicalTargetDir = path.join(parent, path.basename(absoluteTarget));
+  const assertSafeLeaf = async () => {
+    const leaf = await fs.lstat(canonicalTargetDir).catch((error: NodeJS.ErrnoException) => {
+      if (error.code === "ENOENT") return null;
+      throw error;
+    });
+    if (leaf && (!leaf.isDirectory() || leaf.isSymbolicLink())) {
+      throw new Error(`Directory publication target at ${canonicalTargetDir} is not a plain directory.`);
+    }
+  };
+  await assertSafeLeaf();
+  return withCanonicalDirectoryMergeLock(canonicalTargetDir, async () => {
+    await assertSafeLeaf();
+    return fn(canonicalTargetDir);
+  }, env);
+}
+
+async function withCanonicalDirectoryMergeLock<T>(
+  canonicalTargetDir: string,
+  fn: (canonicalTargetDir: string) => Promise<T>,
+  env: NodeJS.ProcessEnv,
+  diagnosticOperation?: DirectoryMergeLockOperation,
+  waitMs: number = LOCK_WAIT_MS,
+): Promise<T> {
   const lockRoot = await resolveDirectoryMergeLockRoot(env);
   const lockKey = createHash("sha256").update(canonicalTargetDir).digest("hex");
   const releaseLock = await acquireDirectoryMergeLock(path.join(lockRoot, `${lockKey}.lock`), diagnosticOperation, waitMs);

server/src/__tests__/task-workspace-repositories-persistence.test.ts

@@ -47,12 +47,14 @@ const support = await getEmbeddedPostgresTestSupport();
       await git("add", ".");
       await git("-c", "user.name=Fixture", "-c", "user.email=fixture@example.test", "commit", "-m", "source");
       const pin = await git("rev-parse", "HEAD");
+      const cloneStarted = path.join(root, "clone-started"), cloneReleased = path.join(root, "clone-released");
       // Exercise actual Git and atomic publication without external networking.
       // Only this fixture's exact remote is redirected; persist the real API origin.
       await writeFile(path.join(bin, "git"), `#!${process.execPath}\n` +
-        `const { spawnSync } = require('node:child_process'); const args = process.argv.slice(2);\n` +
+        `const { spawnSync } = require('node:child_process'); const fs = require('node:fs'); const args = process.argv.slice(2);\n` +
         `const remote = args.includes('clone') ? args.at(-2) : null;\n` +
         `if (remote && remote !== 'https://github.com/team/source') process.exit(97);\n` +
+        `if (remote) { fs.writeFileSync(${JSON.stringify(cloneStarted)}, 'started'); const deadline = Date.now() + 10000; while (!fs.existsSync(${JSON.stringify(cloneReleased)})) { if (Date.now() > deadline) process.exit(96); Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 20); } }\n` +
         `if (remote) args[args.length - 2] = ${JSON.stringify(source)};\n` +
         `const result = spawnSync(${JSON.stringify(realGit)}, args, { stdio: 'inherit' });\n` +
         `if (result.status) process.exit(result.status);\n` +
@@ -68,9 +70,34 @@ const support = await getEmbeddedPostgresTestSupport();
       expect(await lstat(checkout).catch(() => null)).toBeNull();
       const admission = { companyId, issueId: task.id, workspaceId: workspace.id, cwd: taskRoot,
         agentId, runId: randomUUID(), responsibleUserId: "local-board" };
+      const [readyWorkspace] = await db.insert(executionWorkspaces).values({ companyId, name: "Shared physical root", cwd: taskRoot,
+        mode: "shared_workspace", strategyType: "operator_directory" }).returning();
+      const [readyTask] = await db.insert(issues).values({ companyId, title: "Reuse another repository", executionWorkspaceId: readyWorkspace.id }).returning();
+      const readyIntent = await executionWorkspaceRepositoryService(db).request({ companyId, issueId: readyTask.id, actor,
+        request: { repository: { kind: "url", url: "https://github.com/team/ready" }, requestKey: "ready-repository" } });
+      const readyCheckout = path.join(taskRoot, readyIntent.repository.relativePath);
+      await mkdir(path.dirname(readyCheckout), { recursive: true });
+      await exec(realGit, ["clone", "--no-hardlinks", source, readyCheckout]);
+      await exec(realGit, ["-C", readyCheckout, "remote", "set-url", "origin", readyIntent.repository.repoUrl]);
+      await db.update(executionWorkspaceRepositories).set({ state: "ready", pinnedCommit: pin })
+        .where(eq(executionWorkspaceRepositories.id, readyIntent.operationId));
       // New service instances emulate the controller restart between request
       // and admission; concurrent admissions must publish one owned checkout.
-      const results = await Promise.all([1, 2].map(() => executionWorkspaceRepositoryService(db).prepareForAdmission(admission)));
+      const pending = Promise.all([1, 2].map(() => executionWorkspaceRepositoryService(db).prepareForAdmission(admission)));
+      let results: Awaited<typeof pending>;
+      let timeout: ReturnType<typeof setTimeout> | undefined;
+      try {
+        await vi.waitFor(async () => expect(await lstat(cloneStarted).catch(() => null)).not.toBeNull());
+        const ready = await Promise.race([
+          executionWorkspaceRepositoryService(db).prepareForAdmission({ ...admission, workspaceId: readyWorkspace.id, issueId: readyTask.id }),
+          new Promise<never>((_, reject) => { timeout = setTimeout(() => reject(new Error("Unrelated repository reuse waited for a blocked clone")), 2000); }),
+        ]);
+        expect(ready[0]).toMatchObject({ id: readyIntent.operationId, pinnedCommit: pin });
+      } finally {
+        clearTimeout(timeout);
+        await writeFile(cloneReleased, "released");
+        results = await pending;
+      }
       for (const prepared of results) expect(prepared).toEqual([expect.objectContaining({ id: intent.operationId, pinnedCommit: pin })]);
       expect(await readFile(path.join(checkout, "source.txt"), "utf8")).toBe("original source\n");
       expect(JSON.parse(await readFile(path.join(checkout, ".git", "paperclip-workspace-owner.json"), "utf8")))

server/src/services/execution-workspace-repositories.ts

@@ -7,7 +7,7 @@ import path from "node:path";
 import { and, eq, or, sql } from "drizzle-orm";
 import { activityLog, executionWorkspaceRepositories, executionWorkspaces, issues, type Db } from "@paperclipai/db";
 import { prepareWorkspaceRepositorySchema, type PrepareWorkspaceRepository, type ProjectRepository } from "@paperclipai/shared";
-import { withDirectoryMergeLock } from "@paperclipai/adapter-utils/workspace-restore-merge";
+import { withDirectoryPublicationLock } from "@paperclipai/adapter-utils/workspace-restore-merge";
 import { conflict, forbidden, notFound } from "../errors.js";
 import { accessService } from "./access.js";
 import { assertTaskWorkspaceAccess } from "./task-workspace-source-access.js";
@@ -135,9 +135,7 @@ export function executionWorkspaceRepositoryService(db: Db) {
       const source = resolveTaskRepository(row.catalogRepositoryId ? { kind: "catalog", id: row.catalogRepositoryId } : { kind: "url", url: row.repoUrl }, available);
       if (source.repositoryIdentity !== row.repositoryIdentity || row.relativePath !== taskRepositoryRelativePath(row.repositoryIdentity)) throw conflict("Repository preparation receipt does not match its source");
       const cwd = path.join(root, row.relativePath);
-      // A new checkout does not exist yet. Lock its validated, existing parent
-      // so publication and reuse serialize without creating an unowned target.
-      await withDirectoryMergeLock(repositoryRoot, async () => {
+      await withDirectoryPublicationLock(cwd, async () => {
         const existing = await fs.lstat(cwd).catch(() => null);
         if (existing && (!existing.isDirectory() || existing.isSymbolicLink() || await fs.realpath(cwd) !== cwd)) throw conflict("Repository checkout path is not a contained directory");
         if (existing) {