fix(workspace): bound queued source-lock admission for imports (#20370)

## Thinking Path

> - Paperclip manages AI agents and their work.
> - Existing local workspaces need durable imports before a volume-free
deployment.
> - Imports must acquire the same source locks as workspace writers.
> - Try-only admission can repeatedly lose to background maintenance,
even when each writer finishes.
> - This pull request adds an optional bounded wait to the canonical
migration APIs.
> - Operators can join the source-lock queue while all existing
preservation checks remain active.

## Linked Issues or Issue Description

**What happened?**

An online import can repeatedly return `workspace_source_busy` while
background workspace maintenance reacquires its project source lock. The
import does not join the PostgreSQL lock queue.

**Expected behavior**

An administrative import can wait for a finite writer to finish. One
wait budget covers the complete source scope. Timeout or cancellation
must release acquired locks before the import reserves a journal.

**Steps to reproduce**

1. Hold the source lock for an eligible local workspace.
2. Start an online import while the writer holds that lock.
3. Release the writer. Before this change, the import has already
refused admission. With `sourceLockWaitMs`, the import acquires the
source scope and commits normally.

**Paperclip version or commit**

Reproduced on master `0e3fdebea8257bdff11573081c76b306f12db230`.

**Deployment mode**

External PostgreSQL with an online workspace import. Related
preservation work: https://github.com/paperclipai/paperclip/pull/20311.

## What Changed

- Add optional `sourceLockWaitMs` to singleton and cohort migration
APIs. Omitted options keep current admission behavior.
- Queue the sorted source keys with one decreasing wait budget. Abort
cancels only the dedicated backend identified by PID and backend start
time.
- Use an independent source transaction for queued singleton imports.
Keep lifecycle checks, fresh source checks, journal ownership, and
publication after commit.
- Document the canonical API option. Add real PostgreSQL importer
regressions for release, timeout, abort, and partial-key cleanup.
- Correct the inherited task-case lookup summary so the existing
search-discovery regression passes without changing its assertion or
search algorithm.

## Verification

- Baseline RED: the new singleton regression fails on the unchanged
product because the import returns busy without queueing.
- `pnpm exec vitest run
server/src/services/remote-workspace-migration.test.ts -t 'queued|bounds
the whole source queue'`: 20 passed, including five new cases.
- `pnpm exec vitest run
server/src/services/isolated-remote-workspace-source-lock.test.ts`: 4
passed.
- Server layer typecheck, `node scripts/check-module-boundaries.mjs`,
and `git diff --check` passed.
- Tests used Node 24.22.2 and the existing embedded PostgreSQL fixture.
Hosted rollout verification remains pending.
- CI build, typecheck, browser suites, all serialized suites, and the
other general shards passed. The final server shard failed only the
inherited `case-history-api-discovery.test.ts:14` search-ranking
assertion. Its unchanged local regression reproduces on the base and
passes 3/3 with a separate one-line route-summary correction. The shared
one-line summary correction is included in follow-up commit
`8163ffa9d4728134676b84c8c242a5bf62c802b5`; the unchanged case-discovery
suite passes 3/3 locally. The original CI attempt remains failed:
https://github.com/paperclipai/paperclip/actions/runs/42704666536/job/129458605783.

## Risks

Queued admission can hold earlier source keys while it waits for a later
key. The caller controls the finite budget. The abort path opens a
short-lived dedicated cancellation connection. If cancellation cannot
connect, PostgreSQL's remaining lock timeout still bounds the wait. This
option must be called at the root of administrative work; an external
scope that already holds the same source locks can cause
self-contention. The existing pre-held journal-resume path must keep its
omitted-option behavior.

## Model Used

OpenAI Codex, based on GPT-6, with tool use and code execution. The
exact deployment identifier and context-window size are not exposed in
this session.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` /
`Closes: #` / `Refs: #` OR (b) described the issue in-PR following the
relevant issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

doc/plans/2026-11-07-online-workspace-migration-entry.md

@@ -96,6 +96,15 @@ The entry does not assert stopped writers, terminate an agent, clear a hold or
 change remote authority. An unrelated active run is not a global-idle blocker.
 An actual scoped writer, foreign alias or unsupported source still refuses.
 
+Programmatic administrative calls to `migrateRemoteWorkspace` and
+`migrateRemoteWorkspaceCohort` can pass `sourceLockWaitMs` with their existing
+`signal` to join the source-lock queue. The budget covers acquisition of the
+complete sorted source scope, rather than restarting for each key. Timeout or
+abort rolls back that admission before any journal reservation. This opt-in uses
+an independent source transaction and preserves the lifecycle and publication
+checks; activity is published after commit. Omitted options retain try-only
+admission. This is a canonical API option, not an additional JSON entry field.
+
 Before each apply, use the existing per-operation memory, CPU and temporary-disk
 resource preflight. A concurrency bound is not resource admission. Budget every
 logical capture and up to two nested-child verifications. Qualify one single

server/src/services/remote-workspace-migration.test.ts

@@ -7,7 +7,7 @@ import { promisify } from "node:util";
 import { createServer, type Server } from "node:http";
 import { pathToFileURL } from "node:url";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
-import { and, eq } from "drizzle-orm";
+import { and, eq, sql } from "drizzle-orm";
 import { activityLog, agents, environmentLeases, heartbeatRuns, nativeRunFinalizations, nativeRunResults, completionContracts, applyPendingMigrations, closeRegisteredClients, companies, createDb, executionWorkspaceSnapshots,
   executionWorkspaceRepositories, executionWorkspaceRuntimeLeases, executionWorkspaces, issueRecoveryActions, issues, projects, projectWorkspaces, workspaceOperations, workspaceRuntimeServices } from "@paperclipai/db";
 import { readRemoteWorkspaceAuthority } from "@paperclipai/shared";
@@ -16,7 +16,7 @@ import { createStorageService } from "../storage/service.js";
 import type { StorageService } from "../storage/types.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
 import { migrateWorkspaceTrees } from "./remote-workspace-migration-batch.js";
-import { inventoryRemoteWorkspaceMigration, migrateRemoteWorkspace } from "./remote-workspace-migration.js";
+import { inventoryRemoteWorkspaceMigration, migrateRemoteWorkspace, migrateRemoteWorkspaceCohort } from "./remote-workspace-migration.js";
 import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "./remote-workspace-snapshot-storage.js";
 import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js";
 import * as workspaceArchive from "./remote-workspace-snapshot-archive.js";
@@ -35,7 +35,7 @@ import { managedProjectCheckoutName } from "./managed-project-checkout-name.js";
 import { captureGitWorkspaceContents, readManagedCheckoutContents } from "./git-workspace-contents.js";
 import { isolatedRemoteSourcePath } from "./isolated-remote-workspace-source.js";
 import { verifiedGitCacheSource } from "./git-workspace-restoration.js";
-import { withWorkspaceSourceLocks } from "./workspace-source-lock.js";
+import { withIndependentWorkspaceSourceLocks, withWorkspaceSourceLocks } from "./workspace-source-lock.js";
 import { inspectOnlineWorkspaceSource } from "./online-workspace-source.js";
 
 const execute = promisify(execFile);
@@ -76,6 +76,84 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
     return { companyId, workspaceId, issueId: issue!.id, cwd };
   }
   async function workspace(id: string) { return (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, id)))[0]!; }
+  async function queuedFixture() {
+    const original = await seed();
+    vi.stubEnv("PAPERCLIP_HOME", await fs.realpath(root));
+    const cwd = path.join(resolvePaperclipInstanceRoot(), "isolated-workspaces", original.companyId, original.issueId);
+    await fs.mkdir(path.dirname(cwd), { recursive: true }); await fs.rename(original.cwd, cwd);
+    await db.update(executionWorkspaces).set({ cwd, providerRef: cwd }).where(eq(executionWorkspaces.id, original.workspaceId));
+    return { ...original, cwd };
+  }
+  async function sourceWriter(companyId: string, issueIds: string[]) {
+    let entered!: () => void, release!: () => void;
+    const ready = new Promise<void>(resolve => { entered = resolve; });
+    const held = new Promise<void>(resolve => { release = resolve; });
+    const done = withIndependentWorkspaceSourceLocks(db, { companyId, issueIds }, async () => { entered(); await held; });
+    await ready;
+    return { release, done };
+  }
+  const queuedBackends = async () => db.execute(sql`select pid from pg_stat_activity where datname=current_database()
+    and wait_event_type='Lock' and query like 'select pg_advisory_xact_lock%'`);
+  it.each(["single", "cohort"] as const)("commits a queued %s import after the finite source writer releases", async kind => {
+    const f = await queuedFixture(), writer = await sourceWriter(f.companyId, [f.issueId]);
+    let settled = false;
+    const importing = (kind === "single"
+      ? migrateRemoteWorkspace({ db, storage, ...f, apply: true, sourceLockWaitMs: 5000 })
+      : migrateRemoteWorkspaceCohort({ db, storage, companyId: f.companyId, workspaceIds: [f.workspaceId],
+        apply: true, signal: AbortSignal.timeout(10_000), sourceLockWaitMs: 5000 }))
+      .then(value => ({ value }), error => ({ error })).finally(() => { settled = true; });
+    try {
+      await expect.poll(async () => (await queuedBackends()).length, { timeout: 2000 }).toBe(1);
+      expect(settled).toBe(false);
+      expect(await db.select().from(workspaceOperations)).toHaveLength(0);
+      writer.release(); await writer.done;
+      expect(await importing).toMatchObject({ value: { applied: true } });
+      expect(await db.select().from(workspaceOperations)).toEqual([expect.objectContaining({ status: "succeeded" })]);
+      expect(await db.select().from(executionWorkspaceSnapshots)).toEqual([expect.objectContaining({ status: "committed" })]);
+      expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).not.toBeNull();
+      expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n");
+    } finally { writer.release(); await writer.done; await importing; }
+  });
+  it.each(["timeout", "abort"] as const)("leaves the original and journal intact on queued admission %s", async mode => {
+    const f = await queuedFixture(), before = await workspace(f.workspaceId);
+    const writer = await sourceWriter(f.companyId, [f.issueId]), abort = new AbortController(), reason = new Error("cancel queued import");
+    const importing = migrateRemoteWorkspace({ db, storage, ...f, apply: true, sourceLockWaitMs: mode === "timeout" ? 300 : 5000,
+      signal: abort.signal }).then(value => ({ value }), error => ({ error }));
+    try {
+      await expect.poll(async () => (await queuedBackends()).length, { timeout: 2000 }).toBe(1);
+      const cancelledAt = performance.now();
+      if (mode === "abort") abort.abort(reason);
+      expect(await importing).toMatchObject({ error: mode === "abort" ? reason : expect.objectContaining({ message: "workspace_source_busy" }) });
+      expect(performance.now() - cancelledAt).toBeLessThan(2000);
+      expect(await workspace(f.workspaceId)).toEqual(before);
+      expect(await db.select().from(workspaceOperations)).toHaveLength(0);
+      expect(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(0);
+      expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n");
+      expect(await queuedBackends()).toHaveLength(0);
+    } finally { writer.release(); await writer.done; await importing; }
+    await expect(withIndependentWorkspaceSourceLocks(db, { companyId: f.companyId, issueIds: [f.issueId] }, async () => true, true)).resolves.toBe(true);
+  });
+  it("bounds the whole source queue and releases keys acquired before a later timeout or abort", async () => {
+    for (const mode of ["timeout", "abort"] as const) {
+      const companyId = randomUUID(), issueIds = [randomUUID(), randomUUID()].sort();
+      const first = await sourceWriter(companyId, [issueIds[0]]), second = await sourceWriter(companyId, [issueIds[1]]);
+      const abort = new AbortController(), reason = new Error("cancel partial admission"), startedAt = performance.now();
+      const waiting = withIndependentWorkspaceSourceLocks(db, { companyId, issueIds }, async () => "unexpected", true,
+        { waitMs: mode === "timeout" ? 600 : 5000, signal: abort.signal }).then(value => ({ value }), error => ({ error }));
+      try {
+        await expect.poll(async () => (await queuedBackends()).length).toBe(1);
+        if (mode === "timeout") await new Promise(resolve => setTimeout(resolve, 350));
+        first.release(); await first.done;
+        await expect.poll(async () => (await queuedBackends()).length).toBe(1);
+        await expect.poll(() => withIndependentWorkspaceSourceLocks(db, { companyId, issueIds: [issueIds[0]] }, async () => false, true)
+          .catch(() => true)).toBe(true);
+        if (mode === "abort") abort.abort(reason);
+        expect(await waiting).toMatchObject({ error: mode === "abort" ? reason : expect.objectContaining({ message: "workspace_source_busy" }) });
+        expect(performance.now() - startedAt).toBeLessThan(mode === "timeout" ? 850 : 2000);
+        await expect(withIndependentWorkspaceSourceLocks(db, { companyId, issueIds: [issueIds[0]] }, async () => true, true)).resolves.toBe(true);
+      } finally { first.release(); second.release(); await Promise.all([first.done, second.done, waiting]); }
+    }
+  });
   async function pendingRepository(f: Awaited<ReturnType<typeof seed>>) {
     const [repository] = await db.insert(executionWorkspaceRepositories).values({ companyId: f.companyId,
       executionWorkspaceId: f.workspaceId, requestedByIssueId: f.issueId, repositoryIdentity: "https://example.invalid/pending.git",

server/src/services/remote-workspace-migration.ts

@@ -325,7 +325,7 @@ export async function inventoryRemoteWorkspaceMigration(db: Db, input: { company
 
 type MigrationInput = {
   db: Db; storage: StorageService; companyId: string; workspaceId: string; apply?: boolean; source?: RemoteWorkspaceSource;
-  writersStopped?: boolean; signal?: AbortSignal;
+  writersStopped?: boolean; signal?: AbortSignal; sourceLockWaitMs?: number;
 };
 type MigrationTransaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
 
@@ -640,7 +640,8 @@ export async function migrateRemoteWorkspace(input: MigrationInput) {
     } finally { await childVerifier?.close(); if (scratch) await removeTaskDirectoryTree(scratch); }
   };
   if (onlineSource) {
-    const result = await withWorkspaceSourceLocks(input.db, {
+    const sourceLocks = input.sourceLockWaitMs === undefined ? withWorkspaceSourceLocks : withIndependentWorkspaceSourceLocks;
+    const result = await sourceLocks(input.db, {
       companyId: input.companyId, issueIds: onlineSource.issueIds,
       projectIds: onlineSource.projectIds,
     }, async lock => {
@@ -649,7 +650,7 @@ export async function migrateRemoteWorkspace(input: MigrationInput) {
           throw new IsolatedRepositorySourceBusy("workspace_source_busy");
       }
       return migrate(lock);
-    }, true);
+    }, true, input.sourceLockWaitMs === undefined ? undefined : { waitMs: input.sourceLockWaitMs, signal: input.signal });
     if (committedPublication) publishActivity(committedPublication);
     return result;
   }
@@ -664,7 +665,7 @@ const cohortSchema = z.object({
 
 export async function migrateRemoteWorkspaceCohort(input: {
   db: Db; storage: StorageService; companyId: string; workspaceIds: readonly string[];
-  apply?: boolean; concurrency?: number; signal: AbortSignal;
+  apply?: boolean; concurrency?: number; signal: AbortSignal; sourceLockWaitMs?: number;
 }) {
   const companyId = z.string().uuid().parse(input.companyId).toLowerCase();
   const workspaceIds = z.array(z.string().uuid()).min(1).max(8).parse(input.workspaceIds).map(id => id.toLowerCase()).sort();
@@ -803,7 +804,7 @@ export async function migrateRemoteWorkspaceCohort(input: {
       if (errors.length) throw new AggregateError(failure ? [failure, ...errors] : errors, "workspace_cohort_cleanup_failed");
       if (!failure) signal.throwIfAborted();
     }
-  }, true);
+  }, true, input.sourceLockWaitMs === undefined ? undefined : { waitMs: input.sourceLockWaitMs, signal });
   for (const publication of publications) publishActivity(publication);
   return result;
 }

server/src/services/workspace-source-lock.ts

@@ -11,6 +11,47 @@ type SourceTransaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
 type SourceLockScope = { db: Db; companyId: string; keys: Set<string>; transaction: SourceTransaction; active: boolean };
 const sourceLocks = new AsyncLocalStorage<SourceLockScope>();
 export class IsolatedRepositorySourceBusy extends Error {}
+export type WorkspaceSourceLockWait = { waitMs: number; signal?: AbortSignal };
+
+async function acquireQueuedSourceLocks(db: Db, transaction: SourceTransaction, keys: string[], wait: WorkspaceSourceLockWait) {
+  wait.signal?.throwIfAborted();
+  const deadline = performance.now() + wait.waitMs;
+  const [settings] = await transaction.execute(sql`select current_setting('lock_timeout') as lock_timeout,
+    pg_backend_pid() as pid, backend_start::text from pg_stat_activity where pid=pg_backend_pid()`);
+  let cancelling: Promise<unknown> | undefined;
+  const cancel = () => {
+    cancelling ??= withDedicatedDbConnection(db, async control => {
+      const timer = setTimeout(() => { void control.$client.end({ timeout: 0 }); }, 1000);
+      try {
+        await control.execute(sql`select pg_cancel_backend(pid) from pg_stat_activity
+          where pid=${Number(settings.pid)} and backend_start=${String(settings.backend_start)}::timestamptz
+            and datname=current_database() and usename=current_user`);
+      } finally { clearTimeout(timer); }
+    }).catch(() => undefined);
+  };
+  wait.signal?.addEventListener("abort", cancel, { once: true });
+  try {
+    for (const key of keys) {
+      wait.signal?.throwIfAborted();
+      const remaining = Math.ceil(deadline - performance.now());
+      if (remaining <= 0) throw new IsolatedRepositorySourceBusy("workspace_source_busy");
+      await transaction.execute(sql`select set_config('lock_timeout', ${`${remaining}ms`}, true)`);
+      await transaction.execute(sql`select pg_advisory_xact_lock(hashtextextended(${key}, 0))`);
+    }
+    wait.signal?.throwIfAborted();
+    if (performance.now() >= deadline) throw new IsolatedRepositorySourceBusy("workspace_source_busy");
+    await transaction.execute(sql`select set_config('lock_timeout', ${String(settings.lock_timeout)}, true)`);
+    wait.signal?.throwIfAborted();
+  } catch (error) {
+    wait.signal?.throwIfAborted();
+    if ((error as { cause?: { code?: string }; code?: string }).cause?.code === "55P03"
+      || (error as { code?: string }).code === "55P03") throw new IsolatedRepositorySourceBusy("workspace_source_busy");
+    throw error;
+  } finally {
+    wait.signal?.removeEventListener("abort", cancel);
+    await cancelling;
+  }
+}
 
 // Timers and detached promises keep the async context that created them, so
 // work can see a scope whose transaction has committed and whose dedicated
@@ -26,8 +67,8 @@ function heldScope(): SourceLockScope | undefined {
  * Joining that scope would tie these locks to the other transaction. */
 export function withIndependentWorkspaceSourceLocks<T>(db: Db,
   input: { companyId: string; issueIds?: readonly string[]; projectIds?: readonly string[] },
-  action: (transaction: SourceTransaction) => Promise<T>, tryOnly = false): Promise<T> {
-  return sourceLocks.exit(() => withWorkspaceSourceLocks(db, input, action, tryOnly));
+  action: (transaction: SourceTransaction) => Promise<T>, tryOnly = false, wait?: WorkspaceSourceLockWait): Promise<T> {
+  return sourceLocks.exit(() => withWorkspaceSourceLocks(db, input, action, tryOnly, wait));
 }
 
 export function withIsolatedRepositorySourceLock<T>(db: Db, companyId: string, issueId: string,
@@ -37,13 +78,16 @@ export function withIsolatedRepositorySourceLock<T>(db: Db, companyId: string, i
 
 export async function withWorkspaceSourceLocks<T>(db: Db,
   input: { companyId: string; issueIds?: readonly string[]; projectIds?: readonly string[] },
-  action: (transaction: SourceTransaction) => Promise<T>, tryOnly = false): Promise<T> {
+  action: (transaction: SourceTransaction) => Promise<T>, tryOnly = false, wait?: WorkspaceSourceLockWait): Promise<T> {
+  if (wait && (!Number.isSafeInteger(wait.waitMs) || wait.waitMs < 1 || wait.waitMs > 2_147_483_647))
+    throw new Error("workspace_source_wait_invalid");
   const keys = [...new Set([
     ...(input.issueIds ?? []).map(id => `isolated_repository_source:${input.companyId}:${id}`),
     ...(input.projectIds ?? []).map(id => `managed_project_source:${input.companyId}:${id}`),
   ])].sort();
   if (keys.length > 128) throw new Error("workspace_source_scope_limit");
   const held = heldScope();
+  if (wait && held?.db === db) throw new Error("workspace_source_wait_requires_independent_scope");
   const acquire = async (scope: SourceLockScope, nonblocking: boolean) => {
     for (const key of keys) {
       if (scope.keys.has(key)) continue;
@@ -65,7 +109,10 @@ export async function withWorkspaceSourceLocks<T>(db: Db,
   return withDedicatedDbConnection(db, connection => connection.transaction(async transaction => {
     const scope: SourceLockScope = { db, companyId: input.companyId, transaction, keys: new Set<string>(), active: true };
     try {
-      await acquire(scope, tryOnly);
+      if (wait) {
+        await acquireQueuedSourceLocks(db, transaction, keys, wait);
+        for (const key of keys) scope.keys.add(key);
+      } else await acquire(scope, tryOnly);
       const rows = await transaction.execute(sql`select setting::bigint as idle_timeout_ms from pg_settings where name = 'idle_in_transaction_session_timeout'`);
       return await withWorkspaceLockKeepalive(transaction, Number(rows[0]?.idle_timeout_ms),
         () => sourceLocks.run(scope, () => action(transaction)));