diff --git a/doc/DATABASE.md b/doc/DATABASE.md index f75e62d1a09..69b3adba55a 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -916,6 +916,17 @@ folder. The existing source, provider-stop and archive checks still apply. Completed copyback recovery receipts remain valid. Unknown or unverified recovery sources continue to block import. +Online import can preserve an established provider session's prepared source +after a terminal provider-artifact failure. It requires the exact board-owned +native failure block and the task's current blocked status version, released +controller and coordinator ownership, and all declared allocation bindings. +Source locks and the finalizer lock protect the full action, task and source +fingerprints through publication. Import saves only the current local files. It +retains the prepared reference, provider checkpoint, remote allocations and +recovery hold. The existing migration recipe routes later file recovery away +from the retired local destination. This neither declares preexecution nor +proves remote export, provider termination or permission to resume execution. + The workspace fingerprint pins the recorded source. An arriving reservation can reach a legacy workspace before refusing it, so publication must follow this order too. The lifecycle advisory lock alone does not fence runtime operations. diff --git a/server/src/services/online-legacy-workspace-recovery.ts b/server/src/services/online-legacy-workspace-recovery.ts index 9af3f026fda..2b3e5dc18f7 100644 --- a/server/src/services/online-legacy-workspace-recovery.ts +++ b/server/src/services/online-legacy-workspace-recovery.ts @@ -1,6 +1,6 @@ import { createHash } from "node:crypto"; import { and, eq } from "drizzle-orm"; -import { environmentLeases, executionWorkspaces, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { environmentLeases, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { adapterExecutionControls } from "./adapter-execution-control.js"; import { inspectFinalizedWorkspaceRecoveryHold, type FinalizedWorkspaceRecoveryHold } from "./finalized-workspace-recovery-migration.js"; import { isLegacyWorkspaceTaskRelated, readLegacyWorkspaceRecoveryAllocations, type LegacyWorkspaceRecoverySource } from "./legacy-workspace-recovery-migration.js"; @@ -129,6 +129,27 @@ export async function inspectOnlineLegacyWorkspaceRecoveryHold(db: ReadDb, input if (typeof recovery.evidence?.runId !== "string") return null; const settled = await settledRun(db, input.workspace, recovery.evidence.runId); if (!settled) return null; + const failureBlock = recovery.evidence.nativeFailureBlock as { runId?: unknown; statusVersion?: unknown } | undefined; + if (recovery.kind === "active_run_watchdog" && recovery.status === "active" + && recovery.cause === "runner_remote_provider_artifact_incompatible" && settled.run.status === "failed" + && settled.owner.phase === "terminal_failure" && settled.owner.nextAttemptAt === null && settled.owner.resultId === null + && settled.owner.failureCode === recovery.cause && recovery.sourceIssueId === settled.run.nativeIssueId + && recovery.ownerType === "board" && recovery.ownerAgentId === null && recovery.returnOwnerAgentId === settled.run.agentId + && recovery.wakePolicy === null && recovery.evidence.sourceFailureCode === recovery.cause + && recovery.evidence.recoveryDisposition === recovery.cause && recovery.evidence.providerSessionEstablished === true + && recovery.evidence.recoveryMode === "exact_checkpoint_resume" && failureBlock?.runId === settled.run.id + && Number.isSafeInteger(failureBlock.statusVersion) && Number(failureBlock.statusVersion) > 0 + && recovery.fingerprint === createHash("sha256").update(`${settled.run.id}:${recovery.cause}`).digest("hex") + && !fileKeys.filter(key => key !== "nativeFailureBlock").some(key => recovery.evidence?.[key] != null)) { + const [task] = await db.select({ id: issues.id, status: issues.status, statusVersion: issues.statusVersion, + executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId }).from(issues).where(and( + eq(issues.companyId, input.workspace.companyId), eq(issues.id, recovery.sourceIssueId))); + if (!task || task.status !== "blocked" || task.statusVersion !== failureBlock.statusVersion + || task.executionRunId !== null || task.checkoutRunId !== null) return null; + const source = await inspectOnlineLegacyWorkspaceRecoverySource(db, { workspace: input.workspace, runId: settled.run.id }); + if (source?.reference?.state === "prepared" && source.issueId === recovery.sourceIssueId && source.recoveryAllocations?.length) + return { fingerprint: fingerprint({ recovery, source, task }), finalizedHold: null, preparedSource: source }; + } if (recovery.kind === "active_run_watchdog" && workspaceExportCauses.includes(recovery.cause) && settled.owner.phase === "terminal_failure" && settled.owner.nextAttemptAt === null && settled.owner.failureCode === recovery.cause && recovery.ownerType === "board" diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index 07ea7d30095..c20bf712ec8 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -33,10 +33,11 @@ const SCHEMA = "paperclip.remote-workspace-migration.v1"; type Workspace = typeof executionWorkspaces.$inferSelect; type ReadDb = Pick; type OnlineRecoveryHold = OnlineLegacyWorkspaceRecoveryHold & { actionId: string }; -async function lockMigrationParents(db: ReadDb, companyId: string, issueId: string | null) { +async function lockMigrationParents(db: ReadDb, companyId: string, issueId: string | null, recoveryIssueIds: readonly string[] = []) { await db.select({ id: companies.id }).from(companies).where(eq(companies.id, companyId)).for("key share"); - if (issueId) await db.select({ id: issues.id }).from(issues).where(and( - eq(issues.companyId, companyId), eq(issues.id, issueId))).for("update"); + for (const id of [...new Set([...(issueId ? [issueId] : []), ...recoveryIssueIds])].sort()) + await db.select({ id: issues.id }).from(issues).where(and( + eq(issues.companyId, companyId), eq(issues.id, id))).for("update", recoveryIssueIds.length ? { noWait: true } : undefined); } const object = (value: unknown): Record => value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; @@ -458,7 +459,10 @@ export async function migrateRemoteWorkspace(input: { await lock.execute(sql`select 1`); const publicationDb = onlineSource ? lock as unknown as Db : input.db; const publication = await publicationDb.transaction(async tx => { - await lockMigrationParents(tx, input.companyId, reserved.workspace.sourceIssueId); + await lockMigrationParents(tx, input.companyId, reserved.workspace.sourceIssueId, onlineSource ? [ + ...reserved.legacyRecoverySources.map(source => source.issueId), + ...onlineSource.companions.flatMap(proof => proof.sources.map(source => source.issueId)), + ] : []); for (const actionId of [...new Set([...reserved.finalizedRecoveryHolds, ...reserved.onlineRecoveryHolds].map(held => held.actionId))].sort()) { await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.id, actionId))).for("update", { noWait: true }); diff --git a/server/src/services/remote-workspace-online-source.test.ts b/server/src/services/remote-workspace-online-source.test.ts index 2a603dcbde1..76c8cb99646 100644 --- a/server/src/services/remote-workspace-online-source.test.ts +++ b/server/src/services/remote-workspace-online-source.test.ts @@ -1,12 +1,12 @@ import { execFile } from "node:child_process"; import { promisify } from "node:util"; -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { sql, eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { activityLog, agents, companies, createDb, environmentLeases, environments, executionWorkspaceSnapshots, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, projects, projectWorkspaces, workspaceOperations, workspaceRuntimeServices, type Db } from "@paperclipai/db"; +import { activityLog, agents, companies, createDb, environmentLeases, environments, executionWorkspaceSnapshots, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, projects, projectWorkspaces, workspaceOperations, workspaceRuntimeServices, type Db } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority, readRemoteWorkspaceRunReference } from "@paperclipai/shared"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { resolveManagedProjectWorkspaceDir, resolvePaperclipInstanceRoot } from "../home-paths.js"; @@ -23,6 +23,10 @@ import { tryLockNativeWorkspaceFinalization } from "./native-runtime/native-work import { recordNativeFinalizationFailure } from "./native-runtime/native-run-finalizer.js"; import { inspectOnlineWorkspaceSource } from "./online-workspace-source.js"; import { inspectOnlineLegacyWorkspaceRecoveryHold, inspectOnlineLegacyWorkspaceRecoverySource } from "./online-legacy-workspace-recovery.js"; +import { nativeWorkspaceSyncInternals } from "./native-runtime/native-workspace-sync.js"; +import { readMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js"; +import { verifiedWorkspaceRepairReceipt } from "./workspace-recovery-receipt.js"; +import { adapterExecutionControls, createAdapterExecutionControl } from "./adapter-execution-control.js"; import { withWorkspaceMoveLocks } from "./isolated-repository-rehome.js"; import { ensureRuntimeServicesForRun, startRuntimeServicesForWorkspaceControl, stopRuntimeServicesForExecutionWorkspace } from "./workspace-runtime.js"; @@ -51,6 +55,7 @@ describe.skipIf(!support.supported)("online ordinary physical source fence", () await db.delete(nativeRunFinalizations); await db.delete(environmentLeases); await db.delete(environments); + await db.delete(heartbeatRunEvents); await db.delete(heartbeatRuns); await db.delete(issues); await db.delete(agents); @@ -486,30 +491,174 @@ describe.skipIf(!support.supported)("online ordinary physical source fence", () expect(readRemoteWorkspaceAuthority((await stored(f.workspaceId)).metadata)).toBeNull(); expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("saved legacy edits\n"); }); - async function recoveryFixture() { + async function recoveryFixture(descendant = false) { const f = await fixture(), runId = randomUUID(); + const issue = descendant ? (await db.insert(issues).values({ companyId: f.companyId, projectId: f.projectId, + parentId: f.aliases[0].issue.id, title: "Descendant prepared recovery", executionWorkspaceId: f.workspaceId }).returning())[0] + : f.aliases[0].issue; const [agent] = await db.insert(agents).values({ companyId: f.companyId, name: "Terminal native worker", adapterType: "paperclip_runner" }).returning(); const reference = { schema: "paperclip.native-workspace-sync/v3", state: "prepared", workspaceId: f.workspaceId, leaseId: randomUUID(), providerLeaseId: randomUUID(), remoteCwd: "/workspaces/retained", descriptorSha256: "12".repeat(32), baselineSha256: "34".repeat(32), resourceDisposition: "destroy" }; await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: agent.id, - nativeIssueId: f.aliases[0].issue.id, status: "failed", runtimeMode: "native", runnerProfileJson: { nativeWorkspaceSync: reference } }).returning(); - await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.aliases[0].issue.id, + nativeIssueId: issue.id, status: "failed", runtimeMode: "native", runnerProfileJson: { nativeWorkspaceSync: reference } }).returning(); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: issue.id, runId, phase: "observed" }); const [environment] = await db.insert(environments).values({ name: runId, driver: "daytona" }).returning(); const [lease] = await db.insert(environmentLeases).values({ id: reference.leaseId, companyId: f.companyId, - issueId: f.aliases[0].issue.id, heartbeatRunId: runId, executionWorkspaceId: f.workspaceId, environmentId: environment.id, + issueId: issue.id, heartbeatRunId: runId, executionWorkspaceId: f.workspaceId, environmentId: environment.id, provider: "daytona", providerLeaseId: reference.providerLeaseId, status: "retained", leasePolicy: "retain_on_failure", metadata: { remoteCwd: reference.remoteCwd } }).returning(); await recordNativeFinalizationFailure({ db, runId, error: new Error("native_workspace_sync_out_unrecoverable"), failureScope: "workspace", permanent: true }); const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); const [owner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); - const [warning] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.aliases[0].issue.id)); + const [warning] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issue.id)); expect(warning).toMatchObject({ kind: "active_run_watchdog", cause: "native_workspace_sync_out_unrecoverable", status: "active", evidence: { runId } }); return { ...f, run, owner, lease, warning }; } + async function artifactFailureFixture(descendant = false) { + const f = await recoveryFixture(descendant), cause = "runner_remote_provider_artifact_incompatible", statusVersion = 2; + const profile = { ...f.run.runnerProfileJson, sessionCheckpoint: { backendKind: "runner", providerSessionId: "fixture-established-session" } }; + const restore = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] }; + await db.update(heartbeatRuns).set({ runnerProfileJson: profile, nextEventSeq: 250, + resultJson: { ...f.run.resultJson, workspaceRestoreRecovery: restore } }).where(eq(heartbeatRuns.id, f.run.id)); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: cause, nextAttemptAt: null }) + .where(eq(nativeRunFinalizations.runId, f.run.id)); + await db.update(issues).set({ status: "blocked", statusVersion, executionRunId: null, checkoutRunId: null }) + .where(eq(issues.id, f.run.nativeIssueId!)); + await db.update(issueRecoveryActions).set({ cause, ownerType: "board", ownerAgentId: null, + returnOwnerAgentId: f.run.agentId, wakePolicy: null, + fingerprint: createHash("sha256").update(`${f.run.id}:${cause}`).digest("hex"), evidence: { + runId: f.run.id, coordinatorAttempt: 3, sourceFailureCode: cause, recoveryDisposition: cause, + nativeFailureBlock: { runId: f.run.id, statusVersion }, providerSessionEstablished: true, recoveryMode: "exact_checkpoint_resume", + } }).where(eq(issueRecoveryActions.id, f.warning.id)); + const events = ["session.started", ...Array.from({ length: 124 }, () => ["tool.execution.started", "tool.execution.completed"]).flat()]; + await db.insert(heartbeatRunEvents).values(events.map((eventType, index) => ({ companyId: f.companyId, + runId: f.run.id, agentId: f.run.agentId, seq: index + 1, eventType }))); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)); + const [owner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)); + const [warning] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.warning.id)); + const [issue] = await db.select().from(issues).where(eq(issues.id, run.nativeIssueId!)); + return { ...f, run, owner, warning, issue }; + } + it("preserves provider-established artifact failure holds while migrating local Git aliases and their source recipe", async () => { + const recovery = await artifactFailureFixture(), f = { ...recovery, ...await gitFixture(recovery) }; + const original = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)); + const nativeReference = nativeWorkspaceSyncInternals.readReference(f.run.runnerProfileJson!.nativeWorkspaceSync)!; + const descriptor = nativeWorkspaceSyncInternals.descriptorPath(f.run.id, nativeReference.descriptorSha256); + expect(await fs.lstat(descriptor).catch(() => null)).toBeNull(); + expect(await inventoryRemoteWorkspaceMigration(db, { ...f, workspaceId: f.child.id })).toMatchObject({ eligible: true }); + await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: f.child.id, apply: true }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0]).toEqual(original[0]); + const put = storage.putFile.bind(storage); + vi.spyOn(storage, "putFile").mockImplementationOnce(async input => { + await db.transaction(async tx => { + expect(await tryLockNativeWorkspaceFinalization(tx as unknown as Db, { companyId: f.companyId, runId: f.run.id })).toBe(false); + }); + return put(input); + }); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true }); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)); + expect(run).toMatchObject({ status: "failed", resultJson: f.run.resultJson, runnerProfileJson: { + nativeWorkspaceSync: f.run.runnerProfileJson!.nativeWorkspaceSync, sessionCheckpoint: f.run.runnerProfileJson!.sessionCheckpoint, + } }); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)))[0]).toEqual(f.owner); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]).toEqual(f.lease); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.warning.id)))[0]).toEqual(f.warning); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0]).toEqual(f.issue); + expect(await db.select().from(heartbeatRunEvents)).toHaveLength(249); + expect(await readMigratedLegacyWorkspaceSource(db, { companyId: f.companyId, runId: f.run.id, + reference: nativeReference })).toMatchObject({ runId: f.run.id, + reference: { state: "prepared" }, recoveryAllocations: [{ leaseId: f.lease.id }] }); + expect(await verifiedWorkspaceRepairReceipt(db, run)).toBeNull(); + const authority = readRemoteWorkspaceAuthority((await stored(f.workspaceId)).metadata)!; + const reference = await getWorkspaceSnapshot(db, { ...f, snapshotId: authority.snapshotId! }); + const archivePath = path.join(home, "local-preserved.pcws"), targetDir = path.join(home, "local-restored"); + await loadWorkspaceSnapshotArchive({ storage, reference: reference!, targetPath: archivePath }); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("saved legacy edits\n"); + await fs.rm(f.cwd, { recursive: true }); + await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: reference!.manifestDigest }); + expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe("saved legacy edits\n"); + expect(await fs.lstat(descriptor).catch(() => null)).toBeNull(); + }); + it.each(["stale-block", "foreign-company", "expired-owner", "live-controller", "live-control", "malformed-block", "missing-allocation", "fingerprint"])( + "keeps a prepared artifact failure held with %s evidence", async mode => { + const f = await artifactFailureFixture(); + if (mode === "stale-block") await db.update(issues).set({ statusVersion: f.issue.statusVersion + 1 }).where(eq(issues.id, f.issue.id)); + if (mode === "expired-owner") await db.update(nativeRunFinalizations).set({ leaseOwner: "unreleased owner", leaseExpiresAt: new Date(0) }) + .where(eq(nativeRunFinalizations.runId, f.run.id)); + if (mode === "live-controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(heartbeatRuns.id, f.run.id)); + if (mode === "malformed-block") await db.update(issueRecoveryActions).set({ evidence: { + ...f.warning.evidence, nativeFailureBlock: { runId: f.run.id, statusVersion: String(f.issue.statusVersion) }, + } }).where(eq(issueRecoveryActions.id, f.warning.id)); + if (mode === "missing-allocation") await db.update(heartbeatRuns).set({ resultJson: { + ...f.run.resultJson, workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [randomUUID()] }, + } }).where(eq(heartbeatRuns.id, f.run.id)); + if (mode === "fingerprint") await db.update(issueRecoveryActions).set({ fingerprint: "00".repeat(32) }).where(eq(issueRecoveryActions.id, f.warning.id)); + const control = createAdapterExecutionControl(); + if (mode === "live-control") adapterExecutionControls.set(f.run.id, control); + try { + const [saved] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.warning.id)); + const recovery = mode === "foreign-company" ? { ...saved, companyId: randomUUID() } : saved; + expect(await inspectOnlineLegacyWorkspaceRecoveryHold(db, { workspace: await stored(f.workspaceId), recovery })).toBeNull(); + if (mode !== "foreign-company") { + expect(await inventoryRemoteWorkspaceMigration(db, f)).toMatchObject({ eligible: false }); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow(); + } + expect(await db.select().from(workspaceOperations)).toHaveLength(0); + expect(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(0); + expect(readRemoteWorkspaceAuthority((await stored(f.workspaceId)).metadata)).toBeNull(); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("saved legacy edits\n"); + } finally { adapterExecutionControls.delete(f.run.id); control.finish(); } + }, + ); + it.each(["warning", "task", "checkpoint"])("rejects a prepared artifact hold when %s changes during capture", async mode => { + const f = await artifactFailureFixture(), put = storage.putFile.bind(storage); + vi.spyOn(storage, "putFile").mockImplementationOnce(async input => { + const result = await put(input); + if (mode === "warning") await db.update(issueRecoveryActions).set({ evidence: { ...f.warning.evidence, coordinatorAttempt: 4 } }) + .where(eq(issueRecoveryActions.id, f.warning.id)); + if (mode === "task") await db.update(issues).set({ statusVersion: f.issue.statusVersion + 1 }).where(eq(issues.id, f.issue.id)); + if (mode === "checkpoint") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.run.runnerProfileJson, + sessionCheckpoint: { backendKind: "runner", providerSessionId: "changed-provider-session" }, + } }).where(eq(heartbeatRuns.id, f.run.id)); + return result; + }); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow(); + expect(readRemoteWorkspaceAuthority((await stored(f.workspaceId)).metadata)).toBeNull(); + expect((await db.select().from(workspaceOperations))[0].status).toBe("running"); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)))[0]).toEqual(f.owner); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]).toEqual(f.lease); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("saved legacy edits\n"); + }); + it.each(["selected", "companion"])("does not publish a %s source while its descendant recovery task is locked", async mode => { + const recovery = await artifactFailureFixture(true), linked = mode === "companion" ? await gitFixture(recovery) : null; + const f = { ...recovery, ...linked }; + const descendant = f.issue; + const input = { companyId: f.companyId, workspaceId: linked?.child.id ?? f.workspaceId }; + expect(await inventoryRemoteWorkspaceMigration(db, input)).toMatchObject({ eligible: true }); + const other = createDb(database.connectionString, { maxConnections: 1 }); + let entered!: () => void, release!: () => void; + const started = new Promise(resolve => { entered = resolve; }); + const pending = new Promise(resolve => { release = resolve; }); + const held = other.transaction(async tx => { + await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, descendant.id)).for("update"); + entered(); await pending; + }); + await started; + try { + await expect(migrateRemoteWorkspace({ db, storage, ...input, apply: true })) + .rejects.toThrow("remote_workspace_migration_blocked:execution_rows_busy"); + expect(readRemoteWorkspaceAuthority((await stored(input.workspaceId)).metadata)).toBeNull(); + expect((await db.select().from(workspaceOperations))[0].status).toBe("running"); + expect((await db.select().from(issues).where(eq(issues.id, descendant.id)))[0]).toEqual(descendant); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.warning.id)))[0]) + .toMatchObject({ status: "active", sourceIssueId: descendant.id, evidence: f.warning.evidence }); + } finally { release(); await held; await other.$client.end({ timeout: 1 }); } + }); it("discovers all Git companions in one fresh company-scoped scan per inspection", async () => { const recovery = await recoveryFixture(), f = await gitFixture(recovery); const [finalizedIssue] = await db.insert(issues).values({ companyId: f.companyId, projectId: f.projectId,