fix(workspaces): verify existing worktrees before fresh retries (#20876)

## Thinking Path

> - Paperclip manages work for AI agents.
> - Retry checks whether the task can keep its existing workspace.
> - A provider can complete checkout before Core receives the result.
> - Recreating a completed worktree costs time and can change its base
commit.
> - Core uses the same ownership and writer checks for local and
provider workspaces.
> - The provider verifies its checkout before Core creates a fresh
workspace.

## Linked Issues or Issue Description

**What happened?**

A retry can allocate a new provider worktree even though the original
checkout completed.

**Steps to reproduce**

Complete workspace checkout through a storage provider. Fail Runner
startup or lose the prepare reply. Retry the task.

**Expected behavior**

Reuse the same worktree when the provider confirms checkout and the
prior run can no longer write to it.

Related: #20865 addresses timeouts while the original prepare is in
progress. This PR checks an earlier checkout when the user retries.

## What Changed

- Check the prior task binding and writer state before asking a storage
provider to verify a checkout.
- Extend the existing prepare operation with an optional verification
mode. Providers opt in through their manifest.
- Return through normal workspace reuse after verification. Keep the
fresh retry fallback when verification cannot establish checkout.

## Verification

- Core workspace preparation, plugin workspace ownership, retry
continuity, and SDK workspace-provider suites passed locally (213
tests).
- Focused cases pass for provider verification, retry through heartbeat
dispatch, and real local Git checkout verification with no plugin
worker. The heartbeat test confirms that ordinary preparation and
adapter dispatch keep the original workspace, branch, and base commit.
- Core server typecheck passed before the final audit and timing edits.
GitHub CI covers the final revision and broader regressions.

## Risks

Provider verification adds one bounded call before a fresh retry. Older
providers keep their current behavior. Verification must preserve the
workspace, branch, and recorded base commit. The normal execution
preparation checks still run before admission. No database migration is
needed.

## Model Used

OpenAI GPT-6 (Codex), with source inspection and local test execution.

## 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 linked existing issues or described the issue in-PR
following the relevant issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links
- [x] My branch name describes the change and contains no internal issue
details
- [x] I have run targeted 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

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

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<PluginWorkerManager, "call" | "getWorker">;
+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");