diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 6d50ff984fd..538c675cd66 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -502,6 +502,19 @@ 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. +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 +leaves that lease `released` under its reusable policy with a stop-and-retain +receipt. Cancelled-workspace preservation admits that stopped source, as it does +an active or ephemeral one. It records the save obligation for the exact writer +run while the workspace is still bound to that writer. Paperclip's workspace +recovery then saves those files, which releases the hold. It never replays the +stopped run. While that save is in progress, its last progress counts as the +ended writer's activity for the holder cutoff. A waiter keeps waiting instead of +failing setup. The save's retry budget bounds that wait. A save that exhausts +its retries is handed to the board as a named recovery action. After that hand-off, +the cutoff no longer advances. + An instance admission drain keeps accepted new turns queued and displays **Waiting for maintenance**. Ending the audited drain requests admission for these existing runs immediately. The shared delivery coordinator owns the drain's diff --git a/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts b/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts index b0d106027f1..7a404ad6e36 100644 --- a/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts +++ b/server/src/__tests__/heartbeat-remote-workspace-busy.test.ts @@ -5,7 +5,7 @@ import { join } from "node:path"; import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, authUsers, companies, companyMemberships, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, - environmentLeases, environments, executionWorkspaces, heartbeatRuns, issues, type Db } from "@paperclipai/db"; + environmentLeases, environments, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, type Db } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { heartbeatService } from "../services/heartbeat.js"; @@ -213,6 +213,49 @@ describe.skipIf(!support.supported)("inherited remote workspace heartbeat wait", await heartbeat.drainActiveRunExecutions(); }); + // A writer stopped by reassignment whose retained files Paperclip is still + // saving (an in-progress workspace recovery) is still settling, even after the + // one-hour cutoff from its end. A waiter already parked for it stays parked + // instead of being released into a setup failure; the save's own retry budget + // bounds the wait. A finished or board-handed-off save no longer extends it. + it.each([ + { state: undefined, waits: true }, + { state: "scheduled", waits: true }, + { state: "awaiting_evidence", waits: true }, + { state: "exhausted", waits: false }, + { state: "complete", waits: false }, + ] as const)("keeps a waiter parked while an ended writer's save is in progress ($state)", async ({ state, waits }) => { + const f = await terminalWriterFixture(new Date()); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.run.id)); + expect(retry).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "workspace_busy" }); + // The writer ended over an hour ago; its save recovery recorded progress a minute ago. + const endedAt = new Date(Date.now() - 2 * 60 * 60_000); + await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "issue_reassigned", finishedAt: endedAt, + startedAt: new Date(endedAt.getTime() - 60_000), lastOutputAt: endedAt }).where(eq(heartbeatRuns.id, f.writer.id)); + await db.insert(issueRecoveryActions).values({ companyId: f.run.companyId, sourceIssueId: f.workspace.sourceIssueId!, + kind: "active_run_watchdog", status: "resolved", outcome: "blocked", ownerType: "board", + cause: "native_continuation_requires_reconciliation", fingerprint: `native-workspace:${f.writer.id}`, + resolvedAt: new Date(), updatedAt: new Date(Date.now() - 60_000), nextAction: "Save the retained files.", + evidence: { runId: f.writer.id, workspaceRestoreFailure: "restore_failed", + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [randomUUID()] }, + automaticRecovery: { policy: "preserve_without_replay_v1", runId: f.writer.id, replay: "blocked" }, + ...(state ? { systemRecovery: { state } } : {}) } }); + const checkAt = new Date(retry.scheduledRetryAt!.getTime() + 1_000); + const promoted = (await heartbeat.promoteDueScheduledRetries(checkAt)).runIds; + if (waits) { + expect(promoted).not.toContain(retry.id); + const parked = await heartbeat.getRun(retry.id); + expect(parked?.status).toBe("scheduled_retry"); + // Bounded by the cutoff measured from the save's last progress. + expect(new Date(parked!.scheduledRetryAt!).getTime()).toBeLessThanOrEqual(Date.now() + 60 * 60_000); + } else expect(promoted).toContain(retry.id); + await heartbeat.cancelRun(retry.id); + await heartbeat.drainActiveRunExecutions(); + // Waiting never takes over or discards the unsaved writer's files. + expect(readRemoteWorkspaceAuthority((await db.select().from(executionWorkspaces) + .where(eq(executionWorkspaces.id, f.workspace.id)))[0].metadata)).toEqual(f.authority); + }); + it("stops waiting for an ended writer whose save stayed unsettled past the holder cutoff", async () => { const f = await terminalWriterFixture(new Date(Date.now() - 2 * 60 * 60_000)); expect(f.run).toMatchObject({ status: "failed", error: "The workspace still has an unfinished writer or save operation." }); 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 0916bd8c90f..91579705ac4 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 @@ -10,6 +10,9 @@ import { requestWorkspaceRecovery } from "../workspace-recovery.js"; import { inspectWorkspaceRecoverySource } from "../workspace-recovery-source.js"; import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace, registerCancelledWorkspacePreservationWork } from "./native-failed-workspace-recovery.js"; +import { prepareSandboxStopAndRetain, settleStopOnlyCleanup } from "../sandbox-stop-and-retain.js"; +import { readRemoteWorkspaceWriterHold } from "../remote-workspace-writer-hold.js"; +import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; (recoveryPostgresSupport.supported ? describe : describe.skip)("lifecycle-cancelled workspace preservation", () => { let database: Awaited>; @@ -158,6 +161,79 @@ import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace, } finally { await worker.stop(); } }, 120_000); + // A warm (reusable) run that ends without success stops its sandbox and keeps + // the lease `released` under its reusable policy. Reassignment Stop reaches + // that state with the workspace still dirty and bound to the stopped writer. + async function stopReusableSource(f: Awaited>) { + const [reusable] = await db.update(environmentLeases).set({ leasePolicy: "reuse_by_environment" }) + .where(eq(environmentLeases.id, f.leaseId)).returning(); + const stopping = (await prepareSandboxStopAndRetain(db, reusable))!; + const stopped = (await settleStopOnlyCleanup(db, stopping, { + attemptId: String(stopping.metadata!.pendingCleanupAttemptId), + receipt: { providerLeaseId: f.reference.providerLeaseId, state: "stopped" }, + }))!; + expect(stopped).toMatchObject({ status: "released", leasePolicy: "reuse_by_environment", cleanupStatus: "success" }); + return stopped; + } + + it.each(["worker", "startup-scan"] as const)("preserves a reassignment-stopped writer's files from its stopped reusable sandbox (%s)", async kind => { + const f = await seed(); + const resultJson = { ...f.run.resultJson, reassignmentStopRequested: true }; + await db.update(heartbeatRuns).set({ errorCode: "issue_reassigned", resultJson }).where(eq(heartbeatRuns.id, f.runId)); + const stopped = await stopReusableSource(f); + const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + if (kind === "startup-scan") await preserveCancelledNativeWorkspaces(db); + else await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + const [retained] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(retained.resultJson).toMatchObject({ ...resultJson, workspaceRestoreFailure: "restore_failed", + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.leaseId] } }); + expect(retained).toMatchObject({ status: "cancelled", errorCode: "issue_reassigned", finishedAt: f.run.finishedAt }); + // The stop receipt and reusable policy stay exactly as the release left them. + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)))[0]).toEqual(stopped); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId)); + expect(action).toMatchObject({ ownerType: "board", evidence: { runId: f.runId, + automaticRecovery: { replay: "blocked" }, workspaceRestoreRecovery: { leaseIds: [f.leaseId] } } }); + expect((await inspectWorkspaceRecoverySource(db, { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, actionId: action.id })).sources).toHaveLength(1); + // Preservation records the save obligation only. The stopped writer keeps + // the workspace until the saved files are published; nothing replays it. + const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)); + expect(await readRemoteWorkspaceWriterHold(db, { companyId: f.companyId, + authority: readRemoteWorkspaceAuthority(workspace.metadata)! })).toMatchObject({ kind: "held", writer: { runId: f.runId, ended: true } }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + expect(worker.nextWakeAt()).toBeNull(); + } finally { await worker.stop(); } + }, 120_000); + + it.each(["unconfirmed-stop", "writer-released", "workspace-saved"] as const)("does not retain a reusable source (%s)", async kind => { + const f = await seed(); + let released: typeof environmentLeases.$inferSelect; + if (kind === "unconfirmed-stop") { + [released] = await db.update(environmentLeases).set({ leasePolicy: "reuse_by_environment", status: "released", + cleanupStatus: "success", releasedAt: new Date() }).where(eq(environmentLeases.id, f.leaseId)).returning(); + } else { + released = await stopReusableSource(f); + // The workspace moved on: another writer owns it, or the writer's save already published. + const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)); + const authority = readRemoteWorkspaceAuthority(workspace.metadata)!; + await db.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspace: kind === "writer-released" + ? { ...authority, writerRunId: randomUUID() } : { ...authority, writerRunId: null, dirty: false } } }) + .where(eq(executionWorkspaces.id, f.workspaceId)); + } + const before = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]; + await preserveCancelledNativeWorkspaces(db); + const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; } + finally { await worker.stop(); } + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(before); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(run.resultJson).not.toHaveProperty("workspaceRestoreRecovery"); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)))[0]).toEqual(released); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId))).toHaveLength(0); + }); + it.each(["pending", "missing-dispatch", "invalid-dispatch", "foreign-company", "foreign-issue", "foreign-run", "turn-scope", "empty-intent-audit", "missing-ack-audit", "empty-ack-audit", "live-owner", "live-controller", "accepted-result", "foreign-allocation", "foreign-workspace", "foreign-cwd", "saved-reference", "unknown-reference", "running"] as const) 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 9681695bef3..1c95720902a 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -16,6 +16,10 @@ import { logger } from "../../middleware/logger.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { preserveRemoteWorkspacePreparationFailure } from "./native-workspace-finalization-state.js"; import { pinFailedObservedWorkspace } from "./native-observed-workspace-preservation.js"; +import { hasConfirmedSandboxStopAndRetain } from "../sandbox-stop-and-retain.js"; +import { preparedRemoteWorkspaceSource } from "../remote-workspace-lifecycle.js"; + +const reusableLeasePolicies = ["reuse_by_environment", "reuse_by_execution_workspace"]; function workspaceRecoveryReference(profile: Record | null | undefined) { @@ -93,6 +97,10 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick(); +/** Leases that may still hold a cancelled writer's prepared files. A reusable + * (warm) lease that ends without success is stopped and kept `released` under + * its reusable policy; a reassignment Stop leaves the writer's unsaved files + * there. Admission only finds candidates: retention still requires the exact + * stop-and-retain receipt and the workspace still bound to this writer. */ +function cancelledWorkspaceSourceLease() { + return or(eq(environmentLeases.status, "active"), + and(inArray(environmentLeases.status, ["pending_cleanup", "released"]), eq(environmentLeases.leasePolicy, "ephemeral")), + and(eq(environmentLeases.status, "released"), eq(environmentLeases.cleanupStatus, "success"), + inArray(environmentLeases.leasePolicy, reusableLeasePolicies), + sql`${environmentLeases.metadata}->'remoteExecutionTermination'->>'state' = 'stopped'`, + sql`${environmentLeases.metadata} ? 'sandboxStopAndRetainReceipt'`)); +} + /** Older controllers could acknowledge Stop without pinning prepared files. * Record only the file obligation; the existing stop-only cleanup worker owns * provider teardown. This never resumes a turn or changes the task's status. */ @@ -178,9 +200,7 @@ export async function preserveCancelledNativeWorkspaces(db: Db) { eq(heartbeatRuns.id, environmentLeases.heartbeatRunId), eq(heartbeatRuns.companyId, environmentLeases.companyId), )).innerJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.runId, heartbeatRuns.id), eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId), eq(nativeRunFinalizations.issueId, heartbeatRuns.nativeIssueId))) - .where(and(or(eq(environmentLeases.status, "active"), and( - inArray(environmentLeases.status, ["pending_cleanup", "released"]), eq(environmentLeases.leasePolicy, "ephemeral"), - )), eq(heartbeatRuns.runtimeMode, "native"), + .where(and(cancelledWorkspaceSourceLease(), eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "cancelled"), eq(nativeRunFinalizations.phase, "terminal_failure"), isNull(nativeRunFinalizations.resultId), isNull(nativeRunFinalizations.leaseOwner), sql`(${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'state' = 'prepared' OR ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'->>'state' = 'prepared')`, @@ -259,8 +279,7 @@ export function registerCancelledWorkspacePreservationWork(db: Db, return { ...entry, dueAt: Date.now() + 30_000 }; const [source] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), - or(eq(environmentLeases.status, "active"), and(inArray(environmentLeases.status, ["pending_cleanup", "released"]), - eq(environmentLeases.leasePolicy, "ephemeral"))), + cancelledWorkspaceSourceLease(), )).limit(1); if (source) await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true, requireCancelledStop: true }); const [current] = await db.select().from(heartbeatRuns) diff --git a/server/src/services/remote-workspace-writer-hold.ts b/server/src/services/remote-workspace-writer-hold.ts index fa4c69bd4a1..6fa682ab7c3 100644 --- a/server/src/services/remote-workspace-writer-hold.ts +++ b/server/src/services/remote-workspace-writer-hold.ts @@ -1,5 +1,5 @@ import { and, eq, sql } from "drizzle-orm"; -import { environmentLeases, heartbeatRuns, issues, type Db } from "@paperclipai/db"; +import { environmentLeases, heartbeatRuns, issueRecoveryActions, issues, type Db } from "@paperclipai/db"; import type { RemoteWorkspaceAuthority } from "@paperclipai/shared"; import { hasRecoveredCleanupStopReceipt, hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { canonicalExecutionUuid } from "./execution-identity.js"; @@ -55,10 +55,25 @@ export async function readRemoteWorkspaceWriterHold(tx: Db, input: { && lease.releasedAt && lease.cleanupStatus === "success" && ["released", "expired", "failed"].includes(lease.status))); if (unusedReservation) return { kind: "free" }; // An ended writer still holds the workspace until its save and sandbox - // release settle, within the same bounded cutoff as a live writer. - if (prior && writerIssue) return { kind: "held", writer: { - runId: authority.writerRunId, agentId: prior.agentId, issueId: writerIssue.id, issueIdentifier: writerIssue.identifier, - lastActivityAt: prior.finishedAt ?? prior.lastOutputAt ?? prior.startedAt ?? prior.createdAt, ended: true, - } }; + // release settle, within the same bounded cutoff as a live writer. While + // Paperclip is still saving the writer's retained files, that save's last + // progress is the writer's activity, so a waiter keeps waiting instead of + // failing setup. The save's own retry budget bounds it; a completed or + // board-handed-off save no longer extends the cutoff. + if (prior && writerIssue) { + const ended = new Date(prior.finishedAt ?? prior.lastOutputAt ?? prior.startedAt ?? prior.createdAt); + const [save] = await tx.select({ progressedAt: sql`max(${issueRecoveryActions.updatedAt})` }) + .from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, writerIssue.id), + sql`${issueRecoveryActions.evidence}->>'runId' = ${authority.writerRunId}`, + sql`${issueRecoveryActions.evidence}->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1'`, + sql`coalesce(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}', 'scheduled') in ('scheduled', 'running', 'awaiting_evidence')`, + sql`${issueRecoveryActions.evidence}#>>'{workspaceRecovery,status}' is distinct from 'recovered'`)); + const saving = save?.progressedAt ? new Date(save.progressedAt) : null; + return { kind: "held", writer: { + runId: authority.writerRunId, agentId: prior.agentId, issueId: writerIssue.id, issueIdentifier: writerIssue.identifier, + lastActivityAt: saving && saving.getTime() > ended.getTime() ? saving : ended, ended: true, + } }; + } return { kind: "unattributed" }; }