From 098f7d370bf05dffb2327b6250dbf301d5146fc6 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 8 Nov 2026 22:37:52 -0600 Subject: [PATCH] fix(workspaces): release an unchanged remote writer without a snapshot and wake its waiters (#20171) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A remote (sandbox) task workspace has one writer run at a time. The writer holds the workspace until its files are saved as a snapshot > - A run that a user comment interrupts right after its claim did no work. No provider turn started and no command ran in the sandbox > - Paperclip still kept its sandbox, then took a full snapshot of the unchanged workspace. On a 2 GB repository this took 3 minutes > - The next runs failed setup with `workspace_busy` ("ended, but its workspace save has not settled"). The retry then waited for the one-minute settling re-check after the save. The user's message waited more than 4 minutes > - This pull request releases such a writer at once, at the saved version it started from, when durable evidence proves that it changed nothing. It also wakes parked `workspace_busy` waiters when a hold settles > - The benefit is that an interrupted run no longer delays the next message by a full workspace save ## Linked Issues or Issue Description **What happened?** A remote-workspace run was claimed (`native.coordinator.claim`). One second later a board comment interrupted it (`operator_interrupted`). It made no provider turn and no tool calls. Cancelled-workspace preservation still retained its sandbox, and workspace recovery took a full `remote_workspace_snapshot` of the unchanged workspace (3 minutes for a 2 GB repository). Two follow-up runs were cancelled with `workspace_busy`. Their retry was due during the save, and it parked for the one-minute settling re-check after the save finished. **Expected behavior** A writer that provably changed nothing releases the workspace at once and keeps its existing `snapshotId`. The next run is admitted without `workspace_busy`. When a save does run, waiters start as soon as it commits. **Steps to reproduce** 1. Run a task on a remote sandbox workspace that already has a saved version. 2. Interrupt the run with a board comment right after the claim, before the provider turn. 3. See the full snapshot operation on the interrupted run and `workspace_busy` on the follow-up runs. **Paperclip version or commit** Staging served revision b43cccbb0. Reproduced on `master` by the new regression tests. ## What Changed - `prepareRemoteWorkspace` records `remoteWorkspaceUnchangedBaseline` (epoch, writer run, saved version) when preparation started from a clean saved version and added nothing to it: no recovered attachments, no repository outside the saved version and no workspace migration. - New `proveRemoteWorkspaceUnchanged` (`remote-workspace-unchanged-proof.ts`). It uses only durable evidence. The writer run must have ended without a result, and its owner fence must be clear. It must have no process fields, no execution, provider or tool events (the shared `executionStartedEventPredicate`), no native result, no runner checkpoint rows (the controller writes one before every runner command), and no provider turn in a checkpoint that names this run. It must have no workspace operation other than a restore or a `workspace_finalize`/`workspace_recovery` receipt, so provisioning and saves count as writes. Every task repository must be in the saved version. Any missing or ambiguous evidence refuses. - New `releaseUnchangedRemoteWorkspaceWriter`. It sets the authority to clean with no writer, keeps `snapshotId`, marks the run reference `saved`, logs `workspace.snapshot_skipped_unchanged` and sends the same notifications as a final save. It never waits on a writer run that another transaction holds (`skip locked`). - The release applies in three places: the next run's `reserveRemoteWorkspace`, cancelled/failed-workspace preservation (`preserveNativeFailedWorkspace`), and `saveRemoteWorkspace` with the final boundary. The final save returns the same version on a repeated call, so a recovery that raced the release finishes without a capture. - The final-save commit and the unchanged release call `wakeWorkspaceBusyWaitersInTx`. A parked waiter whose holder was that writer is due at once, not at the one-minute settling re-check. - `doc/execution-semantics.md` describes the release and the wake. ## Verification - `pnpm --filter @paperclipai/server exec vitest run src/services/remote-workspace-unchanged-release.test.ts` (13 tests, embedded Postgres, real archive capture): - Staging shape (claim, then interrupt with no provider work): the next reservation is admitted. No snapshot operation is recorded. The run reference is `saved` at the old `snapshotId`. Red on `master`: "The workspace still has an unfinished writer or save operation." - A later final save, behind the `workspace_finalize` receipt that production finalizers publish first, returns the same version with no capture, a repeat returns it again, and a parked waiter becomes due now. Red on `master`: a new snapshot is captured. - Negatives that still do a full save: `turn.started`, `provider.event`, `tool.execution.started`, `native.process_start_requested`, a provisioning command that wrote into the workspace, a recorded process, a live owner fence, and a writer that is still running. - Upgrade safety: no baseline (a writer prepared by the previous release), a baseline without its saved version, and a baseline with another schema each keep the full save. - A full final save wakes the parked waiter. Red on `master`. - `native-cancelled-workspace-preservation.test.ts`: a new case shows that a cancelled writer with a baseline and no work is released, with no retained lease and no recovery action. Red on `master` (the restore-recovery hold was recorded). The other 44 cases pass unchanged. - I also ran the related remote-workspace, workspace-recovery, native preservation and run-dispatch suites locally. ## Risks - A false "unchanged" verdict would drop unsaved files. The proof requires a positive baseline from preparation and refuses on any write record. Only `.paperclip-runtime` (excluded from snapshots) is written by runner staging before a process starts. A deploy that lands mid-run keeps the full save, because the proof reads only durable rows. - Previous-release state: a writer prepared without the baseline (or with an unknown or partial one) keeps today's full save. An older server ignores the new metadata key. - Waiters now start sooner after a final save. Arrival order among waiters is unchanged (the park check still admits the earliest first). - Other ways a hold can settle (for example a stop receipt that turns an unused reservation free) still rely on the one-minute re-check. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used in Claude Code with tool use (shell, file edits and a read-only browser for staging evidence). ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/execution-semantics.md | 19 +- ...e-cancelled-workspace-preservation.test.ts | 25 ++ .../native-failed-workspace-recovery.ts | 7 + .../remote-workspace-unchanged-proof.ts | 120 ++++++++++ ...remote-workspace-unchanged-release.test.ts | 223 ++++++++++++++++++ server/src/services/remote-workspaces.ts | 101 +++++++- 6 files changed, 488 insertions(+), 7 deletions(-) create mode 100644 server/src/services/remote-workspace-unchanged-proof.ts create mode 100644 server/src/services/remote-workspace-unchanged-release.test.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index ea215ff7344..0f5c7d37023 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -569,7 +569,24 @@ writer, measured from the time the writer ended. After the cutoff, preparation fails with the unfinished-writer error and leaves the held files unchanged. This wait is also one parked retry, for example after a reassignment stops the writer. It is re-checked every minute while the ended writer's hold settles, is -admitted once the hold releases, and stops parking at the same cutoff. +admitted once the hold releases, and stops parking at the same cutoff. The +writer's final save commit and an unchanged release (below) make these waiters +due immediately instead of at the next re-check. + +A writer that ended (failed, cancelled, interrupted or timed out) without +changing its files releases the hold at once, with no snapshot. Its workspace +keeps the saved version it started from. Preparation records that starting +point only when it began from a clean saved version and added nothing to it: +no recovered attachments, no repository outside the saved version and no +workspace migration. The release requires durable evidence that nothing ran +after that point: no execution or provider event, process, runner checkpoint, +native result or provider turn in a checkpoint for that run; no workspace +operation other than a restore or a finalize or recovery receipt; and no task +repository outside the saved version. The writer's owner fence must also be clear. Any missing or ambiguous evidence, +including a writer prepared by an earlier release without that starting point, +keeps the full save. The next run's reservation, cancelled-workspace +preservation and a later final save each apply this release. A repeated save +returns the same version. A writer stopped mid-run, for example by reassignment, keeps its unsaved files in its stopped sandbox. When the run used a reusable (warm) sandbox, the stop diff --git a/server/src/services/native-runtime/native-cancelled-workspace-preservation.test.ts b/server/src/services/native-runtime/native-cancelled-workspace-preservation.test.ts index 23b1847db48..8b72d5e589a 100644 --- a/server/src/services/native-runtime/native-cancelled-workspace-preservation.test.ts +++ b/server/src/services/native-runtime/native-cancelled-workspace-preservation.test.ts @@ -460,6 +460,31 @@ import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; } finally { await worker.stop(); } }); + it("hands the workspace on at its saved version when the stopped writer never started provider work", async () => { + const f = await seed(); + const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)); + await db.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspaceUnchangedBaseline: { + schema: "paperclip.remote-workspace-unchanged-baseline.v1", epoch: 2, runId: f.runId, snapshotId: f.reference.snapshotId, + } } }).where(eq(executionWorkspaces.id, f.workspaceId)); + const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(run.resultJson).not.toHaveProperty("workspaceRestoreRecovery"); + expect(run.runnerProfileJson?.remoteWorkspace).toMatchObject({ state: "saved", snapshotId: f.reference.snapshotId }); + const [released] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)); + const authority = readRemoteWorkspaceAuthority(released.metadata)!; + expect(authority).toMatchObject({ epoch: 2, snapshotId: f.reference.snapshotId, writerRunId: null, dirty: false }); + expect(await readRemoteWorkspaceWriterHold(db, { companyId: f.companyId, authority })).toEqual({ kind: "free" }); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)))[0]) + .toMatchObject({ status: "active", leasePolicy: "ephemeral" }); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId))).toHaveLength(0); + // A later pass finds nothing left to keep. + await preserveCancelledNativeWorkspaces(db); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId))).toHaveLength(0); + } finally { await worker.stop(); } + }); + it("rolls preservation back with its outer transaction", async () => { const f = await seed(); await expect(db.transaction(async tx => { diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts index 19b87205a18..7d2de469019 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -18,6 +18,7 @@ import { preserveRemoteWorkspacePreparationFailure } from "./native-workspace-fi import { pinFailedObservedWorkspace } from "./native-observed-workspace-preservation.js"; import { hasConfirmedSandboxStopAndRetain } from "../sandbox-stop-and-retain.js"; import { preparedRemoteWorkspaceSource } from "../remote-workspace-lifecycle.js"; +import { releaseUnchangedRemoteWorkspaceWriterById } from "../remote-workspaces.js"; const reusableLeasePolicies = ["reuse_by_environment", "reuse_by_execution_workspace"]; @@ -137,6 +138,12 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick; +type Repository = { id: string; relativePath: string; repoUrl: string; ref: string | null; pinnedCommit?: string }; + +const object = (value: unknown): Record => + value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + +/** Missing or malformed fields (an older or partial record) never prove anything. */ +export function readUnchangedBaseline(metadata: Record | null | undefined): UnchangedBaseline | null { + const value = object(metadata?.[UNCHANGED_BASELINE_KEY]); + if (value.schema !== UNCHANGED_BASELINE_SCHEMA || !Number.isSafeInteger(value.epoch) || (value.epoch as number) < 1 + || typeof value.runId !== "string" || !value.runId || typeof value.snapshotId !== "string" || !value.snapshotId + || (value.releasedAt !== undefined && typeof value.releasedAt !== "string")) return null; + return value as UnchangedBaseline; +} + +/** Operations that cannot change the saved files: a restore writes exactly the + * saved version, and finalize and recovery rows are receipts that wrap a save + * (whose own snapshot operation is still counted). */ +const FILE_NEUTRAL_OPERATION_PHASES = ["remote_workspace_restore", "workspace_finalize", "workspace_recovery"]; +/** Terminal states of a run that ended without a result. */ +const ENDED_WITHOUT_RESULT = ["failed", "cancelled", "interrupted", "timed_out"]; + +export type UnchangedProofRefusal = + | "baseline_missing" | "authority_mismatch" | "writer_not_ended" | "writer_owner_live" | "process_recorded" + | "reference_mismatch" | "provider_session_started" | "execution_recorded" | "result_recorded" + | "runner_commanded" | "workspace_operation_recorded" | "repository_unsaved"; + +/** + * Whether the ended writer of `authority` provably left its files exactly as + * the saved version it started from. The caller holds the workspace row lock. + * Any missing, ambiguous or older-release evidence refuses; a refusal keeps + * today's full save. + */ +export async function proveRemoteWorkspaceUnchanged(tx: Pick, input: { + workspace: Workspace; authority: RemoteWorkspaceAuthority; + /** The caller owns the writer's finalization, so its own fence is expected. */ + callerOwnsWriter?: boolean; + isSavedRepository: (repository: Repository) => boolean; +}): Promise<{ proven: true; baseline: UnchangedBaseline } | { proven: false; reason: UnchangedProofRefusal }> { + const { workspace, authority } = input; + const runId = authority.writerRunId; + const baseline = readUnchangedBaseline(workspace.metadata); + if (!baseline || baseline.releasedAt || !runId || baseline.runId !== runId) return { proven: false, reason: "baseline_missing" }; + if (!authority.dirty || !authority.location || authority.epoch !== baseline.epoch || authority.snapshotId !== baseline.snapshotId) + return { proven: false, reason: "authority_mismatch" }; + + const [run] = await tx.select({ status: heartbeatRuns.status, processPid: heartbeatRuns.processPid, + processGroupId: heartbeatRuns.processGroupId, processStartedAt: heartbeatRuns.processStartedAt, + sessionIdAfter: heartbeatRuns.sessionIdAfter, runnerProfileJson: heartbeatRuns.runnerProfileJson, + ownerFenceHeld: runOwnerFenceHeldSql() }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, workspace.companyId), eq(heartbeatRuns.id, runId))); + if (!run || !ENDED_WITHOUT_RESULT.includes(run.status)) return { proven: false, reason: "writer_not_ended" }; + if (!input.callerOwnsWriter && run.ownerFenceHeld) return { proven: false, reason: "writer_owner_live" }; + if (run.processPid !== null || run.processGroupId !== null || run.processStartedAt !== null || run.sessionIdAfter !== null) + return { proven: false, reason: "process_recorded" }; + const profile = object(run.runnerProfileJson); + const reference = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace); + if (profile.nativeWorkspaceSync != null || !reference.success || reference.data.state !== "prepared" + || reference.data.workspaceId !== workspace.id || reference.data.epoch !== authority.epoch + || reference.data.snapshotId !== authority.snapshotId || reference.data.leaseId !== authority.location.leaseId + || reference.data.remoteCwd !== authority.location.remoteCwd) return { proven: false, reason: "reference_mismatch" }; + // A rebound admission checkpoint may name this run; it proves nothing ran + // only while it holds no turn, result or terminal state of its own. + const checkpoint = object(profile.sessionCheckpoint); + if (object(checkpoint.identity).runId === runId && (checkpoint.activeTurnId != null || checkpoint.semanticResult != null + || checkpoint.terminal != null || (checkpoint.terminalTurns != null + && (!Array.isArray(checkpoint.terminalTurns) || checkpoint.terminalTurns.length > 0)))) + return { proven: false, reason: "provider_session_started" }; + + const [event] = await tx.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, workspace.companyId), eq(heartbeatRunEvents.runId, runId), + executionStartedEventPredicate())).limit(1); + if (event) return { proven: false, reason: "execution_recorded" }; + const [result] = await tx.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( + eq(nativeRunResults.companyId, workspace.companyId), eq(nativeRunResults.runId, runId))).limit(1); + if (result) return { proven: false, reason: "result_recorded" }; + // The controller writes this checkpoint before every runner command. + const [commanded] = await tx.select({ runId: nativeRunnerCheckpoints.runId }).from(nativeRunnerCheckpoints).where(and( + eq(nativeRunnerCheckpoints.companyId, workspace.companyId), eq(nativeRunnerCheckpoints.runId, runId))).limit(1); + if (commanded) return { proven: false, reason: "runner_commanded" }; + // Provisioning and saves record their operation before they touch files. + const [operation] = await tx.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, workspace.companyId), eq(workspaceOperations.heartbeatRunId, runId), + sql`(${workspaceOperations.executionWorkspaceId} is null or ${workspaceOperations.executionWorkspaceId} = ${workspace.id})`, + sql`${workspaceOperations.phase} not in (${sql.join(FILE_NEUTRAL_OPERATION_PHASES.map(phase => sql`${phase}`), sql`, `)})`)).limit(1); + if (operation) return { proven: false, reason: "workspace_operation_recorded" }; + // A requested repository is cloned before its save is recorded. Only + // repositories already in the saved version are proven untouched. + const repositories = await tx.select().from(executionWorkspaceRepositories).where(and( + eq(executionWorkspaceRepositories.companyId, workspace.companyId), + eq(executionWorkspaceRepositories.executionWorkspaceId, workspace.id))); + if (repositories.some(row => !row.pinnedCommit || !input.isSavedRepository({ id: row.id, relativePath: row.relativePath, + repoUrl: row.repoUrl, ref: row.requestedRef === "HEAD" ? null : row.requestedRef, pinnedCommit: row.pinnedCommit }))) + return { proven: false, reason: "repository_unsaved" }; + return { proven: true, baseline }; +} diff --git a/server/src/services/remote-workspace-unchanged-release.test.ts b/server/src/services/remote-workspace-unchanged-release.test.ts new file mode 100644 index 00000000000..d571f2ad533 --- /dev/null +++ b/server/src/services/remote-workspace-unchanged-release.test.ts @@ -0,0 +1,223 @@ +import { execFile } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { cp, mkdir, mkdtemp, realpath, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { promisify } from "node:util"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agents, closeRegisteredClients, companies, createDb, environmentLeases, environments, executionWorkspaces, + heartbeatRunEvents, heartbeatRuns, issues, workspaceOperations, +} from "@paperclipai/db"; +import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import type { EnvironmentLease } from "@paperclipai/shared"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { createStorageService } from "../storage/service.js"; +import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; +import { + prepareRemoteWorkspace, readRemoteWorkspaceAuthority, RemoteWorkspaceWriterActiveError, reserveRemoteWorkspace, saveRemoteWorkspace, +} from "./remote-workspaces.js"; +import { UNCHANGED_BASELINE_KEY } from "./remote-workspace-unchanged-proof.js"; + +const exec = promisify(execFile); +const support = await getEmbeddedPostgresTestSupport(); + +describe.skipIf(!support.supported)("remote workspace writer that ended without changing its files", () => { + let database: Awaited>; + let db: ReturnType; + let root: string; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("workspace-unchanged-release-"); + db = createDb(database.connectionString); + root = await realpath(await mkdtemp(path.join(os.tmpdir(), "workspace-unchanged-release-"))); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "63".repeat(32)); + vi.stubEnv("DATABASE_URL", database.connectionString); + }, 60_000); + afterAll(async () => { + if (database) await closeRegisteredClients(database.connectionString); + await database?.cleanup(); + if (root) await rm(root, { recursive: true, force: true }); + vi.unstubAllEnvs(); + }); + + /** A saved workspace, then a follow-up writer B prepared on the same sandbox (staging PAP-3367 shape). */ + async function preparedFollowUp(options: { runtimeProvisionCommand?: string } = {}) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Unchanged release", issuePrefix: `U${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Codie", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Daily repair", status: "in_progress", assigneeAgentId: agentId }); + const [environment] = await db.insert(environments).values({ name: issueId, driver: "sandbox" }).returning(); + const storage = { ...createStorageService(createLocalDiskStorageProvider(path.join(root, issueId, "objects"))), provider: "s3" as const }; + const remoteCwd = path.join(root, issueId, "remote"); + await mkdir(remoteCwd, { recursive: true }); + const providerLeaseId = randomUUID(); + const transfer: NonNullable["syncIn"]> = async operations => { + for (const operation of operations) for (const file of operation.files) { + await mkdir(path.dirname(file.targetPath), { recursive: true }); + await cp(file.sourcePath, file.targetPath, { recursive: file.kind === "directory" }); + } + return { operations: operations.map(operation => ({ operationId: operation.operationId, + filesTransferred: operation.files.length, bytesTransferred: 0 })) }; + }; + let captures = 0; + const target = (leaseId: string): AdapterSandboxExecutionTarget => ({ + kind: "remote", transport: "sandbox", leaseId, remoteCwd, timeoutMs: 30_000, + runner: { syncIn: transfer, syncOut: async operations => { captures += 1; return transfer(operations); }, + execute: async input => { + const result = await exec(input.command, input.args ?? [], { cwd: input.cwd ?? remoteCwd, + env: { ...process.env, ...input.env }, timeout: input.timeoutMs, maxBuffer: 8 * 1024 * 1024 }); + return { exitCode: 0, stdout: result.stdout, stderr: result.stderr, signal: null, timedOut: false, + pid: process.pid, startedAt: new Date().toISOString() }; + } }, + }); + const writer = async (runId: string, existingWorkspaceId: string | null, runtimeProvisionCommand?: string) => { + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, runtimeMode: "legacy", + status: "running", contextSnapshot: { issueId } }); + const [lease] = await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: runId, + environmentId: environment.id, provider: "daytona", providerLeaseId, status: "active" }).returning(); + const reserved = await reserveRemoteWorkspace({ db, storage, companyId, issueId, runId, agentId, + environmentId: environment.id, projectId: null, projectWorkspaceId: null, existingWorkspaceId, source: { kind: "empty" } }); + const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId, + lease: lease as EnvironmentLease, target: target(lease.id), runtimeProvisionCommand }); + return { lease: prepared.lease, workspaceId: reserved.id, remoteRoot: prepared.remoteRoot }; + }; + // Run A does real work and saves it. + const runA = randomUUID(); + const a = await writer(runA, null); + await writeFile(path.join(a.remoteRoot, "report.md"), "earlier work\n"); + const [workspaceA] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, a.workspaceId)); + const savedA = await saveRemoteWorkspace({ db, storage, workspace: workspaceA, runId: runA, lease: a.lease, + target: target(a.lease.id), assertOwned: async () => undefined }); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runA)); + // Run B is the follow-up: claimed, then interrupted before any provider work. + const runB = randomUUID(); + const b = await writer(runB, a.workspaceId, options.runtimeProvisionCommand); + const end = async (patch: Partial = {}) => { + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), errorCode: "operator_interrupted", ...patch }) + .where(eq(heartbeatRuns.id, runB)); + }; + const workspace = async () => (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, a.workspaceId)))[0]!; + const saveB = async () => saveRemoteWorkspace({ db, storage, workspace: await workspace(), runId: runB, lease: b.lease, + target: target(b.lease.id), assertOwned: async () => undefined }); + const snapshotOperations = () => db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.heartbeatRunId, runB), eq(workspaceOperations.phase, "remote_workspace_snapshot"))); + const reserveNext = async () => { + const runC = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runC, companyId, agentId, issueId, runtimeMode: "legacy", + status: "running", contextSnapshot: { issueId } }); + return { runC, reserved: await reserveRemoteWorkspace({ db, storage, companyId, issueId, runId: runC, agentId, + environmentId: environment.id, projectId: null, projectWorkspaceId: null, existingWorkspaceId: a.workspaceId, source: { kind: "empty" } }) }; + }; + const parkWaiter = async () => { + const waiterId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: waiterId, companyId, agentId, issueId, runtimeMode: "legacy", + status: "scheduled_retry", scheduledRetryReason: "workspace_busy", scheduledRetryAt: new Date(Date.now() + 5 * 60_000), + contextSnapshot: { issueId, workspaceBusyWait: { key: `execution:${a.workspaceId}`, holderRunId: runB, + waitingSince: new Date().toISOString(), issueExecutionWorkspaceId: a.workspaceId, issueProjectWorkspaceId: null } } }); + return async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, waiterId)))[0]!.scheduledRetryAt!; + }; + const event = (eventType: string) => db.insert(heartbeatRunEvents).values({ companyId, runId: runB, agentId, + seq: 1, stream: "system", level: "info", eventType }); + return { companyId, runA, runB, savedA, workspace, end, saveB, snapshotOperations, reserveNext, parkWaiter, event, + captures: () => captures }; + } + + async function expectFullSave(f: Awaited>) { + await expect(f.reserveNext()).rejects.toBeInstanceOf(RemoteWorkspaceWriterActiveError); + const before = f.captures(); + const saved = await f.saveB(); + expect(saved).not.toHaveProperty("unchanged"); + expect(saved.snapshotId).not.toBe(f.savedA.snapshotId); + expect(f.captures()).toBeGreaterThan(before); + expect(await f.snapshotOperations()).toHaveLength(1); + } + + it("admits the next run at once, without a save, when the interrupted writer never started provider work", async () => { + const f = await preparedFollowUp(); + await f.end(); + const { runC, reserved } = await f.reserveNext(); + const authority = readRemoteWorkspaceAuthority(reserved.metadata); + expect(authority).toMatchObject({ writerRunId: runC, snapshotId: f.savedA.snapshotId, dirty: false, epoch: 3 }); + expect(await f.snapshotOperations()).toEqual([]); + const [runB] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runB)); + expect(runB.runnerProfileJson?.remoteWorkspace).toMatchObject({ state: "saved", snapshotId: f.savedA.snapshotId, epoch: 2 }); + }); + + it("settles a later save of that writer without capturing, and wakes runs parked behind it", async () => { + const f = await preparedFollowUp(); + await f.end(); + const dueAt = await f.parkWaiter(); + // Production finalizers publish their wrapper receipt before the save. + await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: (await f.workspace()).id, + heartbeatRunId: f.runB, phase: "workspace_finalize", status: "running", metadata: { owningService: "native_workspace_finalizer" } }); + const before = f.captures(); + const saved = await f.saveB(); + expect(saved).toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true }); + expect(f.captures()).toBe(before); + expect((await dueAt()).getTime()).toBeLessThanOrEqual(Date.now()); + // A repeated save (a recovery retry) returns the same version. + expect(await f.saveB()).toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true }); + expect(await f.snapshotOperations()).toEqual([]); + const { runC } = await f.reserveNext(); + expect(readRemoteWorkspaceAuthority((await f.workspace()).metadata)?.writerRunId).toBe(runC); + }); + + it.each(["turn.started", "provider.event", "tool.execution.started", "native.process_start_requested"])( + "keeps the full save after a %s event", async eventType => { + const f = await preparedFollowUp(); + await f.event(eventType); + await f.end(); + await expectFullSave(f); + }); + + it("keeps the full save when setup wrote into the workspace", async () => { + const f = await preparedFollowUp({ runtimeProvisionCommand: "echo provisioned > provisioned.txt" }); + await f.end(); + await expectFullSave(f); + }); + + it("keeps the full save while the writer's owner may still act, and while it is still running", async () => { + const f = await preparedFollowUp(); + await f.end({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }); + await expect(f.reserveNext()).rejects.toBeInstanceOf(RemoteWorkspaceWriterActiveError); + await db.update(heartbeatRuns).set({ status: "running", controllerLeaseExpiresAt: null, finishedAt: null }) + .where(eq(heartbeatRuns.id, f.runB)); + const saved = await f.saveB(); + expect(saved).not.toHaveProperty("unchanged"); + expect(await f.snapshotOperations()).toHaveLength(1); + }); + + it("keeps the full save when a process was recorded", async () => { + const f = await preparedFollowUp(); + await f.end({ processPid: 4242, processStartedAt: new Date() }); + await expectFullSave(f); + }); + + it("wakes parked runs when a full final save settles", async () => { + const f = await preparedFollowUp(); + await f.event("turn.started"); + await f.end(); + const dueAt = await f.parkWaiter(); + await f.saveB(); + expect((await dueAt()).getTime()).toBeLessThanOrEqual(Date.now()); + }); + + it.each([ + { label: "a writer prepared by the previous release (no baseline)", baseline: undefined }, + { label: "a baseline missing its saved version", baseline: "drop_snapshot" }, + { label: "a baseline from another schema", baseline: "other_schema" }, + ] as const)("keeps the full save for $label", async ({ baseline }) => { + const f = await preparedFollowUp(); + const current = await f.workspace(); + const recorded = current.metadata?.[UNCHANGED_BASELINE_KEY] as Record | undefined; + expect(recorded).toMatchObject({ runId: f.runB, epoch: 2, snapshotId: f.savedA.snapshotId }); + const { [UNCHANGED_BASELINE_KEY]: _dropped, ...previousRelease } = current.metadata ?? {}; + const next = baseline === undefined ? previousRelease + : baseline === "drop_snapshot" ? { ...previousRelease, [UNCHANGED_BASELINE_KEY]: { ...recorded, snapshotId: undefined } } + : { ...previousRelease, [UNCHANGED_BASELINE_KEY]: { ...recorded, schema: "paperclip.remote-workspace-unchanged-baseline.v0" } }; + await db.update(executionWorkspaces).set({ metadata: next }).where(eq(executionWorkspaces.id, current.id)); + await f.end(); + await expectFullSave(f); + }); +}); diff --git a/server/src/services/remote-workspaces.ts b/server/src/services/remote-workspaces.ts index e45cbb99b55..f61475bce95 100644 --- a/server/src/services/remote-workspaces.ts +++ b/server/src/services/remote-workspaces.ts @@ -42,6 +42,9 @@ import { attachPreservedRemoteWorkspaceSnapshots, retainRemoteRecoveryAttachment import { notifyRemoteWorkspaceSnapshots } from "./remote-workspace-snapshot-retention.js"; import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; import { preserveRemoteWorkspacePreparationFailure } from "./native-runtime/native-workspace-finalization-state.js"; +import { proveRemoteWorkspaceUnchanged, readUnchangedBaseline, UNCHANGED_BASELINE_KEY, UNCHANGED_BASELINE_SCHEMA, + type UnchangedBaseline } from "./remote-workspace-unchanged-proof.js"; +import { wakeWorkspaceBusyWaitersInTx } from "../modules/run-dispatch/index.js"; export { readRemoteWorkspaceAuthority, type RemoteWorkspaceAuthority, type RemoteWorkspaceSource } from "@paperclipai/shared"; export { remoteWorkspaceRepositoryPath } from "./remote-workspace-transport.js"; @@ -206,9 +209,9 @@ export async function reserveRemoteWorkspace(input: { selectedSharedSource = selected !== null; } const [selectedRow] = selected ? await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, selected))).for("update") : []; - const existing = selectedRow ?? null; + let existing = selectedRow ?? null; if (selected && !existing && !input.fresh) throw new Error("remote_workspace_missing"); - const authority = readRemoteWorkspaceAuthority(existing?.metadata); + let authority = readRemoteWorkspaceAuthority(existing?.metadata); if (existing && !authority) throw conflict("Import this existing workspace before enabling remote storage.", { code: "remote_workspace_migration_required" }); if (existing && authority) { if (["archived", "cleanup_failed"].includes(existing.status)) throw conflict("Workspace ownership changed."); @@ -236,6 +239,12 @@ export async function reserveRemoteWorkspace(input: { await settleUnverifiedProvision(tx as unknown as Db, operation, input.runId); provenProvisions.push(operation.id); } + // An ended writer that provably changed nothing hands the workspace on + // at its saved version, so this run is admitted instead of waiting. + if (authority.dirty && authority.writerRunId && authority.writerRunId !== input.runId) { + const released = await releaseUnchangedRemoteWorkspaceWriter(tx, { workspace: existing, authority, wakeWaiters: false }); + if (released) ({ workspace: existing, authority } = released); + } const hold = await readRemoteWorkspaceWriterHold(tx as unknown as Db, { companyId: input.companyId, authority, runId: input.runId }); if (hold.kind === "held") throw new RemoteWorkspaceWriterActiveError(existing.id, hold.writer); if (hold.kind === "unattributed") throw conflict("The workspace still has an unfinished writer or save operation.", { code: "remote_workspace_writer_pending" }); @@ -551,6 +560,13 @@ export async function prepareRemoteWorkspace(input: { || current.location.remoteCwd !== remoteRoot)) throw new Error("remote_workspace_authority_changed"); const authority: RemoteWorkspaceAuthority = { ...current, dirty: true, location: { environmentId: input.lease.environmentId!, leaseId: input.lease.id, providerLeaseId: input.lease.providerLeaseId!, remoteCwd: remoteRoot } }; + // Files published here equal the saved version only when this preparation + // started from a clean saved version and added nothing to it. Later writes + // (provisioning, runner commands, saves) leave their own durable records. + const unchangedBaseline: UnchangedBaseline | undefined = initial.snapshotId && !initial.dirty && migration === null + && current.snapshotId === initial.snapshotId && recoveredFiles.length === 0 + && ((input.source ?? initial.source).repositories ?? []).every(repository => isSavedWorkspaceRepository(input.workspace, initial, repository)) + ? { schema: UNCHANGED_BASELINE_SCHEMA, epoch: authority.epoch, runId: input.runId, snapshotId: initial.snapshotId } : undefined; const [lease] = await tx.select().from(environmentLeases).where(and(eq(environmentLeases.id, input.lease.id), eq(environmentLeases.companyId, workspace.companyId))).for("update"); if (!lease || lease.status !== "active" || lease.environmentId !== input.lease.environmentId @@ -559,7 +575,7 @@ export async function prepareRemoteWorkspace(input: { const [savedLease] = await tx.update(environmentLeases).set({ metadata: { ...lease.metadata, remoteCwd: remoteRoot, remoteWorkspaceId: workspace.id }, updatedAt: new Date() }) .where(eq(environmentLeases.id, lease.id)).returning(); const [savedWorkspace] = await tx.update(executionWorkspaces).set({ cwd: remoteRoot, providerRef: remoteRoot, - metadata: { ...workspace.metadata, remoteWorkspace: authority, + metadata: { ...workspace.metadata, remoteWorkspace: authority, [UNCHANGED_BASELINE_KEY]: unchangedBaseline, // A live copy names the run it is shown to; the next preparation retires // it once that run has succeeded. Retired records stay so a sandbox // restored from an older save is cleaned up too. @@ -700,6 +716,57 @@ async function settleSaveInvocation(input: { db: Db; workspace: Workspace; runId }); } +/** + * Hand a workspace on at its saved version when its writer ended without + * changing it (see proveRemoteWorkspaceUnchanged). The caller holds the + * workspace row lock. Returns null, changing nothing, when the proof fails or + * the writer run is locked by another transaction; its full save still applies. + */ +export async function releaseUnchangedRemoteWorkspaceWriter(tx: Transaction, input: { + workspace: typeof executionWorkspaces.$inferSelect; authority: RemoteWorkspaceAuthority; + callerOwnsWriter?: boolean; wakeWaiters?: boolean; +}): Promise<{ workspace: typeof executionWorkspaces.$inferSelect; authority: RemoteWorkspaceAuthority } | null> { + const { workspace, authority } = input; + const runId = authority.writerRunId; + if (!runId || !authority.dirty || !authority.snapshotId || !readUnchangedBaseline(workspace.metadata)) return null; + // Never wait on a writer run another transaction holds; that owner settles it. + const [writer] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, workspace.companyId), eq(heartbeatRuns.id, runId))).for("update", { skipLocked: true }); + if (!writer) return null; + const proof = await proveRemoteWorkspaceUnchanged(tx as unknown as Db, { workspace, authority, + callerOwnsWriter: input.callerOwnsWriter, isSavedRepository: repository => isSavedWorkspaceRepository(workspace, authority, repository) }); + if (!proof.proven) { + logger.info({ runId, workspaceId: workspace.id, reason: proof.reason }, "Remote workspace keeps its full save"); + return null; + } + const at = new Date(); + const released: RemoteWorkspaceAuthority = { ...authority, dirty: false, writerRunId: null }; + const [saved] = await tx.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspace: released, + [UNCHANGED_BASELINE_KEY]: { ...proof.baseline, releasedAt: at.toISOString() } }, updatedAt: at }) + .where(eq(executionWorkspaces.id, workspace.id)).returning(); + await persistRunReference(tx, workspace.companyId, runId, reference(workspace.id, released, "saved")); + await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`); + await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement); + if (input.wakeWaiters !== false) await wakeWorkspaceBusyWaitersInTx(tx as unknown as Db, { companyId: workspace.companyId, + runId, contextSnapshot: null, scheduledRetryReason: null, now: at }); + await persistActivity(tx as unknown as Db, { companyId: workspace.companyId, actorType: "system", actorId: "remote-workspaces", + action: "workspace.snapshot_skipped_unchanged", entityType: "execution_workspace", entityId: workspace.id, runId, + details: { snapshotId: authority.snapshotId, epoch: authority.epoch } }); + return { workspace: saved!, authority: released }; +} + +/** Locks the workspace and releases `runId` as its unchanged ended writer, if provable. */ +export async function releaseUnchangedRemoteWorkspaceWriterById(tx: Transaction, input: { + companyId: string; workspaceId: string; runId: string; +}) { + const [workspace] = await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), + eq(executionWorkspaces.id, input.workspaceId))).for("update"); + const authority = readRemoteWorkspaceAuthority(workspace?.metadata); + if (!workspace || !authority || authority.writerRunId !== input.runId) return null; + return releaseUnchangedRemoteWorkspaceWriter(tx, { workspace, authority }); +} + export async function saveRemoteWorkspace(input: { db: Db; workspace: Workspace; runId: string; lease: EnvironmentLease; target: AdapterExecutionTarget | null; assertOwned: () => Promise; storage?: StorageService; boundary?: string; @@ -711,11 +778,28 @@ export async function saveRemoteWorkspace(input: { if (!expected?.location || expected.location.providerLeaseId !== input.lease.providerLeaseId || expected.location.leaseId !== input.lease.id) throw new Error("remote_workspace_save_binding_mismatch"); await input.assertOwned(); const invocationId = randomUUID(); - const operation = await input.db.transaction(async tx => { + const reserved = await input.db.transaction(async tx => { const issueId = await lockWorkspaceRunParents(tx, input.workspace.companyId, input.runId); const workspace = await currentWorkspace(tx, input.workspace.companyId, input.workspace.id); const current = readRemoteWorkspaceAuthority(workspace.metadata); if (!current || current.epoch !== expected.epoch || (current.writerRunId !== input.runId && current.dirty)) throw new Error("remote_workspace_authority_changed"); + // A writer that ended without changing its files keeps its saved version. + // No capture runs; a repeated call returns the same version. + if (boundary === "final") { + const baseline = readUnchangedBaseline(workspace.metadata); + const replayed = baseline?.releasedAt && baseline.runId === input.runId && baseline.epoch === current.epoch + && baseline.snapshotId === current.snapshotId && !current.dirty && current.writerRunId === null; + const released = replayed ? { authority: current } : current.writerRunId === input.runId + ? await releaseUnchangedRemoteWorkspaceWriter(tx, { workspace, authority: current, callerOwnsWriter: true }) : null; + if (released?.authority.snapshotId) { + const [snapshot] = await tx.select().from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), + eq(executionWorkspaceSnapshots.id, released.authority.snapshotId))); + if (!snapshot?.sha256) throw new Error("remote_workspace_snapshot_missing"); + return { kind: "unchanged" as const, saved: { snapshotId: snapshot.id, generation: snapshot.generation, sha256: snapshot.sha256, + savedAt: snapshot.updatedAt.toISOString(), unchanged: true as const } }; + } + } const [previous] = await tx.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, workspace.companyId), eq(workspaceOperations.executionWorkspaceId, workspace.id), eq(workspaceOperations.heartbeatRunId, input.runId), eq(workspaceOperations.phase, "remote_workspace_snapshot"), sql`${workspaceOperations.metadata}->>'boundary' = ${boundary}`, @@ -729,7 +813,7 @@ export async function saveRemoteWorkspace(input: { const [registered] = await tx.update(workspaceOperations).set({ ...reopened, metadata: { ...previous.metadata, saveWork: { schema: SAVE_WORK_SCHEMA, activeIds: [...(work?.activeIds ?? []), invocationId], unknown: work?.unknown ?? true, quiescedAt: null } }, updatedAt: new Date() }).where(eq(workspaceOperations.id, previous.id)).returning(); - return registered!; + return { kind: "registered" as const, operation: registered! }; } // The workspace lock serializes generation reservations. Failed/deleted // snapshots still own their generation, and an operation can be committed @@ -753,8 +837,10 @@ export async function saveRemoteWorkspace(input: { saveWork: { schema: SAVE_WORK_SCHEMA, activeIds: [invocationId], unknown: false, quiescedAt: null }, exportCleanup: { ...current.location, state: "pending" } }, }).returning(); - return created!; + return { kind: "registered" as const, operation: created! }; }); + if (reserved.kind === "unchanged") return reserved.saved; + const { operation } = reserved; let captured: Awaited> | null = null; let publicationScratch: string | undefined; let failure: { code: string; cause?: string } | undefined; @@ -809,6 +895,9 @@ export async function saveRemoteWorkspace(input: { if (boundary === "final") { await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`); await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement); + // The hold ends with this commit; runs parked behind it start now. + await wakeWorkspaceBusyWaitersInTx(tx as unknown as Db, { companyId: workspace.companyId, runId: input.runId, + contextSnapshot: null, scheduledRetryReason: null, now: at }); } const receipt = object(workspace.metadata?.remoteWorkspaceSnapshotRepositories); const priorRepositories = receipt.snapshotId === current.snapshotId && Array.isArray(receipt.repositories)