From 989ee668e08b4161ff7cfdedb5f19ccfafe740e4 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Thu, 22 Oct 2026 17:46:27 -0700 Subject: [PATCH] Support provider-owned fresh retry workspaces (#17252) ## Thinking Path > - Paperclip coordinates AI agents and retains their work across failures. > - A retry can require a new workspace while preserving the failed attempt. > - Core rejects this recovery strategy for provider-owned storage. > - Providers should implement the required outcome using their own storage design. > - This pull request adds a semantic fresh-preparation contract and delegates recovery preparation. > - The benefit is that a supported provider can run a fresh retry without losing prior work. ## Linked Issues or Issue Description **What happened?** A retry that selects `fresh_isolated_workspace` fails with `fresh_workspace_storage_unsupported` for provider-owned storage. **Expected behavior** Core asks a capable provider to prepare new storage and attach verified recovery files. The failed workspace remains intact. **Steps to reproduce** Run a task in a task-capable plugin environment. Let the task fail without a verified workspace preservation receipt. Retry the task so Core selects a fresh attempt. **Paperclip version or commit** 6204239cf (rebased on current master). **Deployment mode** Self-hosted server with a plugin task environment and managed workspace provider. Search found no duplicate implementation. This branch includes the default-provider changes from #17247 and preserves their source selection behavior. ## What Changed - Define an opt-in fresh-preparation contract with a stable attempt identity and verified recovery file descriptors. - Describe retained-storage and recovery-attachment guarantees without requiring a Git storage layout. - Reserve replacement task bindings with revision checks and keep the failed workspace intact. Projectless and project-backed task environments both replay their stable attempt reservation. - Reconcile an interrupted preparation against its reserved workspace and stable operation identity. - Stage verified recovery archives at stable transfer paths. Delegate attachment materialization to the provider. - Preserve projectless attempt directory identities through later turns and pin the intent into each execution lease. - Protect fresh intent and projectless ownership from caller metadata replacement or injection. ## Verification - SDK workspace-provider tests: 4 passed. - Targeted reservation and task-dispatch tests: 31 passed. - Recovery transfer staging and lifecycle focus: 7 passed. The separate cache-cleanup failure reconciliation case also passed. - After rebasing on current master, 7 focused preparation and ownership cases passed. These cover inherited provider defaults, fresh preparation, stable projectless identity, and authorization denial. - The focused metadata authority regression passed, including forged intent, replacement, and null metadata patches. - Three assembled interrupted task-reservation cases passed: projectless, project with a saved source, and project with a newly selected default source. - Targeted tests use worktree SDK sources. No full local build or typecheck ran. - Local checks remain targeted as requested. Latest-head GitHub CI passed: 54 successful checks, including typecheck, build, server suites, Runner validation, and E2E shards; 2 unrelated Storybook checks were skipped. - Greptile reviewed commit 73595b0c616edeccf924e23fdd39bf9c083c955e at 5/5 with zero unresolved threads. - Live VM E2E has not run. ## Risks - Existing providers retain their existing behavior. Fresh attempts require explicit version support. - The companion provider implementation must materialize recovery snapshots safely and preserve the old workspace. - Fresh retries require the companion provider implementation. Providers without the version marker fail closed. ## Model Used - OpenAI GPT-6.1 (`gpt-6.1-sol`), reasoning and coding tools. The runtime does not expose the context-window limit. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md | 31 ++++ packages/plugins/sdk/src/environment-tasks.ts | 2 + packages/plugins/sdk/src/worker-rpc-host.ts | 2 + .../plugins/sdk/src/workspace-provider.ts | 17 ++ .../sdk/tests/workspace-provider.test.ts | 9 ++ packages/shared/src/types/plugin.ts | 2 + packages/shared/src/validators/plugin.ts | 2 + .../environment-task-runtime.test.ts | 33 +++- ...lugin-fresh-workspaces.integration.test.ts | 124 +++++++++++++++ .../services/environment-run-orchestrator.ts | 8 + .../src/services/environment-task-runtime.ts | 18 ++- server/src/services/execution-workspaces.ts | 8 + .../heartbeat/workspace-preparation.test.ts | 67 ++++++-- .../heartbeat/workspace-preparation.ts | 61 +++++-- server/src/services/plugin-workspaces.ts | 65 ++++++-- .../services/provider-fresh-workspace.test.ts | 41 +++++ .../src/services/provider-fresh-workspace.ts | 150 ++++++++++++++++++ .../services/workspace-recovery-snapshot.ts | 2 +- 18 files changed, 592 insertions(+), 50 deletions(-) create mode 100644 server/src/__tests__/plugin-fresh-workspaces.integration.test.ts create mode 100644 server/src/services/provider-fresh-workspace.test.ts create mode 100644 server/src/services/provider-fresh-workspace.ts 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;