diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 7de63c7af15..5f521eefa9c 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -4622,3 +4622,15 @@ and background revalidation still apply. Recent-window catch-up retains messages observed before navigation. Agent Chat can render from the current person's company-scoped navigation list, carrying its original fetch timestamp so stale conversations refresh in the background. No conversation is created by a read. + +### Remote repository admission saves + +Remote admission validates each registered repository against its current sandbox, +origin, and pinned commit. A matching descriptor in the committed snapshot receipt +allows admission without another archive upload. New repositories and legacy +repositories without that receipt require one workspace snapshot for the admission +batch. Repository ready receipts publish after the snapshot commits. An upload +failure remains retryable even when the clone directory already exists. Later +workspace saves retain repository coverage only for directories present in the +captured archive. Deleted checkouts lose coverage and can be acquired again. +A failed batch upload marks the affected repository rows as needing repair. diff --git a/server/src/__tests__/heartbeat-workspace-session.test.ts b/server/src/__tests__/heartbeat-workspace-session.test.ts index 790c59bd656..f39b0b28ff2 100644 --- a/server/src/__tests__/heartbeat-workspace-session.test.ts +++ b/server/src/__tests__/heartbeat-workspace-session.test.ts @@ -1982,6 +1982,22 @@ describe("effective run execution workspace config freshness", () => { expect(serialized).not.toContain("0.0.0.0"); }); + it("reports actual reuse when freshness policy requested replacement", () => { + const decision = resolveExecutionWorkspaceConfigFreshness({ + hasExistingWorkspace: true, + existingWorkspaceMetadata: {}, + nextMetadata: buildWorkspaceConfigMetadata(), + }); + const operation = buildWorkspaceConfigFreshnessOperation({ + decision, hasExistingWorkspace: true, reuseRequested: true, + workspaceReused: true, configSnapshotRefreshed: true, + previousWorkspaceId: "same", activeWorkspaceId: "same", + }); + expect(operation?.metadata.action).toBe("replace"); + expect(operation?.system).toContain("reused execution workspace"); + expect(operation?.system).not.toContain("replaced execution workspace"); + }); + it("does not record a freshness operation when an unchanged workspace is simply reused", () => { const metadata = buildWorkspaceConfigMetadata(); const decision = resolveExecutionWorkspaceConfigFreshness({ diff --git a/server/src/__tests__/task-workspace-repositories-persistence.test.ts b/server/src/__tests__/task-workspace-repositories-persistence.test.ts index 0f023f500b0..316a4b4088d 100644 --- a/server/src/__tests__/task-workspace-repositories-persistence.test.ts +++ b/server/src/__tests__/task-workspace-repositories-persistence.test.ts @@ -66,7 +66,7 @@ const support = externalDatabaseUrl ? { supported: true } : await getEmbeddedPos request: { repository: { kind: "catalog", id: "123" }, requestKey: "remote-repository" } }); const pin = "a".repeat(40); const prepare = vi.spyOn(remoteWorkspaces, "prepareRemoteWorkspaceRepository").mockResolvedValue({ - cwd: `${remoteCwd}/${intent.repository.relativePath}`, pinnedCommit: pin, branchName: "main" }); + cwd: `${remoteCwd}/${intent.repository.relativePath}`, pinnedCommit: pin, branchName: "main", requiresSave: false }); try { const prepared = await prepareForAdmission({ companyId, issueId: taskId, workspaceId: remoteWorkspaceId, cwd: remoteCwd, agentId, runId, responsibleUserId: "local-board", diff --git a/server/src/services/execution-workspace-repositories.ts b/server/src/services/execution-workspace-repositories.ts index 6ac6150a864..3c8e4559b5f 100644 --- a/server/src/services/execution-workspace-repositories.ts +++ b/server/src/services/execution-workspace-repositories.ts @@ -1,5 +1,5 @@ import { measureRunTransaction } from "./execution-transaction-timing.js"; -import { prepareRemoteWorkspaceRepository, readRemoteWorkspaceAuthority } from "./remote-workspaces.js"; +import { prepareRemoteWorkspaceRepository, saveRemoteWorkspace, readRemoteWorkspaceAuthority } from "./remote-workspaces.js"; import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; import type { EnvironmentLease } from "@paperclipai/shared"; import { beginRunPhase, measureRunPhase } from "./execution-preparation-timing.js"; @@ -366,6 +366,8 @@ export function executionWorkspaceRepositoryService(db: Db) { const resolveAuth = createGitRemoteAuthProvider(db, input.companyId, { issueId: input.issueId, heartbeatRunId: input.runId, agentId: input.agentId, responsibleUserId: input.responsibleUserId }); const prepared: Array<{ id: string; cwd: string; repoUrl: string; relativePath: string; pinnedCommit: string; branchName: string | null }> = []; + let requiresSave = false; + const verifications: Array<() => Promise> = []; for (const row of rows) { assertAcquisitionReceipt(row, acquisitionRestricted); const source = resolveAcquisitionSource(row.catalogRepositoryId ? { kind: "catalog", id: row.catalogRepositoryId } @@ -392,13 +394,12 @@ export function executionWorkspaceRepositoryService(db: Db) { repository: { id: row.id, relativePath: row.relativePath, repoUrl: row.repoUrl, ref: row.requestedRef === "HEAD" ? null : row.requestedRef, ...(row.pinnedCommit ? { pinnedCommit: row.pinnedCommit } : {}) }, - assertOwned: verify, verifyOnly: acquisitionRestricted, + assertOwned: verify, verifyOnly: acquisitionRestricted, deferSave: true, resolveGitAuth: source.catalogRepositoryId ? resolveAuth : async () => anonymous, }); await verify(); - await db.update(executionWorkspaceRepositories).set({ state: "ready", pinnedCommit: result.pinnedCommit, - branchName: result.branchName, failureCode: null, updatedAt: new Date() }) - .where(and(eq(executionWorkspaceRepositories.id, row.id), eq(executionWorkspaceRepositories.companyId, input.companyId))); + requiresSave ||= result.requiresSave; + verifications.push(verify); prepared.push({ ...result, id: row.id, repoUrl: row.repoUrl, relativePath: row.relativePath }); } catch (error) { await db.update(executionWorkspaceRepositories).set({ state: "needs_repair", failureCode: "repository_preparation_failed", updatedAt: new Date() }) @@ -406,6 +407,26 @@ export function executionWorkspaceRepositoryService(db: Db) { throw error; } } + for (const verify of verifications) await verify(); + if (requiresSave) { + try { + await saveRemoteWorkspace({ db, workspace, runId: input.runId, lease: input.remoteLease, + target: input.remoteTarget, boundary: `repositories:${prepared.map(row => `${row.id}:${row.pinnedCommit}`).sort().join(",")}`, + repositories: prepared.map(row => ({ id: row.id, relativePath: row.relativePath, repoUrl: row.repoUrl, + ref: rows.find(source => source.id === row.id)!.requestedRef, pinnedCommit: row.pinnedCommit })), + assertOwned: async () => { for (const verify of verifications) await verify(); } }); + } catch (error) { + for (const result of prepared) await db.update(executionWorkspaceRepositories) + .set({ state: "needs_repair", failureCode: "repository_preparation_failed", updatedAt: new Date() }) + .where(and(eq(executionWorkspaceRepositories.id, result.id), eq(executionWorkspaceRepositories.companyId, input.companyId))); + throw error; + } + } + for (const result of prepared) { + await db.update(executionWorkspaceRepositories).set({ state: "ready", pinnedCommit: result.pinnedCommit, + branchName: result.branchName, failureCode: null, updatedAt: new Date() }) + .where(and(eq(executionWorkspaceRepositories.id, result.id), eq(executionWorkspaceRepositories.companyId, input.companyId))); + } return prepared; } const root = await measureRunPhase("workspace.repository.paths", () => fs.realpath(input.cwd)); diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 8b361dfe0ba..4a4601844bf 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -272,7 +272,7 @@ function preserveWorkspaceMetadataForPatch( if (patch[key] !== undefined && patch[key] !== current[key]) throw conflict("Remote workspace coordinates belong to its sandbox storage"); } } - const remoteAuthorityKeys = ["remoteWorkspace", "remoteWorkspaceMigration", "remoteWorkspaceRecoverySourceRunIds", "remoteWorkspaceRecoveredFiles", "legacyWorkspaceRecoveryMigration"] as const; + const remoteAuthorityKeys = ["remoteWorkspace", "remoteWorkspaceSnapshotRepositories", "lifecycleGeneration", "reopenPendingConsumption", "reopenPendingConsumptionSince", "remoteWorkspaceMigration", "remoteWorkspaceRecoverySourceRunIds", "remoteWorkspaceRecoveredFiles", "legacyWorkspaceRecoveryMigration"] as const; if (patch.metadata !== undefined && remoteAuthorityKeys.some(key => current.metadata?.[key] !== undefined || patch.metadata?.[key] !== undefined)) { nextPatch.metadata = { ...patch.metadata }; for (const key of remoteAuthorityKeys) { diff --git a/server/src/services/heartbeat/workspace-preparation.test.ts b/server/src/services/heartbeat/workspace-preparation.test.ts index f0d5d898ce6..136775ebbf4 100644 --- a/server/src/services/heartbeat/workspace-preparation.test.ts +++ b/server/src/services/heartbeat/workspace-preparation.test.ts @@ -331,6 +331,8 @@ describe.skipIf(!support.supported && !externalDatabaseUrl)("heartbeat workspace expect(retained.metadata?.configFingerprint).toMatchObject({ workspaceHash: (saved.metadata?.configFingerprint as { workspaceHash: string }).workspaceHash, }); + if (!strategyChange) expect((retained.metadata?.configFingerprint as { lastEvaluatedAt: string }).lastEvaluatedAt) + .not.toBe((saved.metadata?.configFingerprint as { lastEvaluatedAt: string }).lastEvaluatedAt); expect(retained.metadata?.remoteWorkspace).toMatchObject({ writerRunId: nextRun.id }); } } finally { diff --git a/server/src/services/heartbeat/workspace-preparation.ts b/server/src/services/heartbeat/workspace-preparation.ts index 5f3bc76406c..06d5f019937 100644 --- a/server/src/services/heartbeat/workspace-preparation.ts +++ b/server/src/services/heartbeat/workspace-preparation.ts @@ -1025,7 +1025,7 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace persistedExecutionWorkspace = await executionWorkspacesSvc.getStoredById(agent.companyId, reserved.id); if (!persistedExecutionWorkspace?.cwd) throw new Error("remote_workspace_reservation_missing"); persistedExecutionWorkspace = await executionWorkspacesSvc.update(reserved.id, { - metadata: { ...nextExecutionWorkspaceMetadata, ...reserved.metadata, + metadata: { ...reserved.metadata, ...nextExecutionWorkspaceMetadata, remoteWorkspace: reserved.metadata?.remoteWorkspace }, }); if (!persistedExecutionWorkspace?.cwd) throw new Error("remote_workspace_reservation_missing"); diff --git a/server/src/services/heartbeat/workspaces.ts b/server/src/services/heartbeat/workspaces.ts index 86023d53494..d9787c96580 100644 --- a/server/src/services/heartbeat/workspaces.ts +++ b/server/src/services/heartbeat/workspaces.ts @@ -2616,7 +2616,8 @@ export function buildWorkspaceConfigFreshnessOperation( previousWorkspaceId: input.previousWorkspaceId, activeWorkspaceId: input.activeWorkspaceId, }, - system: `[paperclip] ${workspaceConfigFreshnessActionLabel(input.decision.action)} after config freshness check${categorySummary}: ${reasonSummary}\n`, + system: `[paperclip] ${(input.workspaceReused && input.decision.action === "replace" + ? "reused execution workspace" : workspaceConfigFreshnessActionLabel(input.decision.action))} after config freshness check${categorySummary}: ${reasonSummary}\n`, }; } diff --git a/server/src/services/remote-workspace-snapshot-archive.test.ts b/server/src/services/remote-workspace-snapshot-archive.test.ts index 078cee2580c..996eac85864 100644 --- a/server/src/services/remote-workspace-snapshot-archive.test.ts +++ b/server/src/services/remote-workspace-snapshot-archive.test.ts @@ -6,7 +6,7 @@ import os from "node:os"; import path from "node:path"; import { promisify } from "node:util"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { captureRemoteWorkspaceDirectory, remoteWorkspaceArchiveScript, remoteWorkspaceCaptureFailureMessage, restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; +import { captureRemoteWorkspaceDirectory, readRemoteWorkspaceArchiveDirectories, remoteWorkspaceArchiveScript, remoteWorkspaceCaptureFailureMessage, restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; const archiveIo = vi.hoisted(() => ({ afterChmod: undefined as ((target: string, mode: number) => void) | undefined, @@ -54,6 +54,25 @@ describe("portable remote workspace archives", () => { } return JSON.stringify(entries); } + it("reads captured directory coverage and rejects a different manifest digest", async () => { + const cwd = await source(); + await fs.mkdir(path.join(cwd, "checkout")); + await fs.writeFile(path.join(cwd, "checkout", "file.txt"), "saved"); + const snapshot = await capture(cwd); + expect(await readRemoteWorkspaceArchiveDirectories(snapshot.archivePath, snapshot.manifestDigest)) + .toEqual(new Set(["checkout"])); + await expect(readRemoteWorkspaceArchiveDirectories(snapshot.archivePath, "0".repeat(64))) + .rejects.toThrow("manifest_mismatch"); + }); + + it("rejects oversized coverage manifests before reading their body", async () => { + const archivePath = path.join(root, "oversized.pcws"), header = Buffer.alloc(10); + header.write("PCWS1\n"); header.writeUInt32BE(64 * 1024 * 1024 + 1, 6); + await fs.writeFile(archivePath, header); + await expect(readRemoteWorkspaceArchiveDirectories(archivePath, "0".repeat(64))) + .rejects.toThrow("manifest_too_large"); + }); + it("keeps every serialized capture validation reason in the closed diagnostic contract", () => { const script = remoteWorkspaceArchiveScript(), start = script.indexOf("const archive = "); const restore = script.indexOf("if (!input.targetDir", start); diff --git a/server/src/services/remote-workspace-snapshot-archive.ts b/server/src/services/remote-workspace-snapshot-archive.ts index dc499c9e26e..cbe63dc7cf5 100644 --- a/server/src/services/remote-workspace-snapshot-archive.ts +++ b/server/src/services/remote-workspace-snapshot-archive.ts @@ -558,6 +558,30 @@ export function captureRemoteWorkspaceDirectory(input: { cwd: string; archivePat return workspaceArchive({ ...input, kind: "capture" }); } +export async function readRemoteWorkspaceArchiveDirectories(archivePath: string, expectedManifestDigest: string) { + const fs = await import("node:fs/promises"); + const crypto = await import("node:crypto"); + const archive = await fs.open(archivePath, "r"); + try { + const header = Buffer.alloc(10); + if ((await archive.read(header, 0, 10, 0)).bytesRead !== 10 || header.subarray(0, 6).toString() !== "PCWS1\n") { + throw new Error("remote_workspace_archive_invalid_header"); + } + const size = header.readUInt32BE(6); + if (size > 64 * 1024 * 1024) throw new Error("remote_workspace_archive_manifest_too_large"); + const bytes = Buffer.alloc(size); + if ((await archive.read(bytes, 0, size, 10)).bytesRead !== size) throw new Error("remote_workspace_archive_truncated"); + if (crypto.createHash("sha256").update(bytes).digest("hex") !== expectedManifestDigest) { + throw new Error("remote_workspace_archive_manifest_mismatch"); + } + const manifest = JSON.parse(bytes.toString()) as { schema: string; entries: Array<{ type: string; path: string }> }; + if (manifest.schema !== "paperclip.remote-workspace-archive.v1" || !Array.isArray(manifest.entries)) { + throw new Error("remote_workspace_archive_manifest_mismatch"); + } + return new Set(manifest.entries.filter(entry => entry.type === "directory").map(entry => entry.path)); + } finally { await archive.close(); } +} + export function restoreRemoteWorkspaceArchive(input: { archivePath: string; targetDir: string; expectedManifestDigest: string; signal?: AbortSignal; maxEntries?: number; maxManifestBytes?: number }) { return workspaceArchive({ ...input, kind: "restore" }); } diff --git a/server/src/services/remote-workspaces.integration.test.ts b/server/src/services/remote-workspaces.integration.test.ts index 7a997f41482..7b8e13bf206 100644 --- a/server/src/services/remote-workspaces.integration.test.ts +++ b/server/src/services/remote-workspaces.integration.test.ts @@ -734,24 +734,41 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem const raw = input.args?.at(-1); if (typeof raw === "string" && raw.startsWith("{")) { const payload = JSON.parse(raw) as { repoUrl?: string; authArgs?: string[] }; - if (payload.repoUrl === repoUrl && Array.isArray(payload.authArgs)) input = { ...input, + if (payload.repoUrl?.startsWith(repoUrl) && Array.isArray(payload.authArgs)) input = { ...input, args: [...input.args!.slice(0, -1), JSON.stringify({ ...payload, - authArgs: [...payload.authArgs, "-c", `url.${fixtureOrigin}.insteadOf=${repoUrl}`] })] }; + authArgs: [...payload.authArgs, "-c", `url.${fixtureOrigin}.insteadOf=${payload.repoUrl}`] })] }; } return execute(input); }; const restored = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId: nextRun, lease: allocated as EnvironmentLease, target: remote }); expect(await fs.readFile(path.join(restored.remoteRoot, "saved.txt"), "utf8")).toBe("saved before repository request\n"); + const secondUrl = `${repoUrl}-two`; + const secondPath = taskRepositoryRelativePath(secondUrl); + await db.insert(executionWorkspaceRepositories).values({ companyId: ids.company, + executionWorkspaceId: local.id, requestedByIssueId: ids.issue, repoUrl: secondUrl, repositoryIdentity: secondUrl, + relativePath: secondPath, requestedRef: "HEAD", requestKey: "second-request", requestKeys: ["second-request"] }); + const admission = { companyId: ids.company, issueId: ids.issue, workspaceId: local.id, cwd: restored.remoteRoot, + agentId: ids.agent, runId: nextRun, responsibleUserId: "local-board", remoteTarget: remote, remoteLease: restored.lease }; + const beforeAdmission = await db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.executionWorkspaceId, local.id), eq(workspaceOperations.phase, "remote_workspace_snapshot"))); const prepared = await executionWorkspaceRepositoryService(db).prepareForAdmission({ companyId: ids.company, issueId: ids.issue, workspaceId: local.id, cwd: restored.remoteRoot, agentId: ids.agent, runId: nextRun, responsibleUserId: "local-board", remoteTarget: remote, remoteLease: restored.lease }); - expect(prepared).toMatchObject([{ id: repository.id, relativePath, - pinnedCommit: (await git(origin, "rev-parse", "HEAD")).stdout.trim() }]); + expect(prepared).toHaveLength(2); + expect(prepared).toEqual(expect.arrayContaining([expect.objectContaining({ id: repository.id, relativePath, + pinnedCommit: (await git(origin, "rev-parse", "HEAD")).stdout.trim() })])); + expect(await fs.readFile(path.join(restored.remoteRoot, secondPath, "code.txt"), "utf8")).toBe("acquired on next admission\n"); + const afterAdmission = await db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.executionWorkspaceId, local.id), eq(workspaceOperations.phase, "remote_workspace_snapshot"))); + expect(afterAdmission).toHaveLength(beforeAdmission.length + 1); + const savedId = readRemoteWorkspaceAuthority((await workspace(local.id)).metadata)!.snapshotId; + await executionWorkspaceRepositoryService(db).prepareForAdmission(admission); + expect(readRemoteWorkspaceAuthority((await workspace(local.id)).metadata)!.snapshotId).toBe(savedId); expect(await fs.readFile(path.join(restored.remoteRoot, relativePath, "code.txt"), "utf8")).toBe("acquired on next admission\n"); expect(await db.select().from(executionWorkspaceRepositories).where(eq(executionWorkspaceRepositories.id, repository.id))) - .toMatchObject([{ ...repository, state: "ready", pinnedCommit: prepared[0]!.pinnedCommit, - branchName: prepared[0]!.branchName, updatedAt: expect.any(Date) }]); + .toMatchObject([{ ...repository, state: "ready", pinnedCommit: prepared.find(row => row.id === repository.id)!.pinnedCommit, + branchName: prepared.find(row => row.id === repository.id)!.branchName, updatedAt: expect.any(Date) }]); } finally { await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); } }, 60_000); @@ -809,6 +826,7 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem // Registered admission encodes the stored default HEAD as null. It must // verify the same saved descriptor without changing the source identity. await expect(prepareRemoteWorkspaceRepository(admission)).resolves.toMatchObject({ cwd: secondary, pinnedCommit }); + expect(readRemoteWorkspaceAuthority((await workspace(local.id)).metadata)!.snapshotId).toBe(imported.snapshotId); const beforeChanged = readRemoteWorkspaceAuthority((await workspace(local.id)).metadata)!; await expect(prepareRemoteWorkspaceRepository({ ...admission, workspace: await workspace(local.id), repository: { ...repository, ref: null, repoUrl: newOrigin } })) @@ -816,9 +834,33 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem expect(readRemoteWorkspaceAuthority((await workspace(local.id)).metadata)).toEqual(beforeChanged); expect(resolveGitAuth).not.toHaveBeenCalled(); const newRepository = { ...added, id: randomUUID(), relativePath: "wrapper-new", ref: null }; + const failingStorage: StorageService = { ...storage, putFile: async () => { throw new Error("repository upload failed"); } }; + await expect(prepareRemoteWorkspaceRepository({ ...admission, workspace: await workspace(local.id), + repository: newRepository, storage: failingStorage })).rejects.toThrow("repository upload failed"); await expect(prepareRemoteWorkspaceRepository({ ...admission, workspace: await workspace(local.id), repository: newRepository })) .resolves.toMatchObject({ cwd: path.join(restored.remoteRoot, "wrapper-new") }); + const durable = await workspace(local.id); + const durableId = readRemoteWorkspaceAuthority(durable.metadata)!.snapshotId; + await prepareRemoteWorkspaceRepository({ ...admission, workspace: durable, + repository: { ...newRepository, pinnedCommit: (await git(newOrigin, "rev-parse", "HEAD")).stdout.trim() } }); + expect(readRemoteWorkspaceAuthority((await workspace(local.id)).metadata)!.snapshotId).toBe(durableId); + await saveRemoteWorkspace({ ...admission, workspace: await workspace(local.id), boundary: "repository-admission-batch", + assertOwned: async () => undefined, + repositories: [{ ...newRepository, pinnedCommit: (await git(newOrigin, "rev-parse", "HEAD")).stdout.trim() }] }); + const afterBatch = await workspace(local.id); + await expect(prepareRemoteWorkspaceRepository({ ...admission, workspace: afterBatch })) + .resolves.toMatchObject({ cwd: secondary, pinnedCommit }); + expect(readRemoteWorkspaceAuthority((await workspace(local.id)).metadata)!.snapshotId) + .toBe(readRemoteWorkspaceAuthority(afterBatch.metadata)!.snapshotId); expect(resolveGitAuth).toHaveBeenCalledOnce(); + await fs.rm(path.join(restored.remoteRoot, "wrapper-new"), { recursive: true }); + await saveRemoteWorkspace({ ...admission, workspace: await workspace(local.id), boundary: "after-checkout-removal", + assertOwned: async () => undefined }); + const removed = await workspace(local.id); + expect((removed.metadata?.remoteWorkspaceSnapshotRepositories as { repositories: Array<{ id: string }> }).repositories) + .not.toEqual(expect.arrayContaining([expect.objectContaining({ id: newRepository.id })])); + await expect(prepareRemoteWorkspaceRepository({ ...admission, workspace: removed, repository: newRepository })) + .resolves.toMatchObject({ cwd: path.join(restored.remoteRoot, "wrapper-new") }); expect(await fs.readFile(path.join(restored.remoteRoot, "wrapper-new", "new.txt"), "utf8")).toBe("newly acquired work\n"); expect(readRemoteWorkspaceAuthority((await workspace(local.id)).metadata)!.source).toEqual(source); }, 60_000); diff --git a/server/src/services/remote-workspaces.ts b/server/src/services/remote-workspaces.ts index 0b0494e5055..8af46c45c4e 100644 --- a/server/src/services/remote-workspaces.ts +++ b/server/src/services/remote-workspaces.ts @@ -32,7 +32,7 @@ import { import { beginWorkspaceSnapshot, getWorkspaceSnapshot, loadWorkspaceSnapshotArchive, uploadWorkspaceSnapshot, } from "./remote-workspace-snapshot-storage.js"; -import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; +import { restoreRemoteWorkspaceArchive, readRemoteWorkspaceArchiveDirectories } from "./remote-workspace-snapshot-archive.js"; import { removeTaskDirectoryTree } from "./isolated-task-directory-snapshot.js"; import { attachPreservedWorkspaceSnapshots } from "./workspace-recovery-snapshot.js"; import { notifyRemoteWorkspaceSnapshots } from "./remote-workspace-snapshot-retention.js"; @@ -642,6 +642,7 @@ async function settleSaveInvocation(input: { db: Db; workspace: Workspace; runId export async function saveRemoteWorkspace(input: { db: Db; workspace: Workspace; runId: string; lease: EnvironmentLease; target: AdapterExecutionTarget | null; assertOwned: () => Promise; storage?: StorageService; boundary?: string; + repositories?: NonNullable; }) { const target = requireRemoteWorkspaceTarget(input.target), boundary = input.boundary ?? "final"; const expected = readRemoteWorkspaceAuthority(input.workspace.metadata); @@ -693,6 +694,7 @@ export async function saveRemoteWorkspace(input: { return created!; }); let captured: Awaited> | null = null; + let publicationScratch: string | undefined; let failure: { code: string; cause?: string } | undefined; try { const snapshot = await beginWorkspaceSnapshot(input.db, { companyId: input.workspace.companyId, workspaceId: input.workspace.id, @@ -709,6 +711,13 @@ export async function saveRemoteWorkspace(input: { companyId: input.workspace.companyId, workspaceId: input.workspace.id, archivePath: captured?.archivePath ?? "", manifestDigest: existingUpload?.manifestDigest ?? captured!.manifestDigest }); await input.assertOwned(); + let publicationArchive = captured?.archivePath; + if (!publicationArchive) { + publicationScratch = await mkdtemp(path.join(os.tmpdir(), "pc-workspace-publication-")); + publicationArchive = path.join(publicationScratch, "workspace.pcws"); + await loadWorkspaceSnapshotArchive({ storage: input.storage ?? getStorageService(), reference: saved, targetPath: publicationArchive }); + } + const capturedDirectories = await readRemoteWorkspaceArchiveDirectories(publicationArchive, saved.manifestDigest); const at = new Date(); await input.db.transaction(async tx => { await lockWorkspaceRunParents(tx, input.workspace.companyId, input.runId, operation.issueId); @@ -726,8 +735,15 @@ export async function saveRemoteWorkspace(input: { if (!committed) throw new Error("remote_workspace_snapshot_publication_changed"); await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`); if (boundary === "final") await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`); + const receipt = object(workspace.metadata?.remoteWorkspaceSnapshotRepositories); + const priorRepositories = receipt.snapshotId === current.snapshotId && Array.isArray(receipt.repositories) + ? remoteWorkspaceSourceSchema.parse({ kind: "empty", repositories: receipt.repositories }).repositories ?? [] : []; + const repositories = [...new Map([ + ...priorRepositories, ...(expected.source.repositories ?? []), ...(input.repositories ?? []), + ].map(repository => [repository.id, repository])).values()] + .filter(repository => !repository.readOnly && capturedDirectories.has(repository.relativePath)); await tx.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspace: authority, - remoteWorkspaceSnapshotRepositories: { snapshotId: saved.id, repositories: expected.source.repositories ?? [] }, + remoteWorkspaceSnapshotRepositories: { snapshotId: saved.id, repositories }, }, updatedAt: at }).where(eq(executionWorkspaces.id, workspace.id)); await tx.update(workspaceOperations).set({ status: "succeeded", finishedAt: at, updatedAt: at, metadata: sql`coalesce(${workspaceOperations.metadata}, '{}'::jsonb) || ${JSON.stringify({ snapshotId: saved.id, sha256: saved.sha256 })}::jsonb` }) @@ -747,6 +763,7 @@ export async function saveRemoteWorkspace(input: { } finally { // Every upload/capture promise has settled before its durable token clears. // Missing completion evidence remains pinned; cleanup never masks the save error. + if (publicationScratch) await rm(publicationScratch, { recursive: true, force: true }).catch(() => undefined); try { await captured?.cleanup(); } catch { logger.warn({ runId: input.runId, operationId: operation.id }, "Remote workspace local archive cleanup failed"); } try { @@ -762,8 +779,9 @@ export async function saveRemoteWorkspace(input: { export async function prepareRemoteWorkspaceRepository(input: { db: Db; workspace: Workspace; runId: string; lease: EnvironmentLease; target: AdapterExecutionTarget | null; repository: { id: string; relativePath: string; repoUrl: string; ref: string | null; pinnedCommit?: string; readOnly?: boolean }; - assertOwned?: () => Promise; resolveGitAuth?: ReturnType; verifyOnly?: boolean; storage?: StorageService; + assertOwned?: () => Promise; resolveGitAuth?: ReturnType; verifyOnly?: boolean; storage?: StorageService; deferSave?: boolean; }) { + input = { ...input, workspace: await input.db.transaction(tx => currentWorkspace(tx, input.workspace.companyId, input.workspace.id)) }; const authority = readRemoteWorkspaceAuthority(input.workspace.metadata); if (!authority?.location || authority.writerRunId !== input.runId || authority.location.leaseId !== input.lease.id || authority.location.providerLeaseId !== input.lease.providerLeaseId) throw new Error("remote_workspace_repository_binding_mismatch"); @@ -773,9 +791,18 @@ export async function prepareRemoteWorkspaceRepository(input: { ...input.repository, verifyOnly: input.verifyOnly, savedArchive: isSavedWorkspaceRepository(input.workspace, authority, input.repository), resolveGitAuth: input.resolveGitAuth ?? await createRemoteWorkspaceGitAuth(input.db, input.workspace.companyId, input.runId) }); await input.assertOwned?.(); - await saveRemoteWorkspace({ ...input, target, boundary: `repository:${input.repository.id}:${result.pinnedCommit}`, - assertOwned: input.assertOwned ?? (async () => undefined) }); - return result; + const descriptor = { ...input.repository, pinnedCommit: result.pinnedCommit }; + const requiresSave = !isSavedWorkspaceRepository(input.workspace, authority, descriptor); + if (requiresSave && !input.deferSave) { + const receipt = object(input.workspace.metadata?.remoteWorkspaceSnapshotRepositories); + const prior = receipt.snapshotId === authority.snapshotId && Array.isArray(receipt.repositories) + ? remoteWorkspaceSourceSchema.parse({ kind: "empty", repositories: receipt.repositories }).repositories ?? [] + : authority.source.repositories ?? []; + await saveRemoteWorkspace({ ...input, target, boundary: `repository:${input.repository.id}:${result.pinnedCommit}`, + repositories: [...prior.filter(repository => repository.id !== descriptor.id), descriptor], + assertOwned: input.assertOwned ?? (async () => undefined) }); + } + return { ...result, requiresSave }; } type SavedWorkspaceFileLimits = {