diff --git a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md index a4bc0be0b53..027da7c0021 100644 --- a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md +++ b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md @@ -172,6 +172,16 @@ It does not qualify a deployed plugin, live account, or production WSS proxy. ## Workspace completion and maintenance +A storage provider can declare `workspaceProvider.workspaceVerificationVersion: 1`. +Before creating a fresh workspace for an admitted Retry, Core may send the +original task-owned workspace's `prepare` request with `verifyExisting: true`. +The provider must verify a completed checkout without creating or fetching +one and return `prepared` with `created: false`. Core checks prior writer +settlement and continues normal preparation on the retained workspace after +verification. A missing or unconfirmed checkout leaves the fresh retry +fallback in place. Local workspaces use the same ownership and writer +checks with Core's existing local Git verification. + A storage provider can declare `workspaceProvider.finalizationVersion: 1`. Core then sends `workspaceOperation` with `operation.kind: "finalize"`, the retained binding, lifecycle generation, and originating `runId` and `agentId`. The operation diff --git a/packages/plugins/sdk/src/workspace-provider.ts b/packages/plugins/sdk/src/workspace-provider.ts index c212c3baffd..5db136efa4b 100644 --- a/packages/plugins/sdk/src/workspace-provider.ts +++ b/packages/plugins/sdk/src/workspace-provider.ts @@ -50,6 +50,7 @@ export const workspaceProviderOperationSchema = z.discriminatedUnion("kind", [ baseRef: z.string().nullable(), existing: workspaceProviderBindingSchema.nullable(), fresh: freshWorkspaceSchema.optional(), + verifyExisting: z.literal(true).optional(), }) .strict(), // Verify the retained workspace and flush run writes. This does not certify diff --git a/packages/shared/src/types/plugin.ts b/packages/shared/src/types/plugin.ts index 36e0f50d97f..016059b873c 100644 --- a/packages/shared/src/types/plugin.ts +++ b/packages/shared/src/types/plugin.ts @@ -721,6 +721,7 @@ export interface PaperclipPluginManifestV1 { workspaceProvider?: { /** Supports bounded run completion separately from removal-safety inspection. */ finalizationVersion?: 1; + workspaceVerificationVersion?: 1; supportsMergeAncestry?: boolean; recoveryPullRequestVersion?: 1; freshWorkspaceVersion?: 1; diff --git a/packages/shared/src/validators/plugin.ts b/packages/shared/src/validators/plugin.ts index 054de474635..2a90d3d00b6 100644 --- a/packages/shared/src/validators/plugin.ts +++ b/packages/shared/src/validators/plugin.ts @@ -812,6 +812,7 @@ export const pluginManifestV1Schema = z.object({ agentLifecycle: z.literal(true).optional(), workspaceProvider: z.object({ finalizationVersion: z.literal(1).optional(), + workspaceVerificationVersion: z.literal(1).optional(), supportsMergeAncestry: z.boolean().optional(), recoveryPullRequestVersion: z.literal(1).optional(), freshWorkspaceVersion: z.literal(1).optional(), diff --git a/server/src/__tests__/heartbeat-session-preparation.test.ts b/server/src/__tests__/heartbeat-session-preparation.test.ts index f918f482b12..8e1499e26f7 100644 --- a/server/src/__tests__/heartbeat-session-preparation.test.ts +++ b/server/src/__tests__/heartbeat-session-preparation.test.ts @@ -8,6 +8,7 @@ import { agents, agentTaskSessions, authUsers, companies, companyMemberships, cr import { registerServerAdapter, unregisterServerAdapter, type AdapterExecutionContext } from "../adapters/index.js"; import { heartbeatService } from "../services/heartbeat.js"; import { instanceSettingsService } from "../services/instance-settings.js"; +import { requestTaskRetry } from "../services/task-retry.js"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; import * as environmentOrchestrator from "../services/environment-run-orchestrator.js"; import { createPluginWorkspaceFixture } from "./helpers/plugin-workspaces.js"; @@ -56,7 +57,7 @@ describePostgres("heartbeat session preparation through adapter dispatch", () => await db.insert(companyMemberships).values({ companyId, principalId: userId, principalType: "user", status: "active", membershipRole: "member" }); await db.insert(agents).values({ id: agentId, companyId, name: "Session reader", adapterType, status: "idle", adapterConfig: { model: "model-a" }, runtimeConfig }); await db.insert(issues).values({ id: issueId, companyId, title: "Session fixture", status: "todo", assigneeAgentId: agentId, responsibleUserId: userId, issueNumber: 1 }); - return { companyId, agentId, issueId }; + return { companyId, agentId, issueId, userId }; } async function execute(f: Awaited>, context: Record = {}) { @@ -113,10 +114,11 @@ describePostgres("heartbeat session preparation through adapter dispatch", () => }); it.each([ - null, - { mode: "isolated_workspace" }, - { workspaceStrategy: { type: "git_worktree", branchTemplate: "work" } }, - ])("keeps a default-attached repository policy stable through binding with settings %j and retains the provider on policy changes", async (settings) => { + { settings: null, retry: false }, + { settings: { mode: "isolated_workspace" }, retry: false }, + { settings: { workspaceStrategy: { type: "git_worktree", branchTemplate: "work" } }, retry: false }, + { settings: null, retry: true }, + ])("keeps a default-attached repository policy stable through binding ($settings, Retry: $retry)", async ({ settings, retry }) => { const f = await seed(); const storage = await createPluginWorkspaceFixture(db, f.companyId); const [plugin] = await db.select().from(plugins).where(eq(plugins.id, storage.pluginId)); @@ -125,7 +127,8 @@ describePostgres("heartbeat session preparation through adapter dispatch", () => await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson, capabilities: [...plugin.manifestJson.capabilities, "environment.drivers.register"], environmentDrivers: [{ driverKey: "tasks", displayName: "Tasks", supportsTasks: true, configSchema: { type: "object" } }], - workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, defaultForProjects: { mode: "isolated_workspace", strategy: "git_worktree" } }, + workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, defaultForProjects: { mode: "isolated_workspace", strategy: "git_worktree" }, + ...(retry ? { workspaceVerificationVersion: 1 as const } : {}) }, } }).where(eq(plugins.id, storage.pluginId)); await db.update(agents).set({ defaultEnvironmentId: environment.id }).where(eq(agents.id, f.agentId)); await db.update(projects).set({ executionWorkspacePolicy: { enabled: true, storageProviderPluginId: storage.pluginId } }).where(eq(projects.id, storage.input.projectId)); @@ -134,6 +137,7 @@ describePostgres("heartbeat session preparation through adapter dispatch", () => }).where(eq(issues.id, f.issueId)); await instanceSettingsService(db).updateExperimental({ enableIsolatedWorkspaces: true, enableIsolatedWorkspacesByDefault: true }); const cwd = await mkdtemp(join(tmpdir(), "session-provider-fixture-")); + const realizedBase = new Map(); const createOrchestrator = environmentOrchestrator.environmentRunOrchestrator; const realization = vi.spyOn(environmentOrchestrator, "environmentRunOrchestrator").mockImplementation((...args) => ({ ...createOrchestrator(...args), @@ -141,13 +145,21 @@ describePostgres("heartbeat session preparation through adapter dispatch", () => // transport while preserving the configured storage/task-driver identity. acquireForRun: async input => createOrchestrator(...args).acquireForRun({ ...input, selectedEnvironmentId: input.localEnvironmentId }), realizeForRun: async input => { + realizedBase.set(input.heartbeatRunId, input.executionWorkspace.baseRefSha); input.executionWorkspace.cwd = cwd; return { lease: input.lease, persistedExecutionWorkspace: input.persistedExecutionWorkspace, workspaceRealization: {}, executionTarget: null, remoteExecution: null }; }, })); const originalHeartbeat = heartbeat; - heartbeat = heartbeatService(db, { pluginWorkerManager: storage.workers as PluginWorkerManager }); + const storageWorkers = { + ...storage.workers, + call: async (...args: Parameters) => { + const result = await storage.workers.call(...args) as Record; + return storage.calls.at(-1)?.operation.verifyExisting ? { ...result, created: false } : result; + }, + }; + heartbeat = heartbeatService(db, { pluginWorkerManager: storageWorkers as PluginWorkerManager }); try { const first = await execute(f); const [bound] = await db.select().from(issues).where(eq(issues.id, f.issueId)); @@ -157,6 +169,28 @@ describePostgres("heartbeat session preparation through adapter dispatch", () => const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, bound.executionWorkspaceId!)); expect([...captured.values()].filter(input => input.context.issueId === f.issueId)).toHaveLength(1); expect(workspace).toMatchObject({ providerType: "plugin", strategyType: "git_worktree" }); + if (retry) { + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "setup_failed" }) + .where(eq(heartbeatRuns.id, first.finished.id)); + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, f.issueId)); + const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, + actorId: f.userId, actorSource: "local_implicit", observedRunId: first.finished.id, idempotencyKey: randomUUID() }); + expect(receipt.accepted).toBe(true); + await heartbeat.resumeTaskRetry(f.companyId, receipt.requestId!); + await heartbeat.drainActiveRunExecutions(); + const [resumed] = [...captured.values()].filter(input => input.context.issueId === f.issueId && input.runId !== first.finished.id); + expect(resumed?.context.paperclipWorkspace).toMatchObject({ branchName: workspace.branchName }); + expect(realizedBase.get(first.finished.id)).toBe("a".repeat(40)); + expect(realizedBase.get(resumed.runId)).toBe(realizedBase.get(first.finished.id)); + const preparations = storage.calls.filter(call => call.runId === resumed.runId && call.operation.kind === "prepare"); + expect(preparations).toHaveLength(2); + expect(preparations[0]).toMatchObject({ executionWorkspaceId: workspace.id, operation: { verifyExisting: true } }); + expect(preparations[1]).toMatchObject({ executionWorkspaceId: workspace.id }); + expect(preparations[1].operation.verifyExisting).toBeUndefined(); + const [afterRetry] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(afterRetry.executionWorkspaceId).toBe(workspace.id); + return; + } for (let turn = 0; turn < 2; turn++) { const next = await execute(f); expect(next.input.runtime.sessionId).toBe("saved-session"); diff --git a/server/src/__tests__/plugin-workspaces.integration.test.ts b/server/src/__tests__/plugin-workspaces.integration.test.ts index 372431d8711..ae1a45b3fcd 100644 --- a/server/src/__tests__/plugin-workspaces.integration.test.ts +++ b/server/src/__tests__/plugin-workspaces.integration.test.ts @@ -5,6 +5,7 @@ import { and, eq, sql } from "drizzle-orm"; import { createDb, subscribeDatabaseWork, + environmentLeases, executionWorkspaces, heartbeatRuns, issues, @@ -19,6 +20,7 @@ import { removePluginWorkspace, type WorkspaceWorkers, } from "../services/plugin-workspaces.js"; +import { verifyRetryWorkspace } from "../services/fresh-workspace-continuation.js"; import { executionWorkspaceService } from "../services/execution-workspaces.js"; import { WorkspaceProviderFailureError } from "../services/workspace-provider-failure.js"; import { findPluginWorkerUnavailableError, PluginWorkerUnavailableError } from "../services/plugin-worker-unavailable-error.js"; @@ -47,6 +49,50 @@ const support = await getEmbeddedPostgresTestSupport(); }); const fixture = (companyId?: string) => createPluginWorkspaceFixture(db, companyId); + it.each(["ready", "lost_reply", "unconfirmed", "live_writer", "unsupported"] as const)( + "verifies the same checkout before choosing a fresh retry (%s)", async (state) => { + const f = await fixture(); + const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); + if (state !== "unsupported") await db.update(plugins).set({ manifestJson: { + ...plugin.manifestJson, workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, workspaceVerificationVersion: 1 }, + } }).where(eq(plugins.id, f.pluginId)); + if (state === "lost_reply") { + f.lose(); + await expect(preparePluginWorkspace(db, f.workers, f.input)).rejects.toThrow(); + } else { + await preparePluginWorkspace(db, f.workers, f.input); + } + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const workspaceId = task.executionWorkspaceId!; + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + if (state !== "lost_reply") await db.insert(environmentLeases).values({ + companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.runId, + provider: "plugin:test:tasks", executionWorkspaceId: workspaceId, leasePolicy: "ephemeral", + status: state === "live_writer" ? "active" : "released", + releasedAt: state === "live_writer" ? null : new Date(), + }); + const retryIntentId = randomUUID(); + const [retry] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.input.agentId, + issueId: f.issueId, status: "running", contextSnapshot: { taskRetryRequestId: retryIntentId }, + }).returning(); + const fresh = { mode: "fresh_isolated_workspace", retryIntentId, sourceRunId: f.runId, retainedSourceIds: [] } as const; + const workers: WorkspaceWorkers = { ...f.workers, call: async (...args) => { + const result = await f.workers.call(...args) as { kind: string }; + if (state === "unconfirmed") throw new Error("checkout incomplete"); + return { ...result, created: false }; + } }; + const result = await verifyRetryWorkspace(db, workers, retry, f.issueId, null, + { ...fresh, retainedSourceIds: [] }); + const verified = state === "ready" || state === "lost_reply"; + expect(result).toEqual(verified + ? { mode: "reuse_issue_workspace", retryIntentId, sourceRunId: f.runId, workspaceId, cwd: null } + : fresh); + expect(f.calls.slice(1)).toHaveLength(state === "live_writer" || state === "unsupported" ? 0 : 1); + if (f.calls.length > 1) expect(f.calls[1]).toMatchObject({ executionWorkspaceId: workspaceId, + runId: retry.id, operation: { kind: "prepare", verifyExisting: true } }); + const [retained] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)); + expect(retained.providerRef).toBe(`checkout:${workspaceId}`); + }); it.each([ ["missing", undefined], ["stopped", { status: "stopped", supportedMethods: [] }], diff --git a/server/src/services/execution-preparation-timing.ts b/server/src/services/execution-preparation-timing.ts index 01041233d1a..3cc34e3f532 100644 --- a/server/src/services/execution-preparation-timing.ts +++ b/server/src/services/execution-preparation-timing.ts @@ -16,7 +16,7 @@ type PreparationPhase = "environment_selection" | "execution_configuration" | "s | "configuration_ai_runtime" | "configuration_skills" | "configuration_connector_assignments" | "configuration_connector_skills" | "configuration_connector_delivery" | "configuration_connection_instructions" | "configuration_agent_identity" - | "workspace_source_resolution" + | "workspace_source_resolution" | "workspace_verification" | "workspace_provision_policy" | "workspace_base_validation" | "workspace_source_authorization" diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 0675377b3ce..836f636e109 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -591,7 +591,7 @@ export async function admitExplicitNativeContinuation(input: { // holds can describe the same failed run; each hold must permit the reuse. const sourceRunIds = new Set(freshCandidates.map(candidate => candidate.run.id)); const only = sourceRunIds.size === 1 && freshCandidates.every(candidate => !candidate.remote) ? freshCandidates[0]! : null; - let retained: { id: string; cwd: string } | null = null; + let retained: { id: string; cwd: string | null } | null = null; const gitProbes = new Map>(); for (const candidate of only ? freshCandidates : []) { const checked = await retainedIssueWorkspaceForRetry(db, { companyId, issueId, agentId, diff --git a/server/src/services/fresh-workspace-continuation.ts b/server/src/services/fresh-workspace-continuation.ts index c912df92970..9888ae7ebf3 100644 --- a/server/src/services/fresh-workspace-continuation.ts +++ b/server/src/services/fresh-workspace-continuation.ts @@ -19,19 +19,20 @@ import { readFailedPreparationSettlement } from "./failed-preparation-settlement import { taskRetryRunAncestry } from "./recovery-run-ancestry.js"; import { parseNativeExecutionInput } from "../vendor/paperclip-runner/index.js"; import { nativeExecutionUsesTaskWorkspace } from "./heartbeat/external-provider.js"; +import { preparePluginWorkspace, type WorkspaceWorkers } from "./plugin-workspaces.js"; export type ContinuationReadiness = | { mode: "resume_preserved_workspace"; preservationReceiptId: string } | { mode: "reuse_current_workspace"; retryIntentId: string; sourceRunId: string; workspaceId: string; cwd: string } | { mode: "resume_stopped_native_workspace"; retryIntentId: string; sourceRunId: string; workspaceId: string; cwd: string } - | { mode: "reuse_issue_workspace"; retryIntentId: string; sourceRunId: string; workspaceId: string; cwd: string } + | { mode: "reuse_issue_workspace"; retryIntentId: string; sourceRunId: string; workspaceId: string; cwd: string | null } | { mode: "fresh_isolated_workspace"; retryIntentId: string; sourceRunId: string; retainedSourceIds: string[] }; const freshReadinessSchema = z.discriminatedUnion("mode", [z.object({ mode: z.literal("fresh_isolated_workspace"), retryIntentId: z.string().uuid(), sourceRunId: z.string().uuid(), retainedSourceIds: z.array(z.string().uuid()), }), z.object({ mode: z.literal("reuse_issue_workspace"), retryIntentId: z.string().uuid(), - sourceRunId: z.string().uuid(), workspaceId: z.string().uuid(), cwd: z.string().min(1), + sourceRunId: z.string().uuid(), workspaceId: z.string().uuid(), cwd: z.string().min(1).nullable(), })]); const workspaceExportCauses = ["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"]; @@ -126,7 +127,9 @@ export async function retainedIssueWorkspaceForRetry(db: Db, input: { companyId: string; issueId: string; agentId: string; successorRunId: string; cause: string | null; source: Pick; evidence: Record | null; gitProbes?: Map>; -}): Promise<{ id: string; cwd: string } | null> { + verifyProviderWorkspace?: (workspace: typeof executionWorkspaces.$inferSelect) => Promise; + sourceRetryIntentId?: unknown; +}): Promise<{ id: string; cwd: string | null } | null> { const keepFresh = (reason: string) => { logger.info({ companyId: input.companyId, issueId: input.issueId, sourceRunId: input.source.id, successorRunId: input.successorRunId, reason }, "Retry keeps a fresh attempt; the task workspace is not reusable"); @@ -140,23 +143,38 @@ export async function retainedIssueWorkspaceForRetry(db: Db, input: { const leases = await db.select().from(environmentLeases).where(and( eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.heartbeatRunId, input.source.id), )); - if (!leases.length || leases.some(lease => lease.provider !== "local" || !lease.releasedAt || - lease.cleanupStatus === "failed" || !lease.executionWorkspaceId)) return keepFresh("source_lease_not_local_released"); + if (leases.some(lease => !lease.releasedAt || lease.cleanupStatus === "failed" || !lease.executionWorkspaceId)) + return keepFresh("source_lease_not_released"); const sourceWorkspaceIds = new Set(leases.map(lease => lease.executionWorkspaceId)); const [bound] = await db.select({ workspace: executionWorkspaces }).from(issues).innerJoin(executionWorkspaces, and( eq(executionWorkspaces.companyId, issues.companyId), eq(executionWorkspaces.id, issues.executionWorkspaceId), )).where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId), eq(issues.assigneeAgentId, input.agentId))).limit(1); const workspace = bound?.workspace; - if (!workspace?.cwd || sourceWorkspaceIds.size !== 1 || !sourceWorkspaceIds.has(workspace.id)) + if (!workspace || workspace.sourceIssueId !== input.issueId || workspace.mode === "shared_workspace" || + !reusableWorkspaceStatuses.includes(workspace.status)) return keepFresh("workspace_not_task_owned"); + const verifyProvider = workspace.providerType === "plugin" ? input.verifyProviderWorkspace : undefined; + const fresh = workspace.metadata?.workspaceProviderFresh as { attemptId?: unknown } | undefined; + const reservedBySource = verifyProvider && !leases.length && + (workspace.metadata?.workspaceProviderReservationRunId === input.source.id || + (typeof input.sourceRetryIntentId === "string" && fresh?.attemptId === input.sourceRetryIntentId)); + if (!reservedBySource && (sourceWorkspaceIds.size !== 1 || !sourceWorkspaceIds.has(workspace.id))) return keepFresh("binding_not_source_workspace"); - if (!isTaskOwnedLocalWorkspace(workspace, input.issueId)) return keepFresh("workspace_not_task_owned_local"); + if (!verifyProvider && (leases.some(lease => lease.provider !== "local") || + !isTaskOwnedLocalWorkspace(workspace, input.issueId) || !workspace.cwd)) + return keepFresh("workspace_not_task_owned_local"); const writer = await workspaceHasOtherWriter(db, { companyId: input.companyId, workspaceId: workspace.id, successorRunId: input.successorRunId, stoppedSource: input.source }); if (writer) return keepFresh(writer); + if (verifyProvider) { + if (!await verifyProvider(workspace)) return keepFresh("provider_workspace_unverified"); + return { id: workspace.id, cwd: workspace.cwd }; + } // The directory must still be the recorded one, not a replacement link. - const stat = await lstat(workspace.cwd).catch(() => null); - const real = stat?.isDirectory() && !stat.isSymbolicLink() ? await realpath(workspace.cwd).catch(() => null) : null; + const cwd = workspace.cwd; + if (!cwd) return keepFresh("directory_missing"); + const stat = await lstat(cwd).catch(() => null); + const real = stat?.isDirectory() && !stat.isSymbolicLink() ? await realpath(cwd).catch(() => null) : null; if (!real) return keepFresh("directory_missing"); let probe = input.gitProbes?.get(real); if (!probe) { @@ -407,7 +425,7 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick>'runId' = ${source.id}`)); - let retained: { id: string; cwd: string } | null = null; + let retained: { id: string; cwd: string | null } | null = null; const gitProbes = new Map>(); for (const hold of holds.length ? holds : [{ cause: null, evidence: null }]) { const checked = await retainedIssueWorkspaceForRetry(db, { companyId: run.companyId, issueId, @@ -461,3 +479,51 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick, + issueId: string, + responsibleUserId: string | null, + readiness: ContinuationReadiness, +): Promise { + if (readiness.mode !== "fresh_isolated_workspace" || + run.contextSnapshot?.taskRetryRequestId !== readiness.retryIntentId) return readiness; + const [source] = await db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId), + eq(heartbeatRuns.id, readiness.sourceRunId), + sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueId}`, + )); + if (!source) return readiness; + const holds = await db.select({ cause: issueRecoveryActions.cause, evidence: issueRecoveryActions.evidence }) + .from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, run.companyId), + eq(issueRecoveryActions.sourceIssueId, issueId), + sql`${issueRecoveryActions.evidence}->>'runId' = ${source.id}`)); + if (holds.some(hold => !recoveryGenerationSettled(hold.evidence, source.id) || + sourceNeedsWorkspaceRepair(hold.cause, source.resultJson))) return readiness; + const retained = await retainedIssueWorkspaceForRetry(db, { + companyId: run.companyId, issueId, agentId: run.agentId, successorRunId: run.id, + source, cause: null, evidence: null, + sourceRetryIntentId: source.contextSnapshot?.taskRetryRequestId, + verifyProviderWorkspace: workers ? async workspace => { + try { + return Boolean(await preparePluginWorkspace(db, workers, { + companyId: run.companyId, issueId, agentId: run.agentId, runId: run.id, responsibleUserId, + projectId: workspace.projectId, projectWorkspaceId: workspace.projectWorkspaceId, + existingWorkspaceId: workspace.id, mode: workspace.mode, strategy: workspace.strategyType, + verifyExisting: true, + })); + } catch { + return false; + } + } : undefined, + }); + if (!retained) return readiness; + const reuse: ContinuationReadiness = { mode: "reuse_issue_workspace", retryIntentId: readiness.retryIntentId, + sourceRunId: source.id, workspaceId: retained.id, cwd: retained.cwd }; + await persistActivity(db, { companyId: run.companyId, actorType: "system", actorId: "task-recovery", + action: "issue.retry_workspace_reused", entityType: "issue", entityId: issueId, runId: run.id, details: reuse }); + return reuse; +} diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 72916c807b0..c36a4eb7871 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -31,7 +31,7 @@ import { registerCancelledWorkspacePreservationWork } from "./native-runtime/nat import { readUpgradeReadiness } from "./heartbeat/upgrade-readiness.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import { writeExecutionControl } from "./execution-control-notifications.js"; -import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; +import { readFreshWorkspaceReadiness, verifyRetryWorkspace } from "./fresh-workspace-continuation.js"; import { settleFailedNativeRecoveryPreparation } from "./native-runtime/native-recovery-preparation-failure.js"; import { promptInjectionPreventionService } from "./prompt-injection-prevention.js"; import { createHeartbeatAgentStatus } from "./heartbeat/agent-status.js"; @@ -2254,7 +2254,7 @@ export function heartbeatService( const taskKey = deriveTaskKeyWithHeartbeatFallback(context, null); const sessionCodec = getAdapterSessionCodec(agent.adapterType); const issueId = readNonEmptyString(context.issueId); - const freshWorkspace = await timePreparation("workspace_selection", async () => { + let freshWorkspace = await timePreparation("workspace_selection", async () => { const readiness = await readFreshWorkspaceReadiness(db, run); if (issueId && canApplyTaskWorkspaceSelectionAtAdmission({ admittedInput: parseObject(run.runnerProfileJson).nativeExecutionInput, restarting: Boolean(runOptions.nativeRestartRecovery), hasLeaseOwner: Boolean(runOptions.nativeLeaseOwner) })) { const changed = await executionWorkspacesSvc.applyPendingTaskWorkspaceSelection({ companyId: agent.companyId, @@ -2537,6 +2537,11 @@ export function heartbeatService( // Initialization has persisted the active context, including an explicit // absence of identity inherited from an automatic continuation. responsibleUserId = identityContext.responsibleUserId; + if (issueId && freshWorkspace?.mode === "fresh_isolated_workspace") { + const fallback = freshWorkspace; + freshWorkspace = await timePreparation("workspace_verification", () => + verifyRetryWorkspace(db, options.pluginWorkerManager, run, issueId, responsibleUserId, fallback)); + } run = { ...run, activeIdentityContextId: identityContext.id, diff --git a/server/src/services/plugin-workspaces.ts b/server/src/services/plugin-workspaces.ts index e7abdb4698d..6faed0e455b 100644 --- a/server/src/services/plugin-workspaces.ts +++ b/server/src/services/plugin-workspaces.ts @@ -1,5 +1,5 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; -import { assertProjectStorageEnvironment, effectiveProjectWorkspacePolicy, isWorkspaceProvider } from "./project-workspace-provider.js"; +import { assertProjectStorageEnvironment, effectiveProjectWorkspacePolicy, requireWorkspaceProviderWorker, type WorkspaceWorkers } from "./project-workspace-provider.js"; import { releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js"; import { retireAbandonedPluginWorkspaceReservation } from "./plugin-workspace-reservations.js"; import { randomUUID, createHash } from "node:crypto"; @@ -8,8 +8,6 @@ import { executionWorkspaces, heartbeatRuns, issues, - pluginCompanySettings, - plugins, projects, projectWorkspaces, type Db, @@ -26,61 +24,22 @@ import { type PluginWorkspaceParams, type PluginFreshWorkspace, } from "@paperclipai/plugin-sdk"; -import type { PluginWorkerManager } from "./plugin-worker-manager.js"; import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; import { withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js"; -import { workspaceProviderCallFailure, workspaceWorkerRefusal } from "./workspace-provider-failure.js"; +import { workspaceProviderCallFailure } from "./workspace-provider-failure.js"; import { assertTaskWorkspaceExecutionAccess, assertTaskWorkspaceSourceProjectAccess, } from "./task-workspace-access.js"; import { forbidden, conflict } from "../errors.js"; -export type WorkspaceWorkers = Pick; +export type { WorkspaceWorkers } from "./project-workspace-provider.js"; type Workspace = typeof executionWorkspaces.$inferSelect; -async function provider( - db: Db, - workers: WorkspaceWorkers | undefined, - companyId: string, - pluginId: string, - /** Before a call, a ready provider whose worker is not running is a retryable refusal: no request is sent. */ - beforeCall = false, -) { - const [plugin] = await db - .select() - .from(plugins) - .where(eq(plugins.id, pluginId)); - const [settings] = await db - .select() - .from(pluginCompanySettings) - .where( - and( - eq(pluginCompanySettings.pluginId, pluginId), - eq(pluginCompanySettings.companyId, companyId), - ), - ); - const worker = workers?.getWorker(pluginId); - const refusal = beforeCall && workers && plugin?.status === "ready" && settings?.enabled !== false && - isWorkspaceProvider(plugin.manifestJson) ? workspaceWorkerRefusal(pluginId, worker) : null; - if (refusal) throw refusal; - if ( - !workers || - !plugin || - plugin.status !== "ready" || - settings?.enabled === false || - !isWorkspaceProvider(plugin.manifestJson) || - !worker?.supportedMethods?.includes("workspaceOperation") - ) { - throw conflict("The workspace storage plugin is unavailable"); - } - return plugin; -} - export async function pluginWorkspaceSupportsMergeAncestry(db: Db, workers: WorkspaceWorkers | undefined, workspace: Workspace) { const pluginId = readWorkspaceProviderBinding(workspace.metadata)?.pluginId; if (!pluginId) return true; - const selected = await provider(db, workers, workspace.companyId, pluginId); + const selected = await requireWorkspaceProviderWorker(db, workers, workspace.companyId, pluginId); return selected.manifestJson.workspaceProvider?.supportsMergeAncestry !== false; } @@ -103,7 +62,10 @@ export async function callWorkspaceProvider( const pluginId = binding?.pluginId ?? input.pluginId; if (!pluginId || !workspace.projectId) throw conflict("Workspace storage ownership is missing"); - const selectedProvider = await provider(db, workers, workspace.companyId, pluginId, true); + const selectedProvider = await requireWorkspaceProviderWorker(db, workers, workspace.companyId, pluginId, true); + const verifyExisting = input.operation.kind === "prepare" && input.operation.verifyExisting === true; + if (verifyExisting && selectedProvider.manifestJson.workspaceProvider?.workspaceVerificationVersion !== 1) + throw conflict("Workspace verification is unavailable"); if (input.operation.kind === "prepare" && input.operation.fresh && selectedProvider.manifestJson.workspaceProvider?.freshWorkspaceVersion !== 1) throw conflict("Fresh workspace preparation is unavailable"); if (input.operation.kind === "finalize" && @@ -120,7 +82,7 @@ export async function callWorkspaceProvider( throw conflict("Workspace removal identity does not match its recorded payload"); if ((operation.kind === "inspect" || operation.kind === "remove") && input.removalOperationId === undefined && selectedProvider.manifestJson.workspaceProvider?.recoveryPullRequestVersion !== 1) delete operation.recoveryPullRequest; - if (operation.kind === "prepare" && operation.fresh) { + if (operation.kind === "prepare" && operation.fresh && !verifyExisting) { 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 }); @@ -140,7 +102,7 @@ export async function callWorkspaceProvider( ) .digest("hex")), generation: operationGeneration, - deadline: new Date(Date.now() + (operation.kind === "finalize" ? 30_000 : 300_000)).toISOString(), + deadline: new Date(Date.now() + (verifyExisting ? 10_000 : operation.kind === "finalize" ? 30_000 : 300_000)).toISOString(), runId: input.runId ?? null, agentId: input.agentId ?? null, operation, @@ -151,7 +113,7 @@ export async function callWorkspaceProvider( pluginId, "workspaceOperation", params, - operation.kind === "finalize" ? 40_000 : 310_000, + verifyExisting ? 12_000 : operation.kind === "finalize" ? 40_000 : 310_000, ); } catch (error) { // A worker that never received the call, a lost reply and a provider-reported @@ -159,7 +121,7 @@ export async function callWorkspaceProvider( throw workspaceProviderCallFailure(error); } // Recheck enablement and ownership after an external effect, including an ambiguous retry. - await provider(db, workers, workspace.companyId, pluginId); + await requireWorkspaceProviderWorker(db, workers, workspace.companyId, pluginId); const [fresh] = await db .select() .from(executionWorkspaces) @@ -204,6 +166,7 @@ export async function preparePluginWorkspace( baseRef?: string | null; fresh?: PluginFreshWorkspace; expectedBindingRevision?: number; + verifyExisting?: true; }, ) { const [previous] = input.existingWorkspaceId @@ -223,6 +186,8 @@ export async function preparePluginWorkspace( sql`${executionWorkspaces.metadata}#>>'{workspaceProviderFresh,attemptId}' = ${input.fresh.attemptId}`, )).limit(1) : []; const existing = input.fresh ? reservation : previous; + if (input.verifyExisting && (!existing || existing.metadata?.workspaceProviderRemoved === true || + existing.metadata?.workspaceProviderRestoration)) return null; const prior = existing ?? previous; if ( existing?.metadata?.workspaceProviderRetired === true || @@ -276,7 +241,7 @@ export async function preparePluginWorkspace( throw conflict( "An existing local workspace cannot change storage providers", ); - const selectedProvider = await provider(db, workers, input.companyId, pluginId, true); + const selectedProvider = await requireWorkspaceProviderWorker(db, workers, input.companyId, pluginId, true); if (input.environment) assertProjectStorageEnvironment(selectedProvider, input.environment); const declaration = selectedProvider.manifestJson.workspaceProvider!; if (input.fresh && declaration.freshWorkspaceVersion !== 1) @@ -430,6 +395,7 @@ export async function preparePluginWorkspace( status: "idle", metadata: { workspaceProviderOwner: pluginId, + workspaceProviderReservationRunId: input.runId, workspaceProviderSource: sourceInput, workspaceProviderBranchMode: input.fresh ? "create" : input.branchMode ?? "create", ...(input.fresh ? { workspaceProviderFresh: input.fresh } : {}), @@ -492,11 +458,19 @@ export async function preparePluginWorkspace( "create", baseRef: workspace.baseRef, existing: saved, + ...(input.verifyExisting ? { verifyExisting: true } : {}), ...(workspace.metadata?.workspaceProviderFresh ? { fresh: workspace.metadata.workspaceProviderFresh as PluginFreshWorkspace } : {}), }, }); if (result.kind !== "prepared") throw conflict("Workspace preparation is incomplete"); + const savedBase = workspace.metadata?.workspaceProviderBaseSha; + if (input.verifyExisting && (result.created || !result.baseSha || + (saved && result.binding.workspaceRef !== saved.workspaceRef) || + (workspace.branchName !== null && result.branchName !== workspace.branchName) || + (typeof savedBase === "string" && result.baseSha !== savedBase))) { + throw conflict("The existing checkout could not be verified"); + } await authorize(); const [currentProject] = await tx .select() @@ -660,7 +634,7 @@ export async function finalizePluginWorkspace(db: Db, workers: WorkspaceWorkers throw conflict("Workspace was removed before finalization"); const binding = readWorkspaceProviderBinding(current.metadata); if (!binding) throw conflict("Workspace storage ownership is missing"); - const selected = await provider(db, workers, current.companyId, binding.pluginId); + const selected = await requireWorkspaceProviderWorker(db, workers, current.companyId, binding.pluginId); // Existing providers continue to use their established inspection contract. if (selected.manifestJson.workspaceProvider?.finalizationVersion !== 1) return inspectBinding(db, workers, current); @@ -848,7 +822,7 @@ export async function removePluginWorkspace( } await beforeRemove?.(); const generation = workspace.metadata?.lifecycleGeneration ?? 0; - const selectedProvider = await provider(db, workers, workspace.companyId, binding.pluginId); + const selectedProvider = await requireWorkspaceProviderWorker(db, workers, workspace.companyId, binding.pluginId); const operation = workspaceProviderOperationSchema.parse({ kind: "remove", binding, expectedHeadSha, ...(recoveryPullRequest && selectedProvider.manifestJson.workspaceProvider?.recoveryPullRequestVersion === 1 ? { recoveryPullRequest } : {}) }); const operationId = createHash("sha256").update(JSON.stringify([workspace.id, generation, operation])).digest("hex"); diff --git a/server/src/services/project-workspace-provider.ts b/server/src/services/project-workspace-provider.ts index 946244a7545..5285e7c61ed 100644 --- a/server/src/services/project-workspace-provider.ts +++ b/server/src/services/project-workspace-provider.ts @@ -2,6 +2,8 @@ import { and, eq, notInArray } from "drizzle-orm"; import { pluginCompanySettings, plugins, type Db } from "@paperclipai/db"; import type { Environment, PaperclipPluginManifestV1, ProjectExecutionWorkspacePolicy, ProjectWorkspaceProviderConfiguration } from "@paperclipai/shared"; import { parseProjectExecutionWorkspacePolicy } from "./execution-workspace-policy.js"; +import type { PluginWorkerManager } from "./plugin-worker-manager.js"; +import { workspaceWorkerRefusal } from "./workspace-provider-failure.js"; import { conflict } from "../errors.js"; export function isWorkspaceProvider(manifest: PaperclipPluginManifestV1) { @@ -10,6 +12,45 @@ export function isWorkspaceProvider(manifest: PaperclipPluginManifestV1) { } type ProviderDb = Pick; +export type WorkspaceWorkers = Pick; + +export async function requireWorkspaceProviderWorker( + db: ProviderDb, + workers: WorkspaceWorkers | undefined, + companyId: string, + pluginId: string, + /** Before a call, a ready provider whose worker is not running is a retryable refusal: no request is sent. */ + beforeCall = false, +) { + const [plugin] = await db + .select() + .from(plugins) + .where(eq(plugins.id, pluginId)); + const [settings] = await db + .select() + .from(pluginCompanySettings) + .where( + and( + eq(pluginCompanySettings.pluginId, pluginId), + eq(pluginCompanySettings.companyId, companyId), + ), + ); + const worker = workers?.getWorker(pluginId); + const refusal = beforeCall && workers && plugin?.status === "ready" && settings?.enabled !== false && + isWorkspaceProvider(plugin.manifestJson) ? workspaceWorkerRefusal(pluginId, worker) : null; + if (refusal) throw refusal; + if ( + !workers || + !plugin || + plugin.status !== "ready" || + settings?.enabled === false || + !isWorkspaceProvider(plugin.manifestJson) || + !worker?.supportedMethods?.includes("workspaceOperation") + ) { + throw conflict("The workspace storage plugin is unavailable"); + } + return plugin; +} export async function listProjectWorkspaceProviders(db: ProviderDb, companyId: string) { const rows = await db.select({ plugin: plugins, enabled: pluginCompanySettings.enabled }) diff --git a/server/src/services/retry-workspace-continuity.test.ts b/server/src/services/retry-workspace-continuity.test.ts index 803a8828648..092ac7f224d 100644 --- a/server/src/services/retry-workspace-continuity.test.ts +++ b/server/src/services/retry-workspace-continuity.test.ts @@ -13,7 +13,7 @@ import { import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js"; import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; -import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; +import { readFreshWorkspaceReadiness, verifyRetryWorkspace } from "./fresh-workspace-continuation.js"; import { describeRetainedPriorWorkspace, parsePorcelainZ } from "./fresh-workspace-seed.js"; import { recordNativeLocalProcessStop } from "./native-local-process-stop.js"; import { createSystemRecovery, failSystemRecovery, readSystemRecovery, SystemRecoveryFailure } from "./system-recovery-state.js"; @@ -135,7 +135,7 @@ describe.skipIf(!support.supported)("Retry workspace continuity after exhausted expect(readiness).toEqual({ mode: "reuse_issue_workspace", retryIntentId: requestId, sourceRunId: f.source.id, workspaceId: f.workspace.id, cwd: f.worktree }); // The retained directory is the same worktree: git history, branch and pre-fault files are intact. - expect((await lstat(join(readiness!.mode === "reuse_issue_workspace" ? readiness!.cwd : "", ".git"))).isFile()).toBe(true); + expect((await lstat(join(readiness!.mode === "reuse_issue_workspace" ? (readiness!.cwd ?? "") : "", ".git"))).isFile()).toBe(true); expect(await git(f.worktree, "rev-parse", "--abbrev-ref", "HEAD")).toBe(f.branchName); expect(await readFile(join(f.worktree, "NOTES.md"), "utf8")).toBe("untracked pre-fault work\n"); expect(await readFile(join(f.worktree, "README.md"), "utf8")).toContain("edited before the fault"); @@ -149,6 +149,15 @@ describe.skipIf(!support.supported)("Retry workspace continuity after exhausted expect(await readFreshWorkspaceReadiness(db, { ...successor, id: randomUUID() })).toBeNull(); }, 30_000); + it("verifies a local checkout after fresh fallback without a plugin worker", async () => { + const f = await seed("blocked_without_generation"); + const { successor, requestId } = await retry(f); + expect(await verifyRetryWorkspace(db, undefined, successor, f.issueId, "board", { + mode: "fresh_isolated_workspace", retryIntentId: requestId, sourceRunId: f.source.id, retainedSourceIds: [], + })).toEqual({ mode: "reuse_issue_workspace", retryIntentId: requestId, + sourceRunId: f.source.id, workspaceId: f.workspace.id, cwd: f.worktree }); + }, 30_000); + it("lists renamed paths from either porcelain status column once", () => { expect(parsePorcelainZ("R new.md\0old.md\0 R moved.md\0before.md\0?? NOTES.md\0 M README.md\0")).toEqual([ "new.md", "moved.md", "NOTES.md", "README.md"]); diff --git a/tests/runner-e2e/source-sdk-public-inputs.json b/tests/runner-e2e/source-sdk-public-inputs.json index 0244d33d848..2cb39222091 100644 --- a/tests/runner-e2e/source-sdk-public-inputs.json +++ b/tests/runner-e2e/source-sdk-public-inputs.json @@ -31,7 +31,7 @@ "packages/plugins/sdk/src/ui/types.ts": "5a5644acf6554d2e5e1d000bfd03b97d57001eb2", "packages/plugins/sdk/src/worker-rpc-host.ts": "72f844508f695737a44d34b01b3949ec3be10aec", "packages/plugins/sdk/src/workspace-operation-error.ts": "2cd247d002d83b163239229e631bca4e519e5075", - "packages/plugins/sdk/src/workspace-provider.ts": "c212c3baffd4151010060c9ec7393bd989a9e833", + "packages/plugins/sdk/src/workspace-provider.ts": "5db136efa4bd904044113e341f443e523e336d0d", "packages/plugins/sdk/tsconfig.json": "d62ebdff94bc4b75dabf1fcad187e1e7a90fc794", "packages/shared/package.json": "6417ef8baef0cb9b836b14f453dc8bc5025709e0", "packages/shared/src/__snapshots__/slack-app-manifest.test.ts.snap": "79020ba3aa6662923f70aa44e2c483844e9901d9", @@ -405,7 +405,7 @@ "packages/shared/src/types/issue.ts": "725bd6f677b9d461460697d1ce10dee739a66ebe", "packages/shared/src/types/live.ts": "b90e55e4de8600c6fc17d7fa8d89079d4ae655ec", "packages/shared/src/types/native-finalization.ts": "c026a73003ae232fea4f7624d561493b383df358", - "packages/shared/src/types/plugin.ts": "36e0f50d97f96532cb1fbdf3223368d92a7e9b1e", + "packages/shared/src/types/plugin.ts": "016059b873c7189b8e65bb0dfa1ffa039c97c2ca", "packages/shared/src/types/project.ts": "ae2e718acd652e362cf1ef91e69aad96d12042fe", "packages/shared/src/types/quota.ts": "a0686a9afb3111a4de093f66200dc0c9152e883e", "packages/shared/src/types/resource-memberships.ts": "9eab6fad50facf3e489f8c40e5bedd92e1d1faf8", @@ -486,7 +486,7 @@ "packages/shared/src/validators/onboarding-seed.ts": "de302a8400d4199053f379d9ed853522a3ac6d32", "packages/shared/src/validators/partial.ts": "fa60ec7481fc14175365e9e500c814b159f1ab53", "packages/shared/src/validators/plugin.test.ts": "23bc443850d46d815877a71a44625fd42aebb00a", - "packages/shared/src/validators/plugin.ts": "054de474635e080073e85413211cca4b3eb8681b", + "packages/shared/src/validators/plugin.ts": "2a90d3d00b64f802b9c30f957e97702250ab2db4", "packages/shared/src/validators/project.ts": "0257acc29400f438aefc0a657250d8f28ec60838", "packages/shared/src/validators/provider-trace.ts": "b38b6200a2e51513f5dca0f8b008f2fe68175e82", "packages/shared/src/validators/resource-memberships.ts": "b124ac8a8f24b4dbb03ae82bcc87541f6b4ecdfd",