diff --git a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md index 6a4affe8a90..64cc24c6794 100644 --- a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md +++ b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md @@ -186,3 +186,34 @@ For a native hello probe, select a saved AI account or use the installed account Core rejects local environment overrides when no saved account is selected. This includes draft credentials. The probe uses the installed VM environment. Core does not test a different installed account in their place. Only Core can supply the selected account identity and local credential home. + +## Fresh attempts + +A workspace provider opts in with `workspaceProvider.freshWorkspaceVersion: 1`. +A task environment also declares `environmentDrivers[].freshWorkspaceVersion: 1` +and returns `freshWorkspaceVersion: 1` from `describe`. + +Core supplies a server-authorized `fresh` intent on workspace preparation and on +launch-related task calls. `attemptId` is stable across interrupted preparation. +`previousWorkspaceId` identifies retained storage; it is not permission to change +or remove it. The provider prepares a new working directory from the authorized +source and base. It may share repository objects or other immutable storage. +Core does not require a particular Git or filesystem layout. + +`recoverySnapshots` contains verified, Core-staged archive and manifest files. +Each file descriptor contains a trusted host path, SHA-256, and byte size. These +paths are host-authored transfer sources, never user-supplied paths or URLs. +Providers verify the bytes and safely materialize each snapshot separately at +`.paperclip-recovered/` beneath the new working directory. Preserve +file links safely and include provenance. Never overlay recovered files onto the +clean working tree or overwrite an earlier attachment. Core includes those +relative attachment paths in the new attempt's continuation context. + +A provider reconciles the same attempt and workspace identity after an ambiguous +reply. It must not replace the failed workspace or allocate another workspace +when preparation is repeated. Providers without this version remain unavailable +for fresh attempts; ordinary preparation and reuse retain their existing contract. + +Core keeps transfer files while preparation is uncertain. Confirmed task completion +and provider workspace removal release these disposable copies. The descriptors +remain stable: a later preparation downloads the same verified files again. diff --git a/packages/plugins/sdk/src/environment-tasks.ts b/packages/plugins/sdk/src/environment-tasks.ts index 60ec0134da0..97c72d866a5 100644 --- a/packages/plugins/sdk/src/environment-tasks.ts +++ b/packages/plugins/sdk/src/environment-tasks.ts @@ -108,6 +108,7 @@ export const environmentTaskResultSchema = z.discriminatedUnion("kind", [ z.object({ kind: z.literal("description"), taskId: providerId, runner: z.object({ version: z.string().min(1), digest: z.string().regex(/^sha256:[a-f0-9]{64}$/) }).strict(), launchPreparationVersion: z.literal(1).optional(), + freshWorkspaceVersion: z.literal(1).optional(), projectAuthCheckVersion: z.literal(1).optional(), }).strict(), z.object({ @@ -150,6 +151,7 @@ export interface PluginEnvironmentTaskParams extends Omit; + const sha = z.string().regex(/^[a-f0-9]{40,64}$/); export const workspaceProviderOperationSchema = z.discriminatedUnion("kind", [ z @@ -25,6 +41,7 @@ export const workspaceProviderOperationSchema = z.discriminatedUnion("kind", [ branchMode: z.enum(["create", "existing"]).optional(), baseRef: z.string().nullable(), existing: workspaceProviderBindingSchema.nullable(), + fresh: freshWorkspaceSchema.optional(), }) .strict(), z diff --git a/packages/plugins/sdk/tests/workspace-provider.test.ts b/packages/plugins/sdk/tests/workspace-provider.test.ts index f64aa7ece68..5f79260f6dd 100644 --- a/packages/plugins/sdk/tests/workspace-provider.test.ts +++ b/packages/plugins/sdk/tests/workspace-provider.test.ts @@ -85,6 +85,15 @@ describe("workspace provider SDK", () => { }), ).toThrow(); }); + it("expresses fresh outcomes without imposing Git storage layout", () => { + const fresh = { attemptId: id(6), sourceRunId: id(7), previousWorkspaceId: id(8), + recoverySnapshots: [{ operationId: id(9), sourceRunId: id(7), + archive: { path: "/controller/recovery/archive.tar", sha256: "a".repeat(64), byteSize: 1024 }, + manifest: { path: "/controller/recovery/manifest.jsonl", sha256: "b".repeat(64), byteSize: 128 } }] }; + expect(workspaceProviderOperationSchema.parse({ ...params.operation, fresh })).toMatchObject({ fresh }); + expect(() => workspaceProviderOperationSchema.parse({ ...params.operation, fresh: { ...fresh, independentGitStorage: true } })).toThrow(); + expect(() => workspaceProviderOperationSchema.parse({ ...params.operation, fresh: { ...fresh, recoverySnapshots: [{ ...fresh.recoverySnapshots[0], archive: { ...fresh.recoverySnapshots[0].archive, sha256: "invalid" } }] } })).toThrow(); + }); it("requires project lifecycle and storage management as one contract", () => { const manifest = { id: "test.storage", diff --git a/packages/shared/src/types/plugin.ts b/packages/shared/src/types/plugin.ts index 3f121a74f45..89b70b6519b 100644 --- a/packages/shared/src/types/plugin.ts +++ b/packages/shared/src/types/plugin.ts @@ -193,6 +193,7 @@ export interface SandboxProviderCapabilities { export interface PluginEnvironmentDriverDeclaration { /** Provides Runner execution, resource preparation, and PRP connections via environmentTask. */ supportsTasks?: boolean; + freshWorkspaceVersion?: 1; /** Stable driver key, unique within the plugin. Namespaced by plugin ID at runtime. */ driverKey: string; /** @@ -696,6 +697,7 @@ export interface PaperclipPluginManifestV1 { agentLifecycle?: true; /** Own project storage and durable workspace operations as one contract. */ workspaceProvider?: { + freshWorkspaceVersion?: 1; modes: Array<"shared_workspace" | "isolated_workspace" | "operator_branch">; strategies: Array<"project_primary" | "git_worktree">; defaultForProjects?: { diff --git a/packages/shared/src/validators/plugin.ts b/packages/shared/src/validators/plugin.ts index 7e333572d83..70a688978dc 100644 --- a/packages/shared/src/validators/plugin.ts +++ b/packages/shared/src/validators/plugin.ts @@ -178,6 +178,7 @@ export type SandboxProviderCapabilitiesInput = z.infer ({ plugin: {} as any })); +const state = vi.hoisted(() => ({ plugin: {} as any, release: vi.fn() })); +vi.mock("../services/provider-fresh-workspace.js", () => ({ + stageProviderFreshWorkspace: async (input: { existing: unknown }) => input.existing, + releaseProviderRecoveryTransfers: state.release, +})); vi.mock("../services/plugin-registry.js", () => ({ pluginRegistryService: () => ({ getById: vi.fn(async () => state.plugin) }) })); const leaseId = "10000000-0000-4000-8000-000000000001"; const projectId = "10000000-0000-4000-8000-000000000002"; @@ -30,9 +34,36 @@ function worker(result: unknown = { kind: "accepted", taskId: "attempt-1" }) { const submit = { kind: "submit" as const, preparedLaunchId: "configuration", projectIds: [projectId], runner: { protocolMin: 1, protocolMax: 2, harness: "codex", runnerId: "runner", leaseId, runId: "run", sessionId: "session", turnId: "turn", itemId: "item" }, bootstrapTicket: "transient-test-ticket" }; beforeEach(() => { + state.release.mockReset(); state.plugin = { id: "original", pluginKey: "test.provider", status: "ready", manifestJson: { capabilities: ["environment.drivers.register"], environmentDrivers: [{ driverKey: "tasks", supportsTasks: true }] } }; }); describe("plugin-provided Runner execution", () => { + it("forwards a retained fresh directory intent on later launch calls and checks provider support", async () => { + const value = row(); + value.lease.metadata.runnerTaskProjectAuthValidation.projectIds = []; + const fresh = { attemptId: leaseId, sourceRunId: secondProjectId, previousWorkspaceId: null, recoverySnapshots: [] }; + Object.assign(value.lease.metadata, { freshWorkspace: fresh }); + Object.assign(state.plugin.manifestJson.environmentDrivers[0], { freshWorkspaceVersion: 1 }); + const workers = worker(); + await executeEnvironmentTask(database(value, []).db, workers as never, { companyId: "company", leaseId, operation: { ...submit, projectIds: [] } }); + expect(workers.call).toHaveBeenCalledWith("original", "environmentTask", expect.objectContaining({ fresh }), 15_000); + delete state.plugin.manifestJson.environmentDrivers[0].freshWorkspaceVersion; + const denied = worker(); + await expect(executeEnvironmentTask(database(value, []).db, denied as never, { companyId: "company", leaseId, + operation: { ...submit, projectIds: [] } })).rejects.toThrow("fresh workspace preparation unavailable"); + expect(denied.call).not.toHaveBeenCalled(); + }); + it("releases fresh transfer copies only after confirmed task completion", async () => { + const value = row(); + const fresh = { attemptId: leaseId, sourceRunId: secondProjectId, previousWorkspaceId: null, recoverySnapshots: [] }; + Object.assign(value.lease.metadata, { freshWorkspace: fresh }); + const workers = worker(); + await executeEnvironmentTask(database(value).db, workers as never, { companyId: "company", leaseId, operation: { kind: "complete" } }); + expect(state.release).not.toHaveBeenCalled(); + Object.assign(value.lease.metadata, { runnerTaskExecutionStopped: true }); + await executeEnvironmentTask(database(value).db, workers as never, { companyId: "company", leaseId, operation: { kind: "complete" } }); + expect(state.release).toHaveBeenCalledWith("company", fresh); + }); it("passes only the workspace binding pinned to this lease and provider", async () => { const value = row(); value.lease.metadata.runnerTaskProjectAuthValidation.projectIds = []; const binding = { pluginId: leaseId, projectId, executionWorkspaceId: secondProjectId, projectWorkspaceId: projectId, diff --git a/server/src/__tests__/plugin-fresh-workspaces.integration.test.ts b/server/src/__tests__/plugin-fresh-workspaces.integration.test.ts new file mode 100644 index 00000000000..5a945bb7e50 --- /dev/null +++ b/server/src/__tests__/plugin-fresh-workspaces.integration.test.ts @@ -0,0 +1,124 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { createDb, executionWorkspaces, heartbeatRuns, issues, plugins } from "@paperclipai/db"; +import { preparePluginWorkspace, removePluginWorkspace } from "../services/plugin-workspaces.js"; +import { executionWorkspaceService } from "../services/execution-workspaces.js"; +import { reserveFreshPluginTaskWorkspace } from "../services/provider-fresh-workspace.js"; +import { createPluginWorkspaceFixture } from "./helpers/plugin-workspaces.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const cache = vi.hoisted(() => ({ release: vi.fn() })); +vi.mock("../services/provider-fresh-workspace.js", async importOriginal => ({ + ...await importOriginal(), + releaseProviderRecoveryTransfers: cache.release, +})); +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("provider-owned fresh retry reservations", () => { + let db: ReturnType, stopDb: (() => Promise) | undefined; + beforeAll(async () => { + const started = await startEmbeddedPostgresTestDatabase("plugin-fresh-workspaces"); + db = createDb(started.connectionString); stopDb = started.stop; + }); + afterAll(async () => { await db?.$client.end(); await stopDb?.(); }); + beforeEach(() => { cache.release.mockReset().mockResolvedValue(undefined); }); + async function fixture() { + const f = await createPluginWorkspaceFixture(db); + const previous = await preparePluginWorkspace(db, f.workers, f.input); + await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.runId)); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.input.agentId, + issueId: f.issueId, status: "running" }).returning(); + const fresh = { attemptId: randomUUID(), sourceRunId: f.runId, previousWorkspaceId: previous!.id, recoverySnapshots: [] }; + const input = { ...f.input, runId: run.id, existingWorkspaceId: previous!.id, fresh, expectedBindingRevision: 1 }; + return { ...f, previous: previous!, fresh, input }; + } + const enable = async (id: string) => db.update(plugins).set({ manifestJson: + sql`jsonb_set(${plugins.manifestJson}, '{workspaceProvider,freshWorkspaceVersion}', '1')` }).where(eq(plugins.id, id)); + it("fails closed for an older provider without replacing the old task binding", async () => { + const f = await fixture(); + await expect(preparePluginWorkspace(db, f.workers, f.input)).rejects.toThrow("fresh attempt is unavailable"); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(task.executionWorkspaceId).toBe(f.previous.id); + expect(f.calls).toHaveLength(1); + }); + it("reserves a new binding once and reconciles the same operation after a lost provider reply", async () => { + const f = await fixture(); await enable(f.pluginId); + f.lose(); + await expect(preparePluginWorkspace(db, f.workers, f.input)).rejects.toThrow("incomplete"); + const ambiguous = f.calls.at(-1); + const workspace = await preparePluginWorkspace(db, f.workers, f.input); + expect(workspace!.id).not.toBe(f.previous.id); + expect(f.calls.at(-1).executionWorkspaceId).toBe(ambiguous.executionWorkspaceId); + expect(f.calls.at(-1).operationId).toBe(ambiguous.operationId); + expect(f.calls.at(-1).operation).toMatchObject({ fresh: f.fresh, branchMode: "create", branchName: null, existing: null }); + const [retained] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.previous.id)); + expect(retained.providerRef).toBe(f.previous.providerRef); + expect(retained.metadata).toEqual(f.previous.metadata); + const rows = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.companyId, f.companyId)); + expect(rows).toHaveLength(2); + }); + it("keeps fresh directory authority server-owned through metadata updates", async () => { + const f = await fixture(); await enable(f.pluginId); + const service = executionWorkspaceService(db); + const original = await service.update(f.previous.id, { metadata: { + note: "ordinary metadata", workspaceProviderFresh: f.fresh, runnerTaskFreshWorkspace: f.fresh, + runnerTaskWorkspaceOwner: { pluginKey: "forged" }, + } }); + expect(original!.metadata).toMatchObject({ note: "ordinary metadata" }); + for (const key of ["workspaceProviderFresh", "runnerTaskFreshWorkspace", "runnerTaskWorkspaceOwner"]) { + expect(original!.metadata?.[key]).toBeUndefined(); + } + const prepared = await preparePluginWorkspace(db, f.workers, f.input); + const updated = await service.update(prepared!.id, { metadata: { note: "caller edit", workspaceProviderFresh: null } }); + expect(updated!.metadata).toMatchObject({ note: "caller edit", workspaceProviderFresh: f.fresh }); + await db.update(issues).set({ projectId: null, projectWorkspaceId: null }).where(eq(issues.id, f.issueId)); + const owner = { agentId: f.input.agentId, environmentId: randomUUID(), pluginKey: "test.tasks", driverKey: "tasks" }; + const next = { ...f.fresh, attemptId: randomUUID(), previousWorkspaceId: prepared!.id }; + const id = await reserveFreshPluginTaskWorkspace({ db, companyId: f.companyId, issueId: f.issueId, + runId: f.input.runId, agentId: f.input.agentId, responsibleUserId: null, fresh: next, owner, + expectedBindingRevision: 2 }); + const directory = await service.update(id, { metadata: { runnerTaskFreshWorkspace: f.fresh, + runnerTaskWorkspaceOwner: { pluginKey: "forged" } } }); + expect(directory!.metadata).toMatchObject({ runnerTaskFreshWorkspace: next, runnerTaskWorkspaceOwner: owner }); + const cleared = await service.update(id, { metadata: null }); + expect(cleared!.metadata).toMatchObject({ runnerTaskFreshWorkspace: next, runnerTaskWorkspaceOwner: owner }); + }); + it("retains the durable removal intent if transfer cache cleanup fails", async () => { + const f = await fixture(); await enable(f.pluginId); + const workspace = await preparePluginWorkspace(db, f.workers, f.input); + const [closed] = await db.update(executionWorkspaces).set({ status: "archived" }) + .where(eq(executionWorkspaces.id, workspace!.id)).returning(); + cache.release.mockRejectedValueOnce(new Error("temporary cache cleanup failure")); + await expect(removePluginWorkspace(db, f.workers, closed, "a".repeat(40))).rejects.toThrow("cache cleanup failure"); + const [pending] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id)); + expect(pending.metadata?.workspaceProviderRemoval).toBeTruthy(); + expect(pending.metadata?.workspaceProviderRemoved).not.toBe(true); + await removePluginWorkspace(db, f.workers, pending, "a".repeat(40)); + const [settled] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id)); + expect(settled.metadata?.workspaceProviderRemoved).toBe(true); + expect(cache.release).toHaveBeenCalledTimes(2); + }); + it("rejects a changed task revision before calling the provider", async () => { + const f = await fixture(); await enable(f.pluginId); + await db.update(issues).set({ workspaceBindingRevision: 2 }).where(eq(issues.id, f.issueId)); + await expect(preparePluginWorkspace(db, f.workers, f.input)).rejects.toThrow("binding changed"); + expect(f.calls).toHaveLength(1); + }); + it("keeps one projectless attempt directory identity across reconciliation and later reuse", async () => { + const f = await fixture(); + await db.update(issues).set({ projectId: null, projectWorkspaceId: null }).where(eq(issues.id, f.issueId)); + const owner = { agentId: f.input.agentId, environmentId: randomUUID(), pluginKey: "test.tasks", driverKey: "tasks" }; + const input = { db, companyId: f.companyId, issueId: f.issueId, runId: f.input.runId, + agentId: f.input.agentId, responsibleUserId: null, fresh: f.fresh, owner, expectedBindingRevision: 1 }; + const id = await reserveFreshPluginTaskWorkspace(input); + expect(await reserveFreshPluginTaskWorkspace(input)).toBe(id); + const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, id)); + expect(workspace.metadata).toMatchObject({ runnerTaskFreshWorkspace: f.fresh, runnerTaskWorkspaceOwner: owner }); + expect(workspace.cwd).toBe(`/workspaces/${f.fresh.attemptId}`); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(task.executionWorkspaceId).toBe(id); + expect(task.workspaceBindingRevision).toBe(2); + await db.update(heartbeatRuns).set({ responsibleUserId: "changed-user" }).where(eq(heartbeatRuns.id, f.input.runId)); + await expect(reserveFreshPluginTaskWorkspace(input)).rejects.toThrow("responsible user changed"); + }); +}); diff --git a/server/src/services/environment-run-orchestrator.ts b/server/src/services/environment-run-orchestrator.ts index 8757efe6cf9..287b4a9fd74 100644 --- a/server/src/services/environment-run-orchestrator.ts +++ b/server/src/services/environment-run-orchestrator.ts @@ -1,3 +1,4 @@ +import { freshWorkspaceSchema } from "@paperclipai/plugin-sdk"; /** * Centralized environment run orchestrator. * @@ -429,6 +430,13 @@ export function environmentRunOrchestrator( if (!updated) throw new Error("Workspace execution lease disappeared"); lease = updated; } + const freshWorkspace = freshWorkspaceSchema.safeParse(persistedExecutionWorkspace?.metadata?.workspaceProviderFresh ?? + persistedExecutionWorkspace?.metadata?.runnerTaskFreshWorkspace); + if (environment.driver === "plugin" && freshWorkspace.success) { + const updated = await environmentsSvc.updateLeaseMetadata(lease.id, { ...lease.metadata, freshWorkspace: freshWorkspace.data }); + if (!updated) throw new Error("Workspace execution lease disappeared"); + lease = updated; + } // Step 1: Build workspace realization request const workspaceRealizationRequest = buildWorkspaceRealizationRequest({ adapterType, diff --git a/server/src/services/environment-task-runtime.ts b/server/src/services/environment-task-runtime.ts index 78e471ad661..f82d73feb5e 100644 --- a/server/src/services/environment-task-runtime.ts +++ b/server/src/services/environment-task-runtime.ts @@ -1,8 +1,9 @@ +import { releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js"; import { readWorkspaceProviderBinding } from "@paperclipai/shared"; import { settlePluginTaskCredentialHandoff } from "./plugin-task-credential-writeback.js"; import { and, eq, inArray, or, sql } from "drizzle-orm"; import { environmentLeases, executionWorkspaces, environments, heartbeatRuns, issues, projects, type Db } from "@paperclipai/db"; -import { environmentTaskOperationSchema, parseEnvironmentTaskResult, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; +import { freshWorkspaceSchema, environmentTaskOperationSchema, parseEnvironmentTaskResult, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; import { pluginRegistryService } from "./plugin-registry.js"; import { hasRemoteTerminationReceipt, remoteExecutionHasStopped } from "./remote-execution-termination.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; @@ -105,6 +106,14 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage throw new Error("Managed workspace execution binding changed"); } } + let freshWorkspace = metadata.freshWorkspace === undefined ? undefined : freshWorkspaceSchema.parse(metadata.freshWorkspace); + if (freshWorkspace && driver.freshWorkspaceVersion !== 1 && (launches || operation.kind === "describe" || operation.kind === "check_project_auth")) + throw new Error("Plugin-provided fresh workspace preparation unavailable"); + if (freshWorkspace && operation.kind === "prepare") { + if (!lease.issueId) throw new Error("Fresh workspace task ownership is missing"); + freshWorkspace = await stageProviderFreshWorkspace({ db, companyId: input.companyId, + issueId: lease.issueId, ...freshWorkspace, existing: freshWorkspace }); + } const projectIds = scopesProjects ? operation.projectIds : []; if (activeProbe && projectIds.length > 0) throw new Error("Plugin Runner probe cannot use project mounts"); if (projectIds.length > 0) { @@ -164,9 +173,12 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage issueId: lease.issueId, config: driverConfig, taskId, runId: run?.id ?? (activeProbe ? String(probe.runId) : null), agentId: run?.agentId ?? null, projectIds, lease: { providerLeaseId: lease.providerLeaseId, metadata: lease.metadata ?? undefined }, - ...(workspaceBinding ? { workspace: workspaceBinding } : {}), operation, + ...(workspaceBinding ? { workspace: workspaceBinding } : {}), + ...(freshWorkspace && (launches || operation.kind === "check_project_auth") ? { fresh: freshWorkspace } : {}), operation, }, operation.kind === "copilot_metadata" ? Math.max(1, Math.min(40_000, Number(probe?.expiresAtMs) - Date.now())) : 15_000); const parsed = parseEnvironmentTaskResult(operation, taskId, result); + if (freshWorkspace && parsed.kind === "description" && parsed.freshWorkspaceVersion !== 1) + throw new Error("Plugin-provided fresh workspace preparation unavailable"); if (operation.kind === "check_project_auth" && parsed.kind === "project_auth") { if (parsed.configurationId !== operation.configurationId || parsed.provider !== operation.provider) throw new Error("Plugin Runner project validation identity mismatch"); if (parsed.allowed) { @@ -183,6 +195,8 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, input.companyId))); } if (operation.kind === "complete") { + if (freshWorkspace && metadata.runnerTaskExecutionStopped === true) + await releaseProviderRecoveryTransfers(input.companyId, freshWorkspace); await db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"runnerTaskDispatched":false}'::jsonb`, }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, input.companyId), diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 1065fcc1362..4c0eb887a41 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -259,6 +259,14 @@ function preserveWorkspaceMetadataForPatch( } } } + const freshIntentKeys = ["workspaceProviderFresh", "runnerTaskFreshWorkspace", "runnerTaskWorkspaceOwner"] as const; + if (patch.metadata !== undefined && (patch.metadata !== null || freshIntentKeys.some(key => current.metadata?.[key] !== undefined))) { + nextPatch.metadata = { ...(nextPatch.metadata ?? patch.metadata) }; + for (const key of freshIntentKeys) { + if (current.metadata?.[key] !== undefined) nextPatch.metadata[key] = current.metadata[key]; + else delete nextPatch.metadata[key]; + } + } const retainedSources = { ...parseObject(parseObject(patch.metadata)._issuePrivacySources), ...parseObject(current.metadata?._issuePrivacySources), diff --git a/server/src/services/heartbeat/workspace-preparation.test.ts b/server/src/services/heartbeat/workspace-preparation.test.ts index cc4d2e16dc7..b51d8713be5 100644 --- a/server/src/services/heartbeat/workspace-preparation.test.ts +++ b/server/src/services/heartbeat/workspace-preparation.test.ts @@ -97,14 +97,16 @@ describe.skipIf(!support.supported)("heartbeat workspace preparation boundary", return { input, source, issue, run, envOrchestrator }; } it.each([ - { inherited: false, mode: "isolated_workspace" as const }, - { inherited: true, mode: "isolated_workspace" as const }, - { inherited: true, mode: "operator_branch" as const }, - ])("prepares managed storage before acquisition (inherited: $inherited, mode: $mode) without touching a Core checkout", async ({ inherited, mode: defaultMode }) => { + { inherited: false, mode: "isolated_workspace" as const, fresh: false }, + { inherited: false, mode: "isolated_workspace" as const, fresh: true }, + { inherited: true, mode: "isolated_workspace" as const, fresh: false }, + { inherited: true, mode: "operator_branch" as const, fresh: false }, + ])("prepares managed storage before acquisition (inherited: $inherited, mode: $mode, fresh: $fresh) without touching a Core checkout", async ({ inherited, mode: defaultMode, fresh }) => { const { input, source, issue } = await fixture(true); - const [plugin] = await db.insert(plugins).values({ pluginKey: "test.managed-storage", packageName: "test-storage", version: "1.0.0", apiVersion: 1, status: "ready", categories: ["workspace"], - manifestJson: { id: "test.managed-storage", apiVersion: 1, version: "1.0.0", displayName: "Storage", description: "Storage", author: "Test", categories: ["workspace"], - workspaceProvider: { modes: ["isolated_workspace", "operator_branch"], strategies: ["git_worktree"], ...(inherited ? { defaultForProjects: { mode: defaultMode, strategy: "git_worktree" as const } } : {}) }, capabilities: ["workspaces.manage", "events.subscribe"], resourceLifecycle: ["project"], entrypoints: { worker: "./worker.js" } } }).returning(); + const pluginKey = `test.managed-storage-${inherited}-${defaultMode}-${fresh}`; + const [plugin] = await db.insert(plugins).values({ pluginKey, packageName: "test-storage", version: "1.0.0", apiVersion: 1, status: "ready", categories: ["workspace"], + manifestJson: { id: pluginKey, apiVersion: 1, version: "1.0.0", displayName: "Storage", description: "Storage", author: "Test", categories: ["workspace"], + workspaceProvider: { freshWorkspaceVersion: 1, modes: ["isolated_workspace", "operator_branch"], strategies: ["git_worktree"], ...(inherited ? { defaultForProjects: { mode: defaultMode, strategy: "git_worktree" as const } } : {}) }, capabilities: ["workspaces.manage", "events.subscribe"], resourceLifecycle: ["project"], entrypoints: { worker: "./worker.js" } } }).returning(); const explicitPolicy = { enabled: true, storageProviderPluginId: plugin.id, defaultMode: "isolated_workspace" as const, workspaceStrategy: { type: "git_worktree" as const, baseRef: "main", branchTemplate: "test-{{issue.identifier}}" } }; if (!inherited) await db.update(projects).set({ executionWorkspacePolicy: explicitPolicy }).where(eq(projects.id, issue.projectId!)); @@ -118,6 +120,7 @@ describe.skipIf(!support.supported)("heartbeat workspace preparation boundary", } await db.update(projectWorkspaces).set({ cwd: null, sourceType: "remote_managed", repoUrl: "https://github.com/example/repo.git" }).where(eq(projectWorkspaces.id, issue.projectWorkspaceId!)); await rm(source, { recursive: true }); + if (fresh) input.task.freshWorkspace = { mode: "fresh_isolated_workspace", retryIntentId: input.run.id, sourceRunId: input.run.id, retainedSourceIds: [] }; const sequence: string[] = []; input.services.workspaceWorkers = { getWorker: () => ({ supportedMethods: ["workspaceOperation"] }), call: async (_: string, __: string, params: any) => { sequence.push("prepare"); @@ -139,7 +142,7 @@ describe.skipIf(!support.supported)("heartbeat workspace preparation boundary", await expect(access(source)).rejects.toThrow(); const [task] = await db.select().from(issues).where(eq(issues.id, issue.id)); const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, task.executionWorkspaceId!)); - expect(workspace).toMatchObject({ providerType: "plugin", cwd: null, mode: defaultMode, strategyType: "git_worktree", branchName: inherited ? "WSP-1-prepare-this-workspace" : "test-WSP-1" }); + expect(workspace).toMatchObject({ providerType: "plugin", cwd: null, mode: defaultMode, strategyType: "git_worktree", branchName: fresh ? null : inherited ? "WSP-1-prepare-this-workspace" : "test-WSP-1" }); }); async function installDefault() { const [plugin] = await db.insert(plugins).values({ pluginKey: "test.default-storage", packageName: "test-storage", version: "1.0.0", status: "ready", @@ -174,11 +177,53 @@ describe.skipIf(!support.supported)("heartbeat workspace preparation boundary", expect(call).not.toHaveBeenCalled(); }); + it.each([ + { project: false, savedSource: false }, + { project: true, savedSource: true }, + { project: true, savedSource: false }, + ])("keeps a fresh task reservation stable through repeated preparation (project: $project, saved source: $savedSource)", async ({ project, savedSource }) => { + const { input, issue, envOrchestrator } = await fixture(); + if (!project) await db.update(issues).set({ projectId: null, projectWorkspaceId: null }).where(eq(issues.id, issue.id)); + else if (!savedSource) { + await db.update(issues).set({ projectWorkspaceId: null }).where(eq(issues.id, issue.id)); + input.task.issueRef = { ...input.task.issueRef!, projectWorkspaceId: null }; + } + const pluginKey = "test.fresh-task"; + await db.insert(plugins).values({ pluginKey, packageName: "test-tasks", version: "1.0.0", apiVersion: 1, + status: "ready", categories: ["workspace"], manifestJson: { id: pluginKey, apiVersion: 1, version: "1.0.0", + displayName: "Tasks", description: "Tasks", author: "Test", categories: ["workspace"], + capabilities: ["environment.drivers.register"], entrypoints: { worker: "./worker.js" }, + environmentDrivers: [{ driverKey: "tasks", displayName: "Task VM", configSchema: { type: "object" }, supportsTasks: true, freshWorkspaceVersion: 1 }] } }); + input.agent.adapterType = "paperclip_runner"; + if (!project) { + input.task.issueRef = { ...input.task.issueRef!, projectId: null, projectWorkspaceId: null }; + input.task.context = { issueId: issue.id }; + } + input.task.freshWorkspace = { mode: "fresh_isolated_workspace", retryIntentId: input.run.id, sourceRunId: input.run.id, retainedSourceIds: [] }; + input.policy.requestedExecutionWorkspaceMode = input.policy.effectiveExecutionWorkspaceMode = "isolated_workspace"; + input.config.mergedConfig = { workspaceStrategy: { type: "project_primary" } }; + input.environment.selectedEnvironmentForConfig = { ...input.environment.localEnvironment, driver: "plugin", + config: { pluginKey, driverKey: "tasks" } }; + const stopped = new Error("Reached projectless environment handoff"); + const acquire = vi.spyOn(envOrchestrator, "acquireForRun").mockRejectedValue(stopped); + const read = vi.spyOn(input.services.executionWorkspacesSvc, "getStoredById"); + for (let attempt = 0; attempt < 2; attempt++) await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(stopped); + expect(read).toHaveBeenCalledWith(input.agent.companyId, input.run.id); + expect(acquire).toHaveBeenCalledTimes(2); + for (const [args] of acquire.mock.calls) expect(args.persistedExecutionWorkspace).toMatchObject({ id: input.run.id, + providerType: "adapter_managed", projectId: project ? issue.projectId : null, + projectWorkspaceId: project ? issue.projectWorkspaceId : null, + metadata: { runnerTaskFreshWorkspace: { attemptId: input.run.id } } }); + const rows = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.companyId, input.agent.companyId)); + expect(rows).toHaveLength(1); + const [task] = await db.select().from(issues).where(eq(issues.id, issue.id)); + expect(task.projectWorkspaceId).toBe(project ? issue.projectWorkspaceId : null); + }); const readIssue = (id: string) => db.select().from(issues).where(eq(issues.id, id)).then(rows => rows[0]!); const readRun = (id: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]!); const leases = (id: string) => db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, id)); - it("keeps plugin-owned storage untouched when a fresh attempt cannot import preserved files", async () => { + it("keeps plugin-owned storage untouched when a fresh attempt provider is unavailable", async () => { const { input, source, issue } = await fixture(true); await db.update(projects).set({ executionWorkspacePolicy: { enabled: true, storageProviderPluginId: "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", @@ -188,9 +233,7 @@ describe.skipIf(!support.supported)("heartbeat workspace preparation boundary", sourceRunId: input.run.id, retainedSourceIds: [] }; const call = vi.fn(); input.services.workspaceWorkers = { getWorker: () => ({ supportedMethods: ["workspaceOperation"] }), call } as any; - await expect(prepareHeartbeatWorkspace(db, input)).rejects.toMatchObject({ - resultJson: { reason: "fresh_workspace_storage_unsupported" }, - }); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toThrow("workspace storage plugin is unavailable"); expect(call).not.toHaveBeenCalled(); expect(await leases(input.run.id)).toHaveLength(0); expect((await readIssue(issue.id)).executionWorkspaceId).toBeNull(); diff --git a/server/src/services/heartbeat/workspace-preparation.ts b/server/src/services/heartbeat/workspace-preparation.ts index 8b1cca96500..68ce0e2db5b 100644 --- a/server/src/services/heartbeat/workspace-preparation.ts +++ b/server/src/services/heartbeat/workspace-preparation.ts @@ -1,3 +1,6 @@ +import { reserveFreshPluginTaskWorkspace, stageProviderFreshWorkspace } from "../provider-fresh-workspace.js"; +import { freshWorkspaceSchema, type PluginFreshWorkspace } from "@paperclipai/plugin-sdk"; +import { pluginRegistryService } from "../plugin-registry.js"; import { attachPreservedWorkspaceSnapshots } from "../workspace-recovery-snapshot.js"; import { materializeFreshWorkspaceSeed } from "../fresh-workspace-seed.js"; import { assertTaskWorkspaceExecutionAccess } from "../task-workspace-access.js"; @@ -27,7 +30,7 @@ import path from "node:path"; import { CONTROLLER_PROVISION_COPY_ONLY, workspaceRequiresCopyOnlyProvisioning } from "../workspace-host-provision.js"; import { and, asc, eq } from "drizzle-orm"; import { type ExecutionWorkspace } from "@paperclipai/shared"; -import { heartbeatRuns, issues, projectWorkspaces } from "@paperclipai/db"; +import { executionWorkspaces as executionWorkspacesTable, heartbeatRuns, issues, projectWorkspaces } from "@paperclipai/db"; import { getStartupTracer } from "../../instrumentation.js"; import { createHostDuplexObservabilityRecorder } from "../duplex-observability-recorder.js"; import { incrementToolRuntimeMetricCounter } from "../tool-runtime-metrics.js"; @@ -140,7 +143,7 @@ export interface HeartbeatWorkspacePreparationInput { services: { workspaceWorkers?: WorkspaceWorkers; envOrchestrator: Pick, "resolveEnvironment" | "acquireForRun" | "realizeForRun">; - executionWorkspacesSvc: Pick, "create" | "update" | "bindTaskWorkspace" | "prepareTaskRepositoriesForAdmission" | "assertTaskRepositoryPublicationAllowed" | "prepareConfiguredRepositoryCopy" | "assertTaskRepositoryRecoveryBinding">; + executionWorkspacesSvc: Pick, "create" | "update" | "getStoredById" | "bindTaskWorkspace" | "prepareTaskRepositoriesForAdmission" | "assertTaskRepositoryPublicationAllowed" | "prepareConfiguredRepositoryCopy" | "assertTaskRepositoryRecoveryBinding">; workspaceOperationsSvc: Pick, "createRecorder">; resolveWorkspaceForRun: WorkspaceResolver["resolveWorkspaceForRun"]; resolveReusedGitWorkspaceAnchor: WorkspaceResolver["resolveReusedGitWorkspaceAnchor"]; @@ -216,11 +219,14 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace appendRunEvent, } = input.services; const pluginTaskWorkspace = selectedEnvironmentForConfig?.driver === "plugin" && agent.adapterType === "paperclip_runner"; + const pluginConfig = parseObject(selectedEnvironmentForConfig?.config); if (fresh && !isDotRun && pluginTaskWorkspace) { - throw new WorkspaceValidationFailure("A fresh attempt is unavailable for this workspace storage provider. The existing workspace is retained.", + const plugin = typeof pluginConfig.pluginKey === "string" ? await pluginRegistryService(db).getByKey(pluginConfig.pluginKey) : null; + const driver = plugin?.manifestJson.environmentDrivers?.find(value => value.driverKey === pluginConfig.driverKey); + if (driver?.freshWorkspaceVersion !== 1) throw new WorkspaceValidationFailure( + "A fresh attempt is unavailable for this workspace storage provider. The existing workspace is retained.", { reason: "fresh_workspace_storage_unsupported" }); } - const pluginConfig = parseObject(selectedEnvironmentForConfig?.config); const taskWorkspaceOwner = pluginTaskWorkspace ? { agentId: agent.id, environmentId: selectedEnvironmentForConfig.id, pluginKey: pluginConfig.pluginKey ?? null, driverKey: pluginConfig.driverKey ?? null } : null; for (const workspace of [existingExecutionWorkspace, reusableExistingExecutionWorkspace]) { @@ -375,20 +381,30 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace }, })); const hostProvisionDriver = selectedEnvironmentForConfig?.driver ?? lowTrustPreflightEnvironmentDriver; - if (fresh && !isDotRun) { - const sourceWorkspace = reusableExistingExecutionWorkspace ?? existingExecutionWorkspace; - if (sourceWorkspace?.providerType === "plugin" || - readWorkspaceProviderBinding(sourceWorkspace?.metadata) || + let providerFresh: PluginFreshWorkspace | undefined; + let previousProviderWorkspaceId: string | null = null; + let previousProviderBindingRevision = 0; + if (input.task.freshWorkspace?.mode === "fresh_isolated_workspace" && !isDotRun && issueId) { + const [task] = await db.select({ workspaceId: issues.executionWorkspaceId, revision: issues.workspaceBindingRevision }).from(issues) + .where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId))); + previousProviderWorkspaceId = task?.workspaceId ?? null; + previousProviderBindingRevision = task?.revision ?? 0; + const [previous] = previousProviderWorkspaceId ? await db.select().from(executionWorkspacesTable) + .where(and(eq(executionWorkspacesTable.id, previousProviderWorkspaceId), eq(executionWorkspacesTable.companyId, agent.companyId))) : []; + if (pluginTaskWorkspace || previous?.providerType === "plugin" || (resolvedWorkspace.workspaceId && projectExecutionWorkspacePolicy?.storageProviderPluginId)) { - throw new WorkspaceValidationFailure("A fresh attempt is unavailable for this workspace storage provider. The existing workspace is retained.", { - reason: "fresh_workspace_storage_unsupported", - }); + const saved = freshWorkspaceSchema.safeParse(previous?.metadata?.workspaceProviderFresh ?? previous?.metadata?.runnerTaskFreshWorkspace); + providerFresh = saved.success && saved.data.attemptId === input.task.freshWorkspace.retryIntentId ? saved.data : + await stageProviderFreshWorkspace({ db, companyId: agent.companyId, issueId, + attemptId: input.task.freshWorkspace.retryIntentId, sourceRunId: input.task.freshWorkspace.sourceRunId, + previousWorkspaceId: previousProviderWorkspaceId }); } } const pluginPrepared = isDotRun ? null : await preparePluginWorkspace(db, input.services.workspaceWorkers, { companyId: agent.companyId, projectId: resolvedWorkspace.projectId, projectWorkspaceId: resolvedWorkspace.workspaceId, issueId, runId: run.id, agentId: agent.id, responsibleUserId, - existingWorkspaceId: reusableExistingExecutionWorkspace?.id ?? issueRef?.executionWorkspaceId ?? null, + existingWorkspaceId: providerFresh ? providerFresh.previousWorkspaceId : reusableExistingExecutionWorkspace?.id ?? issueRef?.executionWorkspaceId ?? null, + ...(providerFresh ? { fresh: providerFresh, expectedBindingRevision: previousProviderBindingRevision } : {}), selectedProviderPluginId: projectExecutionWorkspacePolicy?.storageProviderPluginId ?? null, mode: requestedExecutionWorkspaceMode, strategy: resolveEffectiveWorkspaceStrategyType(requestedExecutionWorkspaceMode, mergedConfig), @@ -401,6 +417,12 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace baseRef: typeof parseObject(mergedConfig.workspaceStrategy).baseRef === "string" ? String(parseObject(mergedConfig.workspaceStrategy).baseRef) : null, }); + const freshTaskReservationId = providerFresh && pluginTaskWorkspace && !pluginPrepared && issueId + ? await reserveFreshPluginTaskWorkspace({ db, companyId: agent.companyId, issueId, runId: run.id, agentId: agent.id, + fresh: providerFresh, responsibleUserId, expectedBindingRevision: previousProviderBindingRevision, owner: taskWorkspaceOwner!, + source: { projectId: resolvedWorkspace.projectId, projectWorkspaceId: resolvedWorkspace.workspaceId, + repoUrl: resolvedWorkspace.repoUrl, baseRef: resolvedWorkspace.repoRef } }) : null; + const freshTaskReservation = freshTaskReservationId ? await executionWorkspacesSvc.getStoredById(agent.companyId, freshTaskReservationId) : null; const copyOnlyProvisioning = pluginPrepared ? true : await timeWorkspace("workspace_provision_policy", () => workspaceRequiresCopyOnlyProvisioning(db, { companyId: agent.companyId, task: issueRef ? { ...issueRef, workspaceSelection: input.task.workspaceSelection } : null, @@ -560,7 +582,7 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace agentId: agent.id, }, ); - if (freshNeedsSeed && issueId) { + if (freshNeedsSeed && issueId && !pluginPrepared && !pluginTaskWorkspace) { await controllerLease.assertOwned(); const verifySourceAccess = async () => { await assertTaskWorkspaceExecutionAccess(db, { @@ -591,7 +613,7 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace warnings: [], created: false, branchCreatedByRuntime: false, baseRefSha: pluginPrepared.metadata?.workspaceProviderBaseSha as string | null, } as RealizedExecutionWorkspace, reusedExecutionWorkspace: Boolean(reusableExistingExecutionWorkspace), policy: workspaceReuseProvisioningPolicy } - : isDotRun || pluginTaskWorkspace ? { executionWorkspace: { ...executionWorkspaceBase, strategy: pluginTaskWorkspace ? latestWorkspaceStrategyType : "project_primary" as const, cwd: resolvedWorkspace.cwd, branchName: null, worktreePath: null, warnings: [], created: false, branchCreatedByRuntime: false } as RealizedExecutionWorkspace, reusedExecutionWorkspace: pluginTaskWorkspace && workspaceReuseProvisioningPolicy.shouldRestoreExistingWorkspace && Boolean(reusableExistingExecutionWorkspace), policy: workspaceReuseProvisioningPolicy } : await timeWorkspace("workspace_provision", () => provisionExecutionWorkspaceForFreshnessDecision( + : isDotRun || pluginTaskWorkspace ? { executionWorkspace: { ...executionWorkspaceBase, strategy: pluginTaskWorkspace ? latestWorkspaceStrategyType : "project_primary" as const, cwd: freshTaskReservation?.cwd ?? resolvedWorkspace.cwd, branchName: null, worktreePath: null, warnings: [], created: false, branchCreatedByRuntime: false } as RealizedExecutionWorkspace, reusedExecutionWorkspace: pluginTaskWorkspace && workspaceReuseProvisioningPolicy.shouldRestoreExistingWorkspace && Boolean(reusableExistingExecutionWorkspace), policy: workspaceReuseProvisioningPolicy } : await timeWorkspace("workspace_provision", () => provisionExecutionWorkspaceForFreshnessDecision( { requestedShouldReuseExisting, existingExecutionWorkspaceId: @@ -696,9 +718,12 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace }, )); if (input.task.freshWorkspace?.mode === "fresh_isolated_workspace" && issueId) { - const snapshots = await attachPreservedWorkspaceSnapshots({ db, companyId: agent.companyId, - issueId, sourceRunIds: [input.task.freshWorkspace.sourceRunId], cwd: executionWorkspace.cwd }); - const recoveredFiles = snapshots.map(snapshot => ({ ...snapshot, path: path.relative(executionWorkspace.cwd, snapshot.path) })); + const recoveredFiles = providerFresh ? providerFresh.recoverySnapshots.map(snapshot => ({ + operationId: snapshot.operationId, sourceRunId: snapshot.sourceRunId, + path: `.paperclip-recovered/${snapshot.operationId}`, + })) : (await attachPreservedWorkspaceSnapshots({ db, companyId: agent.companyId, + issueId, sourceRunIds: [input.task.freshWorkspace.sourceRunId], cwd: executionWorkspace.cwd })) + .map(snapshot => ({ ...snapshot, path: path.relative(executionWorkspace.cwd, snapshot.path) })); context.recoveredWorkspaceFiles = recoveredFiles; const continuation = context.executionContinuation as import("@paperclipai/shared").ExecutionContinuationEnvelope | null; if (continuation) { @@ -841,6 +866,7 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace ...baseExecutionWorkspaceMetadata, ...(pluginPrepared?.metadata ?? {}), ...(taskWorkspaceOwner && !pluginPrepared ? { runnerTaskWorkspaceOwner: taskWorkspaceOwner } : {}), + ...(providerFresh && pluginTaskWorkspace && !pluginPrepared ? { runnerTaskFreshWorkspace: providerFresh } : {}), ...(copyOnlyProvisioning ? { [CONTROLLER_PROVISION_COPY_ONLY]: true } : {}), ...(persistedWorktreeInstanceRoot ? { @@ -857,6 +883,7 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace try { persistedExecutionWorkspace = pluginPrepared ? await executionWorkspacesSvc.update(pluginPrepared.id, { metadata: nextExecutionWorkspaceMetadata }) : + freshTaskReservation ? await executionWorkspacesSvc.update(freshTaskReservation.id, { metadata: nextExecutionWorkspaceMetadata }) : resolvedWorkspaceReusePolicy.shouldRestoreExistingWorkspace && reusableExistingExecutionWorkspace ? await timeWorkspace("workspace_record_update", () => executionWorkspacesSvc.update( diff --git a/server/src/services/plugin-workspaces.ts b/server/src/services/plugin-workspaces.ts index e519bd7123a..9e97724144c 100644 --- a/server/src/services/plugin-workspaces.ts +++ b/server/src/services/plugin-workspaces.ts @@ -1,4 +1,5 @@ import { effectiveProjectWorkspacePolicy, isWorkspaceProvider } from "./project-workspace-provider.js"; +import { releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js"; import { randomUUID, createHash } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { @@ -20,6 +21,7 @@ import { workspaceProviderOperationSchema, type PluginWorkspaceOperation, type PluginWorkspaceParams, + type PluginFreshWorkspace, } from "@paperclipai/plugin-sdk"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; @@ -85,13 +87,20 @@ export async function callWorkspaceProvider( const pluginId = binding?.pluginId ?? input.pluginId; if (!pluginId || !workspace.projectId) throw conflict("Workspace storage ownership is missing"); - await provider(db, workers, workspace.companyId, pluginId); + const selectedProvider = await provider(db, workers, workspace.companyId, pluginId); + if (input.operation.kind === "prepare" && input.operation.fresh && selectedProvider.manifestJson.workspaceProvider?.freshWorkspaceVersion !== 1) + throw conflict("Fresh workspace preparation is unavailable"); const generation = typeof workspace.metadata?.lifecycleGeneration === "number" ? workspace.metadata.lifecycleGeneration : 0; const operationGeneration = generation + (input.advanceGeneration ? 1 : 0); const operation = workspaceProviderOperationSchema.parse(input.operation); + if (operation.kind === "prepare" && operation.fresh) { + if (!workspace.sourceIssueId) throw conflict("Fresh workspace task ownership is missing"); + operation.fresh = await stageProviderFreshWorkspace({ db, companyId: workspace.companyId, + issueId: workspace.sourceIssueId, ...operation.fresh, existing: operation.fresh }); + } const params: PluginWorkspaceParams = { pluginId, companyId: workspace.companyId, @@ -167,9 +176,11 @@ export async function preparePluginWorkspace( branchMode?: "create" | "existing"; branchName?: string | null; baseRef?: string | null; + fresh?: PluginFreshWorkspace; + expectedBindingRevision?: number; }, ) { - const [existing] = input.existingWorkspaceId + const [previous] = input.existingWorkspaceId ? await db .select() .from(executionWorkspaces) @@ -180,6 +191,13 @@ export async function preparePluginWorkspace( ), ) : []; + const [reservation] = input.fresh ? await db.select().from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, input.companyId), + eq(executionWorkspaces.sourceIssueId, input.issueId!), + sql`${executionWorkspaces.metadata}#>>'{workspaceProviderFresh,attemptId}' = ${input.fresh.attemptId}`, + )).limit(1) : []; + const existing = input.fresh ? reservation : previous; + const prior = existing ?? previous; if ( existing?.metadata?.workspaceProviderRetired === true || existing?.metadata?.workspaceProviderRemoval @@ -187,9 +205,10 @@ export async function preparePluginWorkspace( throw conflict("Workspace storage is retired or awaiting cleanup"); if (!existing && (!input.projectId || !input.projectWorkspaceId)) return null; const saved = readWorkspaceProviderBinding(existing?.metadata); - const mode = existing?.mode ?? input.mode, + const priorBinding = readWorkspaceProviderBinding(prior?.metadata); + const mode = input.fresh ? "isolated_workspace" : existing?.mode ?? input.mode, strategy = existing?.strategyType ?? input.strategy; - const projectId = saved?.projectId ?? input.projectId; + const projectId = priorBinding?.projectId ?? input.projectId; const [project] = projectId ? await db .select() @@ -201,8 +220,8 @@ export async function preparePluginWorkspace( ), ) : []; - if (existing && existing.providerType !== "plugin" && !saved && !existing.metadata?.workspaceProviderOwner) return null; - const owner = saved?.pluginId ?? existing?.metadata?.workspaceProviderOwner ?? + if (prior && prior.providerType !== "plugin" && !priorBinding && !prior.metadata?.workspaceProviderOwner) return null; + const owner = priorBinding?.pluginId ?? prior?.metadata?.workspaceProviderOwner ?? (!existing && input.selectedProviderPluginId !== undefined ? input.selectedProviderPluginId : !existing && project ? (await effectiveProjectWorkspacePolicy(db, input.companyId, project.executionWorkspacePolicy, false))?.storageProviderPluginId : null); const pluginId = typeof owner === "string" ? owner : null; @@ -238,6 +257,8 @@ export async function preparePluginWorkspace( pluginId, ); const declaration = selectedProvider.manifestJson.workspaceProvider!; + if (input.fresh && declaration.freshWorkspaceVersion !== 1) + throw conflict("A fresh attempt is unavailable for this workspace storage provider. The existing workspace is retained.", { reason: "fresh_workspace_storage_unsupported" }); if ( !declaration.modes.includes(mode as "isolated_workspace") || !declaration.strategies.includes(strategy as "git_worktree") @@ -302,7 +323,7 @@ export async function preparePluginWorkspace( ), ); if (!source) throw conflict("Workspace repository source is unavailable"); - const sourceInput = (existing?.metadata?.workspaceProviderSource as + const sourceInput = (prior?.metadata?.workspaceProviderSource as | Extract["source"] | undefined) ?? { id: source.id, @@ -322,7 +343,7 @@ export async function preparePluginWorkspace( repoRef: existing?.baseRef ?? source.repoRef, defaultRef: source.defaultRef, }; - const branchName = existing + const branchName = input.fresh ? null : existing ? existing.branchName : (input.branchName ?? (strategy === "git_worktree" ? `workspace-${randomUUID()}` : null)); @@ -363,7 +384,7 @@ export async function preparePluginWorkspace( ), ) .for("update"); - if (!task || task.executionWorkspaceId) + if (!task || (input.fresh ? task.executionWorkspaceId !== input.fresh.previousWorkspaceId || task.workspaceBindingRevision !== input.expectedBindingRevision : Boolean(task.executionWorkspaceId))) throw conflict("Task workspace binding changed; retry preparation"); const id = randomUUID(); await notifyTerminalWorkspaceWork(tx, `workspace:${id}`); @@ -375,8 +396,8 @@ export async function preparePluginWorkspace( projectId: project.id, projectWorkspaceId: source.id, sourceIssueId: input.issueId, - mode: input.mode, - strategyType: input.strategy, + mode, + strategyType: strategy, name: branchName ?? source.name, providerType: "plugin", providerRef: null, @@ -388,7 +409,8 @@ export async function preparePluginWorkspace( metadata: { workspaceProviderOwner: pluginId, workspaceProviderSource: sourceInput, - workspaceProviderBranchMode: input.branchMode ?? "create", + workspaceProviderBranchMode: input.fresh ? "create" : input.branchMode ?? "create", + ...(input.fresh ? { workspaceProviderFresh: input.fresh } : {}), }, }) .returning(); @@ -437,7 +459,7 @@ export async function preparePluginWorkspace( source: sourceInput, mode: workspace.mode as "isolated_workspace", strategy: workspace.strategyType as "git_worktree", - branchName, + branchName: workspace.metadata?.workspaceProviderFresh ? null : branchName, branchMode: (workspace.metadata?.workspaceProviderBranchMode as | "create" @@ -447,6 +469,7 @@ export async function preparePluginWorkspace( "create", baseRef: workspace.baseRef, existing: saved, + ...(workspace.metadata?.workspaceProviderFresh ? { fresh: workspace.metadata.workspaceProviderFresh as PluginFreshWorkspace } : {}), }, }); if (result.kind !== "prepared") @@ -522,10 +545,11 @@ async function reconcilePluginWorkspaceReservation( source: workspace.metadata.workspaceProviderSource, mode: workspace.mode, strategy: workspace.strategyType, - branchName: workspace.branchName, + branchName: workspace.metadata.workspaceProviderFresh ? null : workspace.branchName, branchMode: workspace.metadata.workspaceProviderBranchMode ?? "create", baseRef: workspace.baseRef, existing: null, + ...(workspace.metadata?.workspaceProviderFresh ? { fresh: workspace.metadata.workspaceProviderFresh as PluginFreshWorkspace } : {}), }); const result = await callWorkspaceProvider(db, workers, { workspace, @@ -701,6 +725,8 @@ async function finishPendingRemoval( }); if (result.kind !== "removed") throw conflict("Workspace removal is incomplete"); + if (workspace.metadata?.workspaceProviderFresh) + await releaseProviderRecoveryTransfers(workspace.companyId, workspace.metadata.workspaceProviderFresh as PluginFreshWorkspace); const [updated] = await db.transaction(async (tx) => { await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`); return tx @@ -739,7 +765,11 @@ export async function removePluginWorkspace( workspace = await reconcilePluginWorkspaceReservation(db, workers, workspace); if (workspace.metadata?.workspaceProviderRestoration) throw conflict("Workspace restoration must be reconciled before removal"); - if (workspace.metadata?.workspaceProviderRemoved === true) return; + if (workspace.metadata?.workspaceProviderRemoved === true) { + if (workspace.metadata?.workspaceProviderFresh) + await releaseProviderRecoveryTransfers(workspace.companyId, workspace.metadata.workspaceProviderFresh as PluginFreshWorkspace); + return; + } if (workspace.metadata?.workspaceProviderRemoval) { await finishPendingRemoval(db, workers, workspace); return; @@ -855,10 +885,11 @@ export async function restorePluginWorkspace( source, mode: workspace.mode, strategy: workspace.strategyType, - branchName: workspace.branchName, - branchMode: "existing", + branchName: workspace.metadata?.workspaceProviderFresh ? null : workspace.branchName, + branchMode: workspace.metadata?.workspaceProviderFresh ? "create" : "existing", baseRef: workspace.baseRef, existing: binding, + ...(workspace.metadata?.workspaceProviderFresh ? { fresh: workspace.metadata.workspaceProviderFresh as PluginFreshWorkspace } : {}), }); const [pending] = await db.transaction(async (tx) => { await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`); diff --git a/server/src/services/provider-fresh-workspace.test.ts b/server/src/services/provider-fresh-workspace.test.ts new file mode 100644 index 00000000000..324f6b9542a --- /dev/null +++ b/server/src/services/provider-fresh-workspace.test.ts @@ -0,0 +1,41 @@ +import { createHash, randomUUID } from "node:crypto"; +import { Readable } from "node:stream"; +import { chmod, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { Db } from "@paperclipai/db"; +import { releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js"; +const state = vi.hoisted(() => ({ getObject: vi.fn() })); +vi.mock("../storage/index.js", () => ({ getStorageService: () => ({ getObject: state.getObject }) })); +let home: string | undefined; +afterEach(async () => { vi.unstubAllEnvs(); state.getObject.mockReset(); if (home) await rm(home, { recursive: true, force: true }); }); +describe("fresh provider recovery transfer staging", () => { + it("verifies immutable transfer files and reuses stable descriptors after interruption", async () => { + home = await mkdtemp(path.join(tmpdir(), "provider-recovery-")); vi.stubEnv("PAPERCLIP_HOME", home); + const companyId = randomUUID(), sourceRunId = randomUUID(), leaseId = randomUUID(), operationId = randomUUID(); + const archive = Buffer.from("verified archive"), manifest = Buffer.from("verified manifest"); + const stored = (bytes: Buffer, extension: string) => { const sha256 = createHash("sha256").update(bytes).digest("hex"); + return { objectKey: `${companyId}/workspace-recovery/${sourceRunId}/${leaseId}/${sha256}.${extension}`, sha256, byteSize: bytes.length }; }; + const metadata = { schema: "paperclip.workspace-recovery-snapshot.v1", recoveryActionId: randomUUID(), leaseId, + providerLeaseId: "retained-task", archive: stored(archive, "tar"), manifest: stored(manifest, "jsonl") }; + const db = { execute: async () => [{ id: sourceRunId }], select: () => ({ from: () => ({ where: async () => [{ id: operationId, metadata }] }) }) } as unknown as Db; + state.getObject.mockImplementation(async (_companyId, key) => ({ stream: Readable.from([key.endsWith(".tar") ? archive : manifest]) })); + const input = { db, companyId, issueId: randomUUID(), sourceRunId, attemptId: randomUUID(), previousWorkspaceId: randomUUID() }; + const first = await stageProviderFreshWorkspace(input); + expect(first.recoverySnapshots).toHaveLength(1); + expect(await readFile(first.recoverySnapshots[0].archive.path)).toEqual(archive); + expect(await stageProviderFreshWorkspace({ ...input, existing: first })).toEqual(first); + await rm(first.recoverySnapshots[0].archive.path); + expect(await stageProviderFreshWorkspace({ ...input, existing: first })).toEqual(first); + expect(state.getObject).toHaveBeenCalledTimes(3); + await releaseProviderRecoveryTransfers(companyId, first); + await expect(readFile(first.recoverySnapshots[0].archive.path)).rejects.toThrow(); + expect(await stageProviderFreshWorkspace({ ...input, existing: first })).toEqual(first); + expect(state.getObject).toHaveBeenCalledTimes(5); + await chmod(first.recoverySnapshots[0].archive.path, 0o600); + await writeFile(first.recoverySnapshots[0].archive.path, "corrupt"); + await expect(stageProviderFreshWorkspace(input)).rejects.toThrow(); + expect(state.getObject).toHaveBeenCalledTimes(5); + }); +}); diff --git a/server/src/services/provider-fresh-workspace.ts b/server/src/services/provider-fresh-workspace.ts new file mode 100644 index 00000000000..e3e0ecfca0e --- /dev/null +++ b/server/src/services/provider-fresh-workspace.ts @@ -0,0 +1,150 @@ +import { assertTaskWorkspaceExecutionAccess } from "./task-workspace-access.js"; +import { isDeepStrictEqual } from "node:util"; +import { createHash, randomUUID } from "node:crypto"; +import { createReadStream, createWriteStream } from "node:fs"; +import { lstat, mkdir, realpath, rename, rm } from "node:fs/promises"; +import path from "node:path"; +import { pipeline } from "node:stream/promises"; +import { Transform } from "node:stream"; +import { and, asc, eq } from "drizzle-orm"; +import { executionWorkspaces, heartbeatRuns, issues, projectWorkspaces, workspaceOperations, type Db } from "@paperclipai/db"; +import { freshWorkspaceSchema, type PluginFreshWorkspace } from "@paperclipai/plugin-sdk"; +import { resolvePaperclipInstanceRoot } from "../home-paths.js"; +import { getStorageService } from "../storage/index.js"; +import { recoveryRunAncestry } from "./recovery-run-ancestry.js"; +import { readSnapshot } from "./workspace-recovery-snapshot.js"; +import { conflict, forbidden } from "../errors.js"; +import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; +import { SystemRecoveryFailure } from "./system-recovery-state.js"; + +export async function stageProviderFreshWorkspace(input: { + db: Db; companyId: string; issueId: string; attemptId: string; sourceRunId: string; + previousWorkspaceId: string | null; existing?: PluginFreshWorkspace; +}): Promise { + const fresh: PluginFreshWorkspace = freshWorkspaceSchema.parse({ attemptId: input.attemptId, + sourceRunId: input.sourceRunId, previousWorkspaceId: input.previousWorkspaceId, recoverySnapshots: [] }); + if (input.existing?.recoverySnapshots.length === 0) return freshWorkspaceSchema.parse(input.existing); + const storage = getStorageService(); + const sources = await recoveryRunAncestry({ ...input, sourceRunIds: [input.sourceRunId] }); + const stagingRoot = path.join(resolvePaperclipInstanceRoot(), "runtime", "provider-recovery", input.companyId, fresh.attemptId); + await mkdir(stagingRoot, { recursive: true, mode: 0o700 }); + const root = await realpath(stagingRoot); + for (const { id: sourceRunId } of sources) { + const operations = await input.db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.issueId, input.issueId), + eq(workspaceOperations.heartbeatRunId, sourceRunId), eq(workspaceOperations.phase, "workspace_recovery_snapshot"), + eq(workspaceOperations.status, "succeeded"))); + for (const operation of operations) { + if (input.existing && !input.existing.recoverySnapshots.some(snapshot => snapshot.operationId === operation.id)) continue; + const saved = readSnapshot(operation.metadata, input.companyId, sourceRunId); + const directory = path.join(root, operation.id); + await mkdir(directory, { recursive: true, mode: 0o700 }); + if (await realpath(directory) !== directory) throw new SystemRecoveryFailure("snapshot_verification_failed", "service_required"); + const files = {} as Pick; + for (const kind of ["archive", "manifest"] as const) { + const expected = saved[kind]; + const destination = path.join(directory, `${expected.sha256}.${kind === "archive" ? "tar" : "jsonl"}`); + const existing = await lstat(destination).catch((error: NodeJS.ErrnoException) => { + if (error.code !== "ENOENT") throw error; + return null; + }); + if (existing && (!existing.isFile() || existing.isSymbolicLink())) + throw new SystemRecoveryFailure("snapshot_verification_failed", "service_required"); + if (!existing) { + const temporary = `${destination}.${process.pid}.${randomUUID()}.tmp`; + const object = await storage.getObject(input.companyId, expected.objectKey); + try { + let received = 0; + const bounded = new Transform({ transform(bytes, _encoding, callback) { + received += bytes.length; + callback(received > expected.byteSize ? new SystemRecoveryFailure("snapshot_verification_failed", "service_required") : null, bytes); + } }); + await pipeline(object.stream, bounded, createWriteStream(temporary, { flags: "wx", mode: 0o400 })); + await verify(temporary, expected); + await rename(temporary, destination); + } finally { object.stream.destroy(); await rm(temporary, { force: true }); } + } + await verify(destination, expected); + files[kind] = { path: destination, sha256: expected.sha256, byteSize: expected.byteSize }; + } + fresh.recoverySnapshots.push({ operationId: operation.id, sourceRunId, ...files }); + } + } + fresh.recoverySnapshots.sort((a, b) => a.operationId.localeCompare(b.operationId)); + if (input.existing && !isDeepStrictEqual(fresh, input.existing)) + throw new SystemRecoveryFailure("snapshot_verification_failed", "service_required"); + return fresh; +} + +async function verify(file: string, expected: { sha256: string; byteSize: number }) { + const hash = createHash("sha256"); + let byteSize = 0; + for await (const bytes of createReadStream(file)) { + byteSize += bytes.length; + if (byteSize > expected.byteSize) throw new SystemRecoveryFailure("snapshot_verification_failed", "service_required"); + hash.update(bytes); + } + if (byteSize !== expected.byteSize || hash.digest("hex") !== expected.sha256) + throw new SystemRecoveryFailure("snapshot_verification_failed", "service_required"); +} + +export async function reserveFreshPluginTaskWorkspace(input: { + db: Db; companyId: string; issueId: string; runId: string; agentId: string; responsibleUserId: string | null; + fresh: PluginFreshWorkspace; expectedBindingRevision: number; + owner: Record; + source?: { projectId: string | null; projectWorkspaceId: string | null; repoUrl: string | null; baseRef: string | null }; +}) { + const source = input.source ?? { projectId: null, projectWorkspaceId: null, repoUrl: null, baseRef: null }; + return input.db.transaction(async tx => { + const [task] = await tx.select().from(issues).where(and(eq(issues.id, input.issueId), + eq(issues.companyId, input.companyId))).for("update"); + const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.companyId, input.companyId))); + if (!task || !run || run.agentId !== input.agentId || run.status !== "running" || + ![run.issueId, run.nativeIssueId].includes(input.issueId)) throw forbidden("Workspace preparation no longer has an active run"); + const actor = { type: "agent" as const, source: "agent_jwt" as const, + companyId: input.companyId, agentId: input.agentId, runId: input.runId, + onBehalfOfUserId: input.responsibleUserId === "local-board" ? null : input.responsibleUserId }; + if (!task.projectWorkspaceId && task.projectId && task.projectId === source.projectId && source.projectWorkspaceId) { + await assertTaskWorkspaceExecutionAccess(tx, actor, input.companyId, { issueId: input.issueId }); + const [defaultSource] = await tx.select({ id: projectWorkspaces.id }).from(projectWorkspaces).where(and( + eq(projectWorkspaces.companyId, input.companyId), eq(projectWorkspaces.projectId, task.projectId))) + .orderBy(asc(projectWorkspaces.createdAt), asc(projectWorkspaces.id)).limit(1); + if (defaultSource?.id === source.projectWorkspaceId) await tx.update(issues).set({ projectWorkspaceId: defaultSource.id }) + .where(eq(issues.id, task.id)); + } + await assertTaskWorkspaceExecutionAccess(tx, actor, input.companyId, { + issueId: input.issueId, projectWorkspaceId: source.projectWorkspaceId }); + const [reserved] = await tx.select().from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.sourceIssueId, input.issueId), + eq(executionWorkspaces.id, input.fresh.attemptId))); + if (reserved) { + if (task.executionWorkspaceId !== reserved.id || reserved.providerType !== "adapter_managed" || + !isDeepStrictEqual(reserved.metadata?.runnerTaskFreshWorkspace, input.fresh) || + !isDeepStrictEqual(reserved.metadata?.runnerTaskWorkspaceOwner, input.owner) || + Object.entries(source).some(([key, value]) => reserved[key as keyof typeof source] !== value)) + throw conflict("Task workspace binding changed; retry preparation"); + return reserved.id; + } + if (task.executionWorkspaceId !== input.fresh.previousWorkspaceId || task.workspaceBindingRevision !== input.expectedBindingRevision) + throw conflict("Task workspace binding changed; retry preparation"); + await notifyTerminalWorkspaceWork(tx, `workspace:${input.fresh.attemptId}`); + await tx.insert(executionWorkspaces).values({ id: input.fresh.attemptId, + companyId: input.companyId, sourceIssueId: input.issueId, ...source, + mode: "isolated_workspace", strategyType: "project_primary", name: "Fresh task workspace", status: "active", + providerType: "adapter_managed", cwd: `/workspaces/${input.fresh.attemptId}`, + metadata: { runnerTaskFreshWorkspace: input.fresh, runnerTaskWorkspaceOwner: input.owner }, + }); + await tx.update(issues).set({ executionWorkspaceId: input.fresh.attemptId, executionWorkspacePreference: "reuse_existing", + workspaceBindingRevision: task.workspaceBindingRevision + 1, + workspacePendingSelection: task.workspacePendingSelection?.expectedBindingRevision === task.workspaceBindingRevision + ? { ...task.workspacePendingSelection, expectedBindingRevision: task.workspaceBindingRevision + 1 } : task.workspacePendingSelection, updatedAt: new Date() }).where(eq(issues.id, task.id)); + return input.fresh.attemptId; + }); +} + +export async function releaseProviderRecoveryTransfers(companyId: string, input: PluginFreshWorkspace) { + const fresh = freshWorkspaceSchema.parse(input); + await rm(path.join(resolvePaperclipInstanceRoot(), "runtime", "provider-recovery", companyId, fresh.attemptId), + { recursive: true, force: true }); +} diff --git a/server/src/services/workspace-recovery-snapshot.ts b/server/src/services/workspace-recovery-snapshot.ts index 62c3d0212ba..37d4f9030d5 100644 --- a/server/src/services/workspace-recovery-snapshot.ts +++ b/server/src/services/workspace-recovery-snapshot.ts @@ -23,7 +23,7 @@ const snapshotSchema = z.object({ leaseId: z.string().uuid(), providerLeaseId: z.string().min(1), archive: storedFileSchema, manifest: storedFileSchema, }); export type PreservedWorkspaceSnapshot = z.infer; -function readSnapshot(value: unknown, companyId: string, runId: string) { +export function readSnapshot(value: unknown, companyId: string, runId: string) { const parsed = snapshotSchema.safeParse(value); if (!parsed.success) throw new SystemRecoveryFailure("snapshot_verification_failed", "service_required"); const s = parsed.data;