fix(workspaces): protect migration sources and add guarded recovery service (#20443)

## Thinking Path

> - Paperclip manages agents and their task workspaces.
> - Online migration saves files before it publishes remote workspace
authority.
> - A saved migration journal can survive the importing process.
> - Git cache expiration did not count that journal as a consumer.
> - Expiration could remove a source and change its fingerprint while
its import was incomplete.
> - This PR protects unfinished imports and exposes a guarded internal
service transition for an already affected cohort.
> - New imports can preserve the current sources while all old recovery
data stays retained.

## Linked Issues or Issue Description

**What happened?**

Git cache expiration could archive and remove a workspace after an
import process ended but before its saved cohort finished. The saved
source fingerprint then differed from the current archived workspace.
Exact cohort resume correctly refused that changed source. Reopening the
workspace cannot recreate the old metadata or fingerprint.

**Expected behavior**

An unfinished migration remains a durable consumer after process loss.
If an older server already expired some sources, one canonical
transition can fail the exact unpublished cohort without losing its
snapshots, objects, source history, or recovery obligations. New
attempts use the current verified sources.

**Steps to reproduce**

1. Start a cohort import of eight Git worktrees with local changes.
2. Interrupt it after four snapshot uploads.
3. Let an older server expire two idle source worktrees.
4. Resume the saved cohort. It refuses the changed archived sources.

Related: #20429 preserves exact cohort identity when independent
companion imports change only the scope proof. #20424 preserves prepared
recovery state for archived sources. This PR handles completed source
expiration, not a scope-only change. Public searches found related
expiration work and no matching compensation owner.

## What Changed

- Count saved migration markers and unfinished migration journals as Git
expiration consumers. Keep failed migration retention holds as consumers
too.
- Expose `supersedeExpiredRemoteWorkspaceMigrationCohort` as an internal
service API for a privileged operator with a real database handle and
reviewed company/cohort inputs. It is called directly by the
PostgreSQL/Git test and reviewed one-time administrative qualification.
It adds no CLI mode or HTTP route. The existing CLI still refuses a
damaged cohort; this PR does not claim automatic CLI recovery.
- Require an exact unpublished cohort of up to eight members. It
requires completed expiration and an unexpired restore recipe for each
missing member. The other members must retain their saved source
fingerprints.
- Recheck current source and recovery state under source, lifecycle,
finalizer, and row locks. One transaction fails all old journals, writes
audit records, and removes only their exact matched active markers.
- Keep old cohort proofs, operations, snapshots, and object references.
Genuine failure receipts use the existing failed-save retention hold.
This transition does not report migration success.
- Reuse the existing bounded source-lock queue when `sourceLockWaitMs`
is set. Keep fast refusal by default. Cancellation targets only this
waiter, and current checks still run after admission.
- Record predecessor IDs on new attempts and reserve a snapshot
generation above the existing ledger maximum. New success does not
retire the old hold.
- Document the owning transition and retention rules.

## Verification

- The two new journal-consumer cases failed under the old expiration
predicate and passed under the new predicate.
- The full new recovery case passed in 21.12 seconds with real
PostgreSQL locks, Git repositories, encrypted local object storage, and
archive restores. It covers four retained uploads, two real expirations,
unchanged old resume refusal, authority change refusal, transaction
rollback, a held native finalizer, idempotence, eight new imports,
dirty/untracked file restores, and a retention pass 30 days later with
no object deletion.
- Server TypeScript and module-boundary checks passed.
- On `7cec8c915acfad2d08874eaf26abaa852f560219`, the queued-admission
regression failed before forwarding the option (24.07 seconds), then
passed (22.25 seconds). It verifies a real held project lock,
cancellation that preserves the holder and all journals, release into
full compensation, retention, and eight restored fresh imports. Current
server TypeScript and module/diff checks passed. The only production
follow-up changes are the optional queue input and its existing-owner
call.
- The earlier `737a0df3a71b10fe48d7bbfe7ffd8ae296b4852c` CI run had one
Runner support failure: the unchanged credential-free Python/Pi
calibration exceeded its outer 25-second subprocess deadline. That job
passed 6,926 other tests and skipped four. The timeout is retained as a
nonpass. The changed logical head requires fresh CI and review; no test
deadline was raised.
- The affected cohort, Git expiration, and eligibility suites passed all
148 cases in 228.41 seconds on
`aa589f3f447e36d0277c32f8b44cbe5e84a673be`. Merged current master
`a9d0946e64`; the only conflict retained both independent added test
blocks. The complete merged cohort suite passed all 30 cases in 70.00
seconds on `737a0df3a71b10fe48d7bbfe7ffd8ae296b4852c`, including
upstream larger-cohort controls. Final-head server TypeScript,
incremental preflight, module-boundary, and diff checks passed.
Unchanged expiration/eligibility evidence remains valid. Fresh
`809ab704e619812d8f1bf38cf5de9c65893861cc` CI and review are pending.
- On `809ab704e619812d8f1bf38cf5de9c65893861cc`, the complete recovery
fixture passed in 21.38 seconds after adding one valid archived recipe
with no restore point. It refuses with
`migration_supersession_expiration_unverified` and keeps all workspaces,
journals, and snapshots unchanged. The same case then verifies queue
cancellation/release, full compensation, retained history, fresh
imports, and restored bytes. Server TypeScript passed. This follow-up
changes only the test and service-scope documentation; product bytes
remain identical to qualified
`7cec8c915acfad2d08874eaf26abaa852f560219`.
- Earlier fixture attempts did not pass. They exposed the selection of a
held source and an incorrect expected refusal string. The fixture now
selects eligible sources and asserts the canonical authority refusal.
Production guards were not relaxed.
- Live qualification of the exact owning product bytes completed in
230.419 seconds. All eight old operations became failed with auditable
supersession lineage. All eight old snapshot rows remained byte-for-byte
identical, and their object references remain retained. The existing
source-lock queue waited 1.151 seconds for a normal finalizer handoff.
The three complete source/recovery passes took about 218 seconds
including their transaction, which explains why the earlier 180-second
total budget was insufficient for that measured workload. No source
proof or publication guard was skipped. Compensation made no provider or
object-storage calls.
- Earlier live qualifications remain nonpasses: the 180-second attempt
did not finish; its instrumented successor completed source checks and
refused a busy source lock before mutation. The successful bounded-queue
attempt retained those failure records. Fresh archived/online imports
are a separate subsequent step and are not claimed complete here.
- The local checks do not claim browser acceptance, app deployment,
fleet enrollment, or a completed zero-downtime cutover.

## Risks

- This owner mutates migration bookkeeping. It preserves all-or-nothing
row comparisons and current source/recovery locks. A changed journal,
authority, writer, finalizer, expired recipe, or partial transaction
refuses the transition.
- Historical snapshots and objects remain pinned after new success. This
intentionally retains storage. Release requires a separate reviewed
action.
- An older importer cannot use the successor generation contract. Use
the containing canonical owner for recovery. Normal app delivery must
also carry the expiration guard; a private maintenance graph does not
patch a running server's cleanup worker.
- The migration owner crosses 1,000 lines. The bounded compensation
transition remains beside its private source inspection, fingerprint,
fence, reservation, and publication contracts. Moving it now would
require exporting or duplicating those safety owners. No unrelated
refactor or new framework is included.

## Model Used

OpenAI GPT-6 through Codex, with reasoning, source editing, Git, and
local execution tools. The exact serving model ID 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/DATABASE.md

@@ -161,6 +161,33 @@ cohort proof for all members. Cancellation or a changed journal rolls back the
 whole refresh. Partial reservations still require the original aggregate proof.
 Archives, recovery obligations and publication checks retain their existing owners.
 
+Git cache expiration treats a saved migration marker or unfinished migration
+journal as a durable consumer. It checks this under the lifecycle lock. A failed
+migration with `lastFailure` and no `retiredSave` also keeps its source and
+snapshot retention hold after process loss.
+
+If an older server already expired members of an unpublished cohort,
+`supersedeExpiredRemoteWorkspaceMigrationCohort` accepts the exact saved cohort
+and the exact expired member IDs. This is an internal service API for a privileged
+operator with a real database handle and reviewed company/cohort inputs. It does
+not add a CLI mode or an HTTP route. An operator calls the exported owner directly;
+its test and the reviewed one-time administrative qualification use that path.
+It requires completed Git expiration and an
+unexpired restore-point recipe for each missing member. Other members must retain
+their saved source fingerprints. Fresh source, lifecycle, finalizer and row locks
+protect one transaction that fails every old journal and removes only its matched
+active marker. It does not report migration success or change task state. The
+old cohort proof, operations, snapshots and object references remain unchanged.
+The failure receipt retains those objects through the existing failed-save hold.
+The optional `sourceLockWaitMs` uses the existing bounded source-lock queue.
+The default still refuses a busy source immediately. Cancellation targets this
+waiter only. All source and recovery checks run again after lock admission.
+
+New canonical imports use the current archived recipe or current online source.
+They record the old operation and supersession IDs and reserve a snapshot
+generation above the existing ledger maximum. A successful new import does not
+retire the old hold. Release of those historical objects requires separate review.
+
 ### Original local Git sources for archived workspaces
 
 When an archived runtime branch has an exact successful deletion receipt, its

server/src/__tests__/git-workspace-expiration-eligibility.test.ts

@@ -1,7 +1,7 @@
 import { randomUUID } from "node:crypto";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import { eq, sql } from "drizzle-orm";
-import { agents, companies, createDb, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, projects, projectWorkspaces } from "@paperclipai/db";
+import { agents, companies, createDb, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, projects, projectWorkspaces, workspaceOperations } from "@paperclipai/db";
 import { isGitWorkspaceExpirationEligible } from "../services/git-workspace-expiration.js";
 import { createSystemRecovery } from "../services/system-recovery-state.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
@@ -34,6 +34,19 @@ describeDatabase("Git expiration eligibility", () => {
     return { companyId, agentId, issueId, workspace, now, old, eligible };
   }
 
+  it.each(["running", "failed-hold"])("keeps a saved migration consumer after its process is gone (%s)", async state => {
+    const f = await fixture();
+    expect(await f.eligible()).toBe(true);
+    const [operation] = await db.insert(workspaceOperations).values({ companyId: f.companyId,
+      executionWorkspaceId: f.workspace.id, issueId: f.issueId, phase: "remote_workspace_migration",
+      status: state === "running" ? "running" : "failed",
+      metadata: state === "running" ? { schema: "paperclip.remote-workspace-migration.v1" }
+        : { lastFailure: { code: "source_expired_during_import" } } }).returning();
+    expect(await f.eligible()).toBe(false);
+    await db.delete(workspaceOperations).where(eq(workspaceOperations.id, operation.id));
+    expect(await f.eligible()).toBe(true);
+  });
+
   async function alias(f: Awaited<ReturnType<typeof fixture>>, kind: string) {
     const companyId = randomUUID();
     await db.insert(companies).values({ id: companyId, name: "Other owner", issuePrefix: `EO${companyId.slice(0, 6)}` });

server/src/services/git-workspace-expiration.ts

@@ -5,7 +5,7 @@ import { randomBytes } from "node:crypto";
 import { acquireGitExpirationLocks } from "./git-workspace-expiration-locks.js";
 import { and, eq, inArray, ne, sql } from "drizzle-orm";
 import { executionWorkspaces, issues, heartbeatRuns, nativeRunFinalizations, environmentLeases,
-  workspaceRuntimeServices, issueRecoveryActions, projectWorkspaces, withDedicatedDbConnection, type Db } from "@paperclipai/db";
+  workspaceRuntimeServices, issueRecoveryActions, projectWorkspaces, workspaceOperations, withDedicatedDbConnection, type Db } from "@paperclipai/db";
 import { createIssueGitRemoteAuthProvider } from "./git-credentials.js";
 import type { StorageService } from "../storage/types.js";
 import { TASK_DIRECTORY_RETENTION_KEY } from "./isolated-task-directory-snapshot.js";
@@ -166,7 +166,7 @@ export async function gitWorkspaceLastActivityAt(db: Db | Tx, workspace: Workspa
 
 export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Workspace, now: Date, idleCutoff?: Date,
   options: { closedSiblingsAllowed?: boolean } = {}) {
-  if (workspace.metadata?.reopenPendingConsumption === true) return false;
+  if (workspace.metadata?.reopenPendingConsumption === true || workspace.metadata?.remoteWorkspaceMigration != null) return false;
   const siblingBlocks = options.closedSiblingsAllowed
     ? sql`AND (other.status NOT IN ('archived', 'cleanup_failed') OR other.closed_at IS NULL
         OR other.metadata->${TASK_DIRECTORY_RETENTION_KEY}->>'removed' = 'false')`
@@ -208,6 +208,10 @@ export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Worksp
         AND (l.execution_workspace_id = ${workspace.id} OR l.heartbeat_run_id IN (SELECT id FROM relevant_runs))
         AND (l.status = 'pending_cleanup' OR l.cleanup_status = 'failed' OR l.metadata ? 'nativeWorkspaceExportResume'
           OR (l.released_at IS NULL AND NOT coalesce(l.status = 'retained' AND l.lease_policy = 'reuse_by_environment' AND l.cleanup_status = 'success', false))))
+      AND NOT EXISTS (SELECT 1 FROM ${workspaceOperations} o WHERE o.company_id = ${workspace.companyId}
+        AND o.execution_workspace_id = ${workspace.id} AND o.phase = 'remote_workspace_migration'
+        AND (o.status IN ('running', 'pending') OR o.metadata->>'uncertain' = 'true'
+          OR (o.status = 'failed' AND o.metadata ? 'lastFailure' AND o.metadata->'retiredSave' IS NULL)))
       AND NOT EXISTS (SELECT 1 FROM ${workspaceRuntimeServices} WHERE company_id = ${workspace.companyId}
         AND (execution_workspace_id = ${workspace.id} OR cwd = ${workspace.cwd}
           OR left(cwd, length(${workspace.cwd}) + 1) = ${workspace.cwd} || '/') AND status <> 'stopped')

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

@@ -4,16 +4,17 @@ import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
 import { promisify } from "node:util";
+import { pathToFileURL } from "node:url";
 import { eq, inArray, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
-import { activityLog, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, executionWorkspaceSnapshots, executionWorkspaces, issues, projects, projectWorkspaces, workspaceOperations, type Db } from "@paperclipai/db";
+import { activityLog, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, executionWorkspaceSnapshots, executionWorkspaceRestorePoints, executionWorkspaces, issues, projects, projectWorkspaces, workspaceOperations, type Db } from "@paperclipai/db";
 import { readRemoteWorkspaceAuthority } from "@paperclipai/shared";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
 import { resolveManagedProjectWorkspaceDir } from "../home-paths.js";
 import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js";
 import { createStorageService } from "../storage/service.js";
 import type { StorageService } from "../storage/types.js";
-import { migrateRemoteWorkspace, migrateRemoteWorkspaceCohort } from "./remote-workspace-migration.js";
+import { migrateRemoteWorkspace, migrateRemoteWorkspaceCohort, supersedeExpiredRemoteWorkspaceMigrationCohort } from "./remote-workspace-migration.js";
 import { inspectOnlineWorkspaceSource } from "./online-workspace-source.js";
 import { IsolatedRepositorySourceBusy, withWorkspaceSourceLocks } from "./workspace-source-lock.js";
 import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "./remote-workspace-snapshot-storage.js";
@@ -22,6 +23,8 @@ import { subscribeCompanyLiveEvents } from "./live-events.js";
 import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js";
 import * as workspaceArchive from "./remote-workspace-snapshot-archive.js";
 import * as onlineWorkspaceSource from "./online-workspace-source.js";
+import { expireReconstructableGitWorkspace } from "./git-workspace-expiration.js";
+import { remoteWorkspaceSnapshotRetentionService } from "./remote-workspace-snapshot-retention.js";
 
 const support = await getEmbeddedPostgresTestSupport(), execute = promisify(execFile);
 const git = async (cwd: string, ...args: string[]) => (await execute("git", ["-c", `core.hooksPath=${os.devNull}`, "-C", cwd, ...args], {
@@ -40,7 +43,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () =>
   });
   afterEach(async () => {
     vi.restoreAllMocks();
-    await db.delete(activityLog); await db.delete(workspaceOperations); await db.delete(executionWorkspaceSnapshots);
+    await db.delete(activityLog); await db.delete(workspaceOperations); await db.delete(executionWorkspaceSnapshots); await db.delete(executionWorkspaceRestorePoints);
     await db.execute(sql`drop trigger if exists cohort_publish_probe on execution_workspaces`);
     await db.execute(sql`drop function if exists cohort_publish_probe()`); await db.execute(sql`drop sequence if exists cohort_publish_attempts`);
     await db.execute(sql`drop trigger if exists cohort_commit_fault on workspace_operations`); await db.execute(sql`drop function if exists cohort_commit_fault()`);
@@ -49,7 +52,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () =>
     await db.delete(issues); await db.delete(agents); await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); await db.delete(projects); await db.delete(companies);
     vi.unstubAllEnvs(); await fs.rm(home, { recursive: true, force: true });
   });
-  async function fixture(repoName?: string) {
+  async function fixture(repoName?: string, leafCount = 2) {
     const companyId = randomUUID(), projectId = randomUUID(), root = resolveManagedProjectWorkspaceDir({ companyId, projectId, repoName });
     await db.insert(companies).values({ id: companyId, name: "Cohort", issuePrefix: `C${companyId.slice(0, 6)}` });
     await db.insert(projects).values({ id: projectId, companyId, name: "Shared Git" });
@@ -57,7 +60,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () =>
     await git(root, "init", "-qb", "main"); await git(root, "add", "."); await git(root, "commit", "-qm", "Original history");
     const repoUrl = "https://example.invalid/cohort.git"; await git(root, "remote", "add", "origin", repoUrl);
     const members = [];
-    for (const name of ["base", "one", "two"]) {
+    for (const name of ["base", "one", "two", ...Array.from({ length: leafCount - 2 }, (_, index) => `leaf-${index}`)]) {
       const cwd = name === "base" ? root : path.join(root, ".paperclip", "worktrees", name);
       if (name !== "base") await git(root, "worktree", "add", "-qb", name, cwd);
       const [issue] = await db.insert(issues).values({ companyId, projectId, title: name }).returning();
@@ -223,6 +226,139 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () =>
     await assertLocal(f);
     expect(await db.select().from(workspaceOperations)).toHaveLength(mode === "foreign" ? 3 : 2);
   });
+  it("compensates an exact unpublished cohort after Git expiration without losing its data or history", async () => {
+    const f = await fixture("cohort", 8), bare = path.join(home, "upstream", "cohort.git"), old = new Date(Date.now() - 4 * 86_400_000);
+    await fs.mkdir(path.dirname(bare), { recursive: true }); await git(f.root, "clone", "--bare", f.root, bare);
+    const repoUrl = pathToFileURL(bare).href; await git(f.root, "remote", "set-url", "origin", repoUrl);
+    await db.update(executionWorkspaces).set({ repoUrl, metadata: { createdByRuntime: true, gitBranchOwnershipVersion: 1 } });
+    await db.update(issues).set({ status: "done", completedAt: old, updatedAt: old });
+    const held = await recoverable({ ...f, leaves: [f.leaves[3]] }), put = storage.putFile.bind(storage); let uploads = 0;
+    const interrupted = vi.spyOn(storage, "putFile").mockImplementation(async input => {
+      if (++uploads === 5) throw new Error("cohort_after_four_uploads"); return put(input);
+    });
+    await expect(run(f, { concurrency: 1 })).rejects.toThrow("cohort_after_four_uploads"); interrupted.mockRestore();
+    const initialOperations = (await db.select().from(workspaceOperations)).filter(row => row.phase === "remote_workspace_migration");
+    expect(initialOperations).toHaveLength(8);
+    const cohort = initialOperations[0].metadata!.cohortMigration as { id: string; members: Array<{ workspaceId: string; operationId: string }> };
+    const expirable = cohort.members.filter(member => member.workspaceId !== f.leaves[3].id);
+    const expiredWorkspaceIds = [expirable[0].workspaceId, expirable.at(-1)!.workspaceId], now = new Date(Date.now() + 3 * 86_400_000);
+    const expireOptions = { now, expireBase: false, dryRun: false, signal: AbortSignal.timeout(60_000),
+      restorePoint: { storage, notBefore: old, keyDurable: true } };
+    const [protectedRow] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, expiredWorkspaceIds[0]));
+    const restarted = createDb(database.connectionString, { maxConnections: 1 });
+    try { expect(await expireReconstructableGitWorkspace(restarted, protectedRow, expireOptions))
+      .toMatchObject({ expired: false, skipReason: "live_or_unresolved_consumer" }); }
+    finally { await restarted.$client.end({ timeout: 1 }); }
+    for (const workspaceId of expiredWorkspaceIds) {
+      const [before] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId));
+      const marker = before.metadata!.remoteWorkspaceMigration, metadata = { ...before.metadata }; delete metadata.remoteWorkspaceMigration;
+      const operation = initialOperations.find(row => row.executionWorkspaceId === workspaceId)!;
+      await db.update(workspaceOperations).set({ status: "failed" }).where(eq(workspaceOperations.id, operation.id));
+      const [unprotected] = await db.update(executionWorkspaces).set({ metadata }).where(eq(executionWorkspaces.id, workspaceId)).returning();
+      const expired = await expireReconstructableGitWorkspace(db, unprotected, expireOptions);
+      expect(expired.skipReason).toBe(null); expect(expired).toMatchObject({ expired: true, restorePoint: true });
+      const [archived] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId));
+      await db.update(executionWorkspaces).set({ metadata: { ...archived.metadata, remoteWorkspaceMigration: marker } }).where(eq(executionWorkspaces.id, workspaceId));
+      await db.update(workspaceOperations).set({ status: "running" }).where(eq(workspaceOperations.id, operation.id));
+      await expect(fs.access(before.cwd!)).rejects.toMatchObject({ code: "ENOENT" });
+    }
+    await expect(run(f)).rejects.toThrow("workspace_cohort_online_source_required");
+    const oldOperations = await db.select().from(workspaceOperations), oldSnapshots = await db.select().from(executionWorkspaceSnapshots);
+    expect(oldSnapshots.filter(row => row.status === "uploaded")).toHaveLength(4);
+    const tasks = await db.select().from(issues), runs = await db.select().from(heartbeatRuns), leases = await db.select().from(environmentLeases);
+    const finalizers = await db.select().from(nativeRunFinalizations);
+    const compensate = (options: Partial<Parameters<typeof supersedeExpiredRemoteWorkspaceMigrationCohort>[0]> = {}) =>
+      supersedeExpiredRemoteWorkspaceMigrationCohort({ db, companyId: f.companyId, cohort,
+        expiredWorkspaceIds, signal: AbortSignal.timeout(60_000), ...options });
+    const [missingProof] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, expiredWorkspaceIds[0]));
+    const withoutRestorePoint = { ...(missingProof.metadata!.gitCacheRestoration as Record<string, unknown>), schema: "paperclip.git-cache-restoration.v1" };
+    delete (withoutRestorePoint as Record<string, unknown>).restorePoint;
+    await db.update(executionWorkspaces).set({ metadata: { ...missingProof.metadata, gitCacheRestoration: withoutRestorePoint } })
+      .where(eq(executionWorkspaces.id, missingProof.id));
+    const unverifiedRows = await rows(f);
+    await expect(compensate()).rejects.toThrow("migration_supersession_expiration_unverified");
+    expect(await rows(f)).toEqual(unverifiedRows); expect(await db.select().from(workspaceOperations)).toEqual(oldOperations);
+    expect(await db.select().from(executionWorkspaceSnapshots)).toEqual(oldSnapshots);
+    await db.update(executionWorkspaces).set({ metadata: missingProof.metadata }).where(eq(executionWorkspaces.id, missingProof.id));
+    const [authorityRow] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, cohort.members[2].workspaceId));
+    await db.update(executionWorkspaces).set({ metadata: { ...authorityRow.metadata, remoteWorkspace: {
+      schema: "paperclip.remote-workspace.v1", epoch: 1, source: initialOperations[0].metadata!.source,
+      snapshotId: oldSnapshots.find(row => row.status === "uploaded")!.id, writerRunId: null, location: null, dirty: false } } })
+      .where(eq(executionWorkspaces.id, authorityRow.id));
+    await expect(compensate()).rejects.toThrow("online_workspace_source_scope_unproven");
+    await db.update(executionWorkspaces).set({ metadata: authorityRow.metadata }).where(eq(executionWorkspaces.id, authorityRow.id));
+    const beforeRows = await rows(f), last = cohort.members.at(-1)!;
+    await db.execute(sql.raw(`create function cohort_commit_fault() returns trigger language plpgsql as $$ begin if new.id='${last.operationId}'::uuid and new.status='failed' then raise exception 'cohort_compensation_fault'; end if; return new; end $$`));
+    await db.execute(sql`create trigger cohort_commit_fault before update on workspace_operations for each row execute function cohort_commit_fault()`);
+    await expect(compensate()).rejects.toThrow();
+    expect(await db.select().from(workspaceOperations)).toEqual(oldOperations); expect(await rows(f)).toEqual(beforeRows);
+    await db.execute(sql`drop trigger cohort_commit_fault on workspace_operations`); await db.execute(sql`drop function cohort_commit_fault()`);
+    const other = createDb(database.connectionString, { maxConnections: 1 });
+    let receipt!: Awaited<ReturnType<typeof compensate>>;
+    const queued = () => db.execute(sql`select exists(select 1 from pg_locks where locktype='advisory' and not granted
+      and database=(select oid from pg_database where datname=current_database())) as waiting`);
+    try {
+      await other.transaction(async tx => {
+        expect(await tryLockNativeWorkspaceFinalization(tx as unknown as Db, { companyId: f.companyId, runId: held.runId })).toBe(true);
+        await expect(compensate()).rejects.toThrow("remote_copyback_owned");
+      });
+      await withWorkspaceSourceLocks(other, { companyId: f.companyId, projectIds: [f.projectId] }, async holder => {
+        const abort = new AbortController(), pending = compensate({ sourceLockWaitMs: 10_000, signal: abort.signal });
+        const outcome = pending.then(value => ({ value, error: null }), error => ({ value: null, error }));
+        try {
+          await expect.poll(async () => (await queued())[0].waiting, { timeout: 8_000 }).toBe(true);
+          abort.abort(new Error("compensation_queue_cancelled"));
+          expect((await outcome).error?.message).toBe("compensation_queue_cancelled");
+          expect((await holder.execute(sql`select exists(select 1 from pg_locks where pid=pg_backend_pid()
+            and locktype='advisory' and granted) as held`))[0].held).toBe(true);
+          expect(await db.select().from(workspaceOperations)).toEqual(oldOperations);
+        } finally { abort.abort(); await outcome; }
+      });
+      let pending: ReturnType<typeof compensate> | undefined;
+      try {
+        await withWorkspaceSourceLocks(other, { companyId: f.companyId, projectIds: [f.projectId] }, async () => {
+          pending = compensate({ sourceLockWaitMs: 10_000 });
+          void pending.catch(() => undefined);
+          await expect.poll(async () => (await queued())[0].waiting, { timeout: 8_000 }).toBe(true);
+        });
+      } finally { if (pending) receipt = await pending; }
+    } finally { await other.$client.end({ timeout: 1 }); }
+    expect(receipt).toMatchObject({ superseded: true, migrationCompleted: false, replayed: false });
+    await expect(compensate()).resolves.toMatchObject({ ...receipt, replayed: true });
+    const failed = await db.select().from(workspaceOperations);
+    for (const operation of failed) expect(operation).toMatchObject({ status: "failed", metadata: {
+      ...oldOperations.find(row => row.id === operation.id)!.metadata, lastFailure: { schema: "paperclip.remote-workspace-migration-supersession.v1",
+        code: "source_expired_during_import", cohortId: cohort.id, supersessionId: receipt.supersessionId } } });
+    expect(await db.select().from(executionWorkspaceSnapshots)).toEqual(oldSnapshots);
+    expect(await db.select().from(issues)).toEqual(tasks); expect(await db.select().from(heartbeatRuns)).toEqual(runs);
+    expect(await db.select().from(environmentLeases)).toEqual(leases); expect(await db.select().from(nativeRunFinalizations)).toEqual(finalizers);
+    for (const workspaceId of expiredWorkspaceIds) await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId,
+      apply: true, signal: AbortSignal.timeout(60_000) });
+    await expect(run(f, { workspaceIds: cohort.members.map(member => member.workspaceId).filter(id => !expiredWorkspaceIds.includes(id)) }))
+      .resolves.toMatchObject({ applied: true });
+    const currentOperations = await db.select().from(workspaceOperations);
+    for (const member of cohort.members) {
+      const successor = currentOperations.find(row => row.executionWorkspaceId === member.workspaceId && row.status === "succeeded")!;
+      expect(successor.id).not.toBe(member.operationId);
+      expect(successor.metadata?.migrationSupersession).toMatchObject({ operationId: member.operationId, cohortId: cohort.id,
+        supersessionId: receipt.supersessionId, generation: oldSnapshots.some(row => row.executionWorkspaceId === member.workspaceId) ? 2 : 1 });
+    }
+    const deleted = vi.spyOn(storage, "deleteObject");
+    const retention = await remoteWorkspaceSnapshotRetentionService(db, { storage, now: () => new Date(Date.now() + 30 * 86_400_000) }).tick();
+    expect(retention.pinned).toBeGreaterThanOrEqual(oldSnapshots.length); expect(deleted).not.toHaveBeenCalled();
+    expect((await db.select().from(workspaceOperations)).filter(row => oldOperations.some(old => old.id === row.id))).toEqual(failed);
+    await fs.rename(f.root, f.root + "-original");
+    for (const workspace of f.leaves) {
+      const [saved] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace.id));
+      const ref = (await getWorkspaceSnapshot(db, { companyId: f.companyId, workspaceId: saved.id,
+        snapshotId: readRemoteWorkspaceAuthority(saved.metadata)!.snapshotId! }))!;
+      const archivePath = path.join(home, `${saved.id}.pcws`), targetDir = path.join(home, `${saved.id}-restored`);
+      await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: archivePath });
+      await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: ref.manifestDigest });
+      expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe(`${workspace.name} dirty tracked work\n`);
+      expect(await fs.readFile(path.join(targetDir, "untracked.txt"), "utf8")).toBe(`${workspace.name} untracked work\n`);
+    }
+  }, 120_000);
   it.each(["fresh", "resume"] as const)("imports a larger same-root cohort with bounded work (%s)", async mode => {
     const f = await fixture(), selected = await aliasMembers(f, f.parent, 10), workspaceIds = selected.map(row => row.id);
     for (const child of f.leaves) await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: child.id,

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

@@ -29,8 +29,13 @@ import { inspectOnlineWorkspaceSource, requiresOnlineWorkspaceSource, type Onlin
 import { withWorkspaceSourceLocks, withIndependentWorkspaceSourceLocks, IsolatedRepositorySourceBusy } from "./workspace-source-lock.js";
 import { inspectOnlineLegacyWorkspaceRecoverySource, inspectOnlineLegacyWorkspaceRecoveryHold, lockOnlineLegacyWorkspaceRecoverySources, ONLINE_LEGACY_WORKSPACE_PRESERVATION, type OnlineLegacyWorkspaceRecoveryHold } from "./online-legacy-workspace-recovery.js";
 import { inspectNativeWorkspaceMigration, nativeWorkspaceMigrationMarker } from "./native-runtime/native-workspace-migration.js";
+import { readGitCacheExpiration, GIT_CACHE_EXPIRATION_REASON } from "./git-workspace-restoration.js";
 
 const SCHEMA = "paperclip.remote-workspace-migration.v1";
+const SUPERSESSION = "paperclip.remote-workspace-migration-supersession.v1";
+const supersessionSchema = z.object({ schema: z.literal(SUPERSESSION), companyId: z.string().uuid(), workspaceId: z.string().uuid(),
+  cohortId: z.string().uuid(), operationId: z.string().uuid(), supersessionId: z.string().uuid(), snapshotIds: z.array(z.string().uuid()).max(8) }).strict();
+const successorSchema = supersessionSchema.extend({ generation: z.number().int().min(1).max(2_147_483_647) });
 type Workspace = typeof executionWorkspaces.$inferSelect;
 type ReadDb = Pick<Db, "select">;
 type OnlineRecoveryHold = OnlineLegacyWorkspaceRecoveryHold & { actionId: string };
@@ -347,7 +352,7 @@ type MigrationInput = {
 };
 type MigrationTransaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
 
-async function resolveMigrationSource(input: MigrationInput) {
+async function resolveMigrationSource(input: Pick<MigrationInput, "db" | "companyId" | "workspaceId" | "writersStopped">) {
   const [sourceWorkspace] = !input.writersStopped ? await input.db.select().from(executionWorkspaces).where(and(
     eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))) : [];
   const onlineSource = sourceWorkspace && !readRemoteWorkspaceAuthority(sourceWorkspace.metadata)
@@ -408,10 +413,19 @@ async function reserveMigration(input: MigrationInput, onlineSource?: OnlineWork
       if (canonical(operation.metadata?.onlineRecoveryHolds ?? []) !== canonical(current.onlineRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed");
       return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds, archivedRepositoryIntents: current.archivedRepositoryIntents };
     }
+    let migrationSupersession;
+    if (workspace.metadata?.remoteWorkspaceMigrationSupersession != null) {
+      const predecessor = supersessionSchema.parse(workspace.metadata.remoteWorkspaceMigrationSupersession);
+      if (predecessor.companyId !== workspace.companyId || predecessor.workspaceId !== workspace.id) throw new Error("migration_supersession_lineage_changed");
+      const [latest] = await tx.select({ generation: sql<number>`coalesce(max(${executionWorkspaceSnapshots.generation}),0)` }).from(executionWorkspaceSnapshots)
+        .where(and(eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id)));
+      migrationSupersession = successorSchema.parse({ ...predecessor, generation: Number(latest.generation) + 1 });
+    }
     const [operation] = await tx.insert(workspaceOperations).values({ ...(cohort ? { id: cohort.members.find(member => member.workspaceId === input.workspaceId)!.operationId } : {}), companyId: workspace.companyId, executionWorkspaceId: workspace.id,
       issueId: workspace.sourceIssueId, cwd: workspace.cwd ?? workspace.providerRef, phase: "remote_workspace_migration", status: "running",
       metadata: { schema: SCHEMA, sourceFingerprint, source, sourceIssueBinding: current.issueBinding,
         ...(cohort ? { cohortMigration: cohort } : {}),
+        ...(migrationSupersession ? { migrationSupersession } : {}),
         ...(nativeSessionContinuity ? { nativeSessionContinuity } : {}),
         nestedWorkspaceScope: { version: 1, state: "pending" },
         ...(current.finalizedRecoveryHolds.length ? { finalizedRecoveryHolds: current.finalizedRecoveryHolds } : {}),
@@ -488,8 +502,11 @@ async function captureMigration(input: MigrationInput, reserved: ReservedMigrati
     if (!refreshed) throw new Error("remote_workspace_migration_operation_changed");
     operation = refreshed;
   }
+  const predecessor = operation.metadata?.migrationSupersession == null ? null : successorSchema.parse(operation.metadata.migrationSupersession);
+  if (predecessor && (predecessor.companyId !== input.companyId || predecessor.workspaceId !== input.workspaceId))
+    throw new Error("migration_supersession_lineage_changed");
   const snapshot = await beginWorkspaceSnapshot(input.db, { companyId: input.companyId, workspaceId: input.workspaceId,
-    operationId: operation.id, generation: 1, epoch: 1 });
+    operationId: operation.id, generation: predecessor?.generation ?? 1, epoch: 1 });
   const scratch = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-migration-")));
   try {
     input.signal?.throwIfAborted();
@@ -932,3 +949,118 @@ export async function migrateRemoteWorkspaceCohort(input: {
   for (const publication of publications) publishActivity(publication);
   return result;
 }
+
+
+export async function supersedeExpiredRemoteWorkspaceMigrationCohort(input: {
+  db: Db; companyId: string; cohort: unknown; expiredWorkspaceIds: readonly string[]; signal: AbortSignal; sourceLockWaitMs?: number;
+}) {
+  const companyId = z.string().uuid().parse(input.companyId).toLowerCase(), cohort = cohortSchema.parse(input.cohort);
+  const expired = z.array(z.string().uuid()).min(1).max(8).parse(input.expiredWorkspaceIds).map(id => id.toLowerCase()).sort();
+  const ids = cohort.members.map(member => member.workspaceId), operationIds = cohort.members.map(member => member.operationId);
+  if (cohort.companyId !== companyId || ids.length > 8 || new Set(ids).size !== ids.length
+    || new Set(operationIds).size !== ids.length || canonical(ids) !== canonical([...ids].sort())
+    || new Set(expired).size !== expired.length || expired.some(id => !ids.includes(id))) throw new Error("migration_supersession_membership_changed");
+  input.signal.throwIfAborted();
+  const readRows = async (db: ReadDb) => ({
+    workspaces: await db.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, ids))).orderBy(executionWorkspaces.id),
+    operations: await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, companyId), inArray(workspaceOperations.id, operationIds))).orderBy(workspaceOperations.id),
+    snapshots: await db.select().from(executionWorkspaceSnapshots).where(and(eq(executionWorkspaceSnapshots.companyId, companyId), inArray(executionWorkspaceSnapshots.sourceOperationId, operationIds))).orderBy(executionWorkspaceSnapshots.id),
+  });
+  const initial = await readRows(input.db);
+  if (initial.workspaces.length !== ids.length || initial.operations.length !== ids.length
+    || initial.operations.some(operation => operation.phase !== "remote_workspace_migration"
+      || cohort.members.find(member => member.operationId === operation.id)?.workspaceId !== operation.executionWorkspaceId
+      || canonical(operation.metadata?.cohortMigration) !== canonical(cohort))
+    || initial.snapshots.some(snapshot => !["pending", "uploaded"].includes(snapshot.status)
+      || cohort.members.find(member => member.operationId === snapshot.sourceOperationId)?.workspaceId !== snapshot.executionWorkspaceId))
+    throw new Error("migration_supersession_journal_changed");
+  const previous = object(initial.operations[0].metadata?.lastFailure);
+  const receipt = (supersessionId: string, replayed: boolean) => ({ superseded: true, migrationCompleted: false, replayed,
+    companyId, cohortId: cohort.id, supersessionId, members: cohort.members.map(member => ({ ...member,
+      snapshotIds: initial.snapshots.filter(snapshot => snapshot.sourceOperationId === member.operationId).map(snapshot => snapshot.id) })) });
+  if (initial.operations.every(operation => operation.status === "failed" && operation.finishedAt !== null
+    && object(operation.metadata?.lastFailure).schema === SUPERSESSION
+    && object(operation.metadata?.lastFailure).code === "source_expired_during_import"
+    && object(operation.metadata?.lastFailure).operationId === operation.id
+    && object(operation.metadata?.lastFailure).workspaceId === operation.executionWorkspaceId
+    && object(operation.metadata?.lastFailure).companyId === companyId
+    && object(operation.metadata?.lastFailure).cohortId === cohort.id
+    && object(operation.metadata?.lastFailure).supersessionId === previous.supersessionId
+    && canonical(object(operation.metadata?.lastFailure).expiredWorkspaceIds) === canonical(expired)
+    && !Object.hasOwn(operation.metadata ?? {}, "retiredSave")) && typeof previous.supersessionId === "string") return receipt(previous.supersessionId, true);
+  if (initial.operations.some(operation => operation.status !== "running" || operation.finishedAt !== null || operation.metadata?.uncertain === true
+    || Object.hasOwn(operation.metadata ?? {}, "lastFailure") || Object.hasOwn(operation.metadata ?? {}, "retiredSave")))
+    throw new Error("migration_supersession_journal_changed");
+  const observe = async (db: ReadDb) => {
+    const rows = await readRows(db), members = [];
+    if (JSON.stringify(rows) !== JSON.stringify(initial)) throw new Error("migration_supersession_source_changed");
+    for (const workspace of rows.workspaces) {
+      input.signal.throwIfAborted();
+      const operation = rows.operations.find(row => row.executionWorkspaceId === workspace.id)!, marker = object(workspace.metadata?.remoteWorkspaceMigration);
+      if (readRemoteWorkspaceAuthority(workspace.metadata) || marker.schema !== SCHEMA || marker.operationId !== operation.id
+        || marker.sourceFingerprint !== operation.metadata?.sourceFingerprint || canonical(marker.source) !== canonical(operation.metadata?.source)
+        || workspace.metadata?.remoteWorkspaceMigrationSupersession != null || operation.issueId !== workspace.sourceIssueId
+        || operation.cwd !== (workspace.cwd ?? workspace.providerRef)) throw new Error("migration_supersession_source_changed");
+      const memberInput = { db: db as Db, companyId, workspaceId: workspace.id, signal: input.signal };
+      const onlineSource = await resolveMigrationSource(memberInput), current = await inspect(db, { ...memberInput, onlineSource }, cohort);
+      if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`);
+      if (expired.includes(workspace.id)) {
+        const expiration = readGitCacheExpiration(workspace), recipe = current.archivedGitSource?.recipe;
+        if (workspace.status !== "archived" || workspace.cleanupReason !== GIT_CACHE_EXPIRATION_REASON
+          || !expiration?.completed || !expiration.workspaceRemoved || !recipe?.restorePoint || recipe.restoredAt
+          || fingerprint(workspace) === marker.sourceFingerprint) throw new Error("migration_supersession_expiration_unverified");
+      } else if (!onlineSource || fingerprint(workspace) !== marker.sourceFingerprint) throw new Error("migration_supersession_source_changed");
+      members.push({ workspace, operation, onlineSource, current });
+    }
+    return members;
+  };
+  const observed = await observe(input.db), projectIds = [...new Set(observed.flatMap(member => member.onlineSource?.projectIds
+    ?? (member.workspace.projectId ? [member.workspace.projectId] : [])))].sort();
+  const issueIds = [...new Set(observed.flatMap(member => member.onlineSource?.issueIds ?? [member.workspace.sourceIssueId!]))].sort();
+  const lifecycleIds = [...new Set(observed.flatMap(member => [member.workspace.id, ...(member.onlineSource?.workspaceIds ?? [])]))].sort();
+  const publications: Parameters<typeof publishActivity>[0][] = [], supersessionId = randomUUID();
+  await withIndependentWorkspaceSourceLocks(input.db, { companyId, projectIds, issueIds }, async lock => {
+    for (const id of lifecycleIds) if (!await tryAcquireExecutionWorkspaceLifecycleLock(lock, id)) throw new IsolatedRepositorySourceBusy("workspace_source_busy");
+    const current = await observe(input.db);
+    if (canonical(current) !== canonical(observed)) throw new Error("migration_supersession_source_changed");
+    const sources = current.flatMap(member => [...member.current.legacyRecoverySources,
+      ...(member.onlineSource?.companions.flatMap(proof => proof.sources) ?? [])]);
+    const holds = current.flatMap(member => [...member.current.onlineRecoveryHolds,
+      ...(member.onlineSource?.companions.flatMap(proof => proof.holds) ?? [])]);
+    if (!await lockOnlineLegacyWorkspaceRecoverySources(lock as unknown as Db, { companyId, sources, holds }))
+      throw new Error("remote_workspace_migration_blocked:remote_copyback_owned");
+    await lock.transaction(async tx => {
+      await lockMigrationParents(tx, companyId, null, [...observed.map(member => member.workspace.sourceIssueId!),
+        ...sources.map(source => source.issueId)]);
+      await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, ids))).orderBy(executionWorkspaces.id).for("update", { noWait: true });
+      await tx.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, companyId), inArray(workspaceOperations.id, operationIds))).orderBy(workspaceOperations.id).for("update", { noWait: true });
+      for (const snapshot of initial.snapshots) await tx.select().from(executionWorkspaceSnapshots).where(and(
+        eq(executionWorkspaceSnapshots.companyId, companyId), eq(executionWorkspaceSnapshots.id, snapshot.id))).for("update", { noWait: true });
+      const checked = await observe(tx);
+      if (canonical(checked) !== canonical(current)) throw new Error("migration_supersession_source_changed");
+      const at = new Date();
+      for (const member of checked) {
+        input.signal.throwIfAborted();
+        if (member.current.archivedGitSource) await lockArchivedGitMigrationSources(tx, member.workspace, member.current.archivedGitSource);
+        const snapshotIds = initial.snapshots.filter(snapshot => snapshot.sourceOperationId === member.operation.id).map(snapshot => snapshot.id);
+        const lineage = { schema: SUPERSESSION, companyId, workspaceId: member.workspace.id, cohortId: cohort.id,
+          operationId: member.operation.id, supersessionId, snapshotIds };
+        const [operation] = await tx.update(workspaceOperations).set({ status: "failed", finishedAt: at, updatedAt: at,
+          metadata: { ...member.operation.metadata, lastFailure: { ...lineage, code: "source_expired_during_import", at: at.toISOString(), expiredWorkspaceIds: expired } } }).where(and(
+          eq(workspaceOperations.companyId, companyId), eq(workspaceOperations.executionWorkspaceId, member.workspace.id), eq(workspaceOperations.id, member.operation.id),
+          eq(workspaceOperations.phase, "remote_workspace_migration"), eq(workspaceOperations.status, "running"), sql`${workspaceOperations.finishedAt} is null`, eq(workspaceOperations.metadata, member.operation.metadata!))).returning();
+        const metadata: Record<string, unknown> = { ...member.workspace.metadata, remoteWorkspaceMigrationSupersession: lineage }; delete metadata.remoteWorkspaceMigration;
+        const [workspace] = await tx.update(executionWorkspaces).set({ metadata, updatedAt: at }).where(and(
+          eq(executionWorkspaces.companyId, companyId), eq(executionWorkspaces.id, member.workspace.id), eq(executionWorkspaces.metadata, member.workspace.metadata!))).returning();
+        if (!operation || !workspace) throw new Error("migration_supersession_compare_and_swap_failed");
+        publications.push((await persistActivity(tx as unknown as Db, { companyId, actorType: "system", actorId: "remote-workspace-migration",
+          action: "workspace.migration_failed", entityType: "execution_workspace", entityId: workspace.id,
+          details: { code: "source_expired_during_import", cohortId: cohort.id, operationId: operation.id, supersessionId, snapshotIds } })).publication);
+      }
+      await notifyRemoteWorkspaceSnapshots(tx, `cohort:${cohort.id}`);
+      input.signal.throwIfAborted();
+    });
+  }, true, input.sourceLockWaitMs === undefined ? undefined : { waitMs: input.sourceLockWaitMs, signal: input.signal });
+  for (const publication of publications) publishActivity(publication);
+  return receipt(supersessionId, false);
+}