From 050365e09d88411960fc3bacc1c08c46c0d9b6fa Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 1 Nov 2026 00:17:46 -0500 Subject: [PATCH] Preserve native workspace files after lifecycle Stop (#18517) 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. > - Agent termination must stop execution and preserve unfinished workspace files. > - A native run can acknowledge a lifecycle Stop, then finish without an accepted result. > - The file preservation worker required a board actor even when the exact native Stop receipt was valid. > - This left a prepared remote workspace allocated without a file recovery action. > - This pull request admits the exact audited receipt for file preservation only. > - The existing recovery service can then save the files and stop the same allocation without another agent turn. ## Linked Issues or Issue Description **What happened?** A cancelled native run can have a terminal, unowned coordinator and a prepared remote workspace. Agent lifecycle cancellation records a valid run-only Stop receipt without a board user. The preservation worker rejects that receipt because it uses the board Stop predicate. The source lease remains active. A resolved execution watchdog does not supply the missing file recovery action. **Steps to reproduce** Seed a cancelled native run with a terminal-failure coordinator, no result or owner, an acknowledged and audited run-only Stop, and a prepared remote workspace. Omit board attribution. Start the registered cancellation preservation worker. Before this change, it records no workspace recovery marker. **Expected behavior** Record file recovery for the exact bound allocation. Keep the cancelled result, accounting, task state and watchdog history. Use the normal recovery service to stop, save and clean up that allocation without another provider turn. **Paperclip version or commit** Reproduced on source `1e8e32290c657227857aeda2620a73db52bab4f2` with real local PostgreSQL 17. The observed deployment uses a sandbox provider and remote workspace storage. Related current workspace work: #18514. Public PR and issue searches found no duplicate repair. ## What Changed - Expose the existing exact run-only Stop receipt check. Reject empty audit identifiers. - Use that receipt only in cancelled workspace preservation. Keep board Stop and continuation checks unchanged. - Defer preservation while the run still holds a live controller lease. - Schedule that wait at the actual controller lease expiry. A lease renewal observed under the preservation lock also sets the next deadline. - Add a PostgreSQL regression for the remote-workspace-only lifecycle record, resolved watchdog history, normal recovery admission, ownership controls, rollback and idle behavior. - Document the separate file recovery obligation and its physical cleanup limits. ## Verification - Before the fix: the registered-worker PostgreSQL regression fails because the file recovery marker is absent. - After the original repair at `6fd1042efd`: 20/20 focused tests and all five sibling suites pass, 320/320 tests. The exact-head server build and typecheck pass. - The repository reviewer then found that a live controller caused a fixed 30-second retry. Both the existing-lease and real row-lock renewal cases fail against that source. The new `adb1d4eccdd48be4c86906d5bd111a6e39f20191` schedules the actual recorded deadline before locking and after the locked reread. The lock still enforces the ownership guard. - At `adb1d4eccd`: all 22 focused PostgreSQL controls pass. The directly affected retention and board receipt sibling suites also pass, 124/124 tests across three files. They prove no intermediate reads or transactions before expiry, preservation after expiry, and no extra execution. Server typecheck and the exact-head server build pass. - Map checks, module boundary checks and `git diff --check` pass. - Run `pnpm exec vitest run server/src/services/native-runtime/native-cancelled-workspace-preservation.test.ts server/src/services/native-runtime/native-failed-workspace-recovery.test.ts server/src/services/acknowledged-native-stop.test.ts` with the supported external test-database variable set to a disposable local PostgreSQL database. The earlier broader run also covered run control and workspace recovery. - Exact-head [canonical CI](https://github.com/paperclipai/paperclip/actions/runs/41274963476) completed successfully. All 59 current-head checks are terminal with zero failures; required `ci / verify` and `ci / e2e` pass. Both [Greptile](https://github.com/paperclipai/paperclip/pull/18517#issuecomment-6553528545) and the [repository assessment](https://github.com/paperclipai/paperclip/pull/18517#issuecomment-6553542139) reviewed exact `adb1d4eccd` at 5/5, with zero findings or unresolved threads. The repository assessment's check concludes `NEUTRAL`; the validated assessment itself is 5/5. Earlier reviews, including the valid 4/5 deadline finding at `6fd1042efd`, remain in review history. The [earlier CI run](https://github.com/paperclipai/paperclip/actions/runs/41273965963) is also retained. No old result substitutes for a current-head gate. - Real sandbox recovery and physical cleanup remain unverified. Earlier failed QA outcomes remain failed. This PR changes no test grade or historical row. - Repo-wide local tests and typecheck were not run. The focused server checks cover this change; canonical CI supplies the remaining repository gates. ## Risks - The newly admitted receipt permits only file preservation. It cannot authorize a continuation, accept a native result or certify provider termination. - Recovery still requires exact company, task, run, lease, allocation and workspace bindings. Accepted results and live owners remain with their existing owners. - Provider or storage failure can leave recovery pending. The source must remain retained until saved-file and physical termination receipts exist. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI GPT-6 through Codex. Used reasoning, source inspection, tool use and code execution. The exact serving model identifier and context window are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/execution-semantics.md | 12 + .../src/services/acknowledged-native-stop.ts | 5 +- ...e-cancelled-workspace-preservation.test.ts | 235 ++++++++++++++++++ .../native-failed-workspace-recovery.ts | 15 +- 4 files changed, 260 insertions(+), 7 deletions(-) create mode 100644 server/src/services/native-runtime/native-cancelled-workspace-preservation.test.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index e70e5de4208..68cc532113b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1645,6 +1645,18 @@ and obtains a fresh stop receipt. An interrupted legacy run does not need an accepted native result. A successful `workspace_recovery` operation and matching recovery record prove persistence; adapter output or a user repair note cannot. The failed run and unknown external effects remain historical facts. +Cancelled native work can owe files after an audited run-only Stop from the +agent lifecycle, without a board actor. The preservation worker admits that +exact acknowledged receipt only when the coordinator is terminal, has no +accepted result or owner, and the run's controller lease is not live. While a +controller is live, the worker waits for its actual lease expiry, including a +renewal observed when preservation rechecks the locked run. Once eligible, it +records the bound prepared workspace and allocation as file recovery work. An existing +resolved execution watchdog does not discharge that file obligation. This does +not grant board Stop or continuation authority, change the cancelled result, +or prove physical termination. Normal recovery stops and retains the same +allocation, saves its files, and obtains a fresh stop receipt before cleanup +can apply the original destroy disposition. It starts no provider turn. A reusable lease can also owe native copyback: the export-repair action must select its exact immutable descriptor and the finalizer's accepted successful result under the same company, task, run and completion contract. Lease reuse diff --git a/server/src/services/acknowledged-native-stop.ts b/server/src/services/acknowledged-native-stop.ts index 9d07f7b61a7..e937f3df2a7 100644 --- a/server/src/services/acknowledged-native-stop.ts +++ b/server/src/services/acknowledged-native-stop.ts @@ -4,7 +4,7 @@ import { hasNativeLocalProcessStop } from "./native-local-process-stop.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; /** A server-recorded run-only Stop must not manufacture a recovery incident. */ -function hasAcknowledgedNativeRunStopReceipt(run: { +export function hasAcknowledgedNativeRunStopReceipt(run: { id: string; companyId: string; status: string; nativeIssueId: string | null; resultJson: Record | null; }): boolean { @@ -14,7 +14,8 @@ function hasAcknowledgedNativeRunStopReceipt(run: { intent.companyId === run.companyId && intent.issueId === run.nativeIssueId && intent.scope === "run" && intent.reasonCode === "cancellation_run_only" && intent.dispatchState === "acknowledged" && intent.dispatched === true && - typeof intent.intentAuditId === "string" && typeof intent.acknowledgementAuditId === "string"; + typeof intent.intentAuditId === "string" && Boolean(intent.intentAuditId) && + typeof intent.acknowledgementAuditId === "string" && Boolean(intent.acknowledgementAuditId); } export function hasAcknowledgedNativeStopIntent(run: Parameters[0]): boolean { 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 new file mode 100644 index 00000000000..72a8df28337 --- /dev/null +++ b/server/src/services/native-runtime/native-cancelled-workspace-preservation.test.ts @@ -0,0 +1,235 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, inArray, sql } from "drizzle-orm"; +import { afterAll, afterEach, describe, expect, it, vi } from "vitest"; +import { agents, agentWakeupRequests, companies, completionContracts, createDb, environmentLeases, environments, + executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import { recoveryPostgresSupport, startRecoveryPostgres } from "../../__tests__/helpers/recovery-postgres.js"; +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { hasAcknowledgedNativeStopIntent } from "../acknowledged-native-stop.js"; +import { requestWorkspaceRecovery } from "../workspace-recovery.js"; +import { inspectWorkspaceRecoverySource } from "../workspace-recovery-source.js"; +import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace, + registerCancelledWorkspacePreservationWork } from "./native-failed-workspace-recovery.js"; + +(recoveryPostgresSupport.supported ? describe : describe.skip)("lifecycle-cancelled workspace preservation", () => { + let database: Awaited>; + let db: ReturnType; + const fixtures = new Set(); + async function seed() { + if (!database) { + database = await startRecoveryPostgres("lifecycle-cancelled-workspace-"); + db = createDb(database.connectionString); + } + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + fixtures.add(runId); + const workspaceId = randomUUID(), leaseId = randomUUID(), environmentId = randomUUID(); + const providerLeaseId = `allocation-${randomUUID()}`, snapshotId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Stopped native files", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Stopped worker", adapterType: "paperclip_runner", + status: "terminated", lifecycleState: "terminating" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Keep cancelled files", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(environments).values({ id: environmentId, name: `Preservation ${runId}`, driver: "sandbox" }); + const reference = { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId, providerLeaseId, + remoteCwd: "/work", epoch: 2, state: "prepared", snapshotId }; + const nativeCancellation = { schema: "paperclip.native-cancellation.v1", companyId, issueId, runId, + scope: "run", reasonCode: "cancellation_run_only", dispatchState: "acknowledged", dispatched: true, + intentAuditId: randomUUID(), acknowledgementAuditId: randomUUID(), acknowledgedAt: new Date().toISOString(), + effects: ["release_run_resources"] }; + const [run] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, nativeIssueId: issueId, + runtimeMode: "native", runtimeModeResolvedAt: new Date(), nativePhase: "observed", status: "cancelled", + errorCode: "cancelled", error: "Agent lifecycle stop requested", finishedAt: new Date(), + costAccountingPending: true, usageJson: { inputTokens: 7 }, + contextSnapshot: { issueId, taskId: issueId }, + runnerProfileJson: { remoteWorkspace: reference, nativeResourceDisposition: "destroy" }, + resultJson: { nativeCancellation, startupCancellation: { beforeNativeSelection: false } }, + }).returning(); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "terminal_failure", + failureCode: "native_continuation_requires_reconciliation", recoveryState: "blocked" }); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId, + name: "Retained remote files", mode: "isolated", strategyType: "scratch", metadata: { remoteWorkspace: { + schema: "paperclip.remote-workspace.v1", epoch: 2, snapshotId, source: { kind: "empty" }, + location: { environmentId, leaseId, providerLeaseId, remoteCwd: "/work" }, writerRunId: runId, dirty: true, + } } }); + await db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, heartbeatRunId: runId, + issueId, executionWorkspaceId: workspaceId, provider: "daytona", providerLeaseId, status: "active", leasePolicy: "ephemeral", + metadata: { driver: "sandbox", pluginId: randomUUID(), sandboxProviderPlugin: true, remoteCwd: "/work" } }); + return { companyId, agentId, issueId, runId, workspaceId, leaseId, reference, nativeCancellation, run }; + } + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + afterEach(async () => { + vi.useRealTimers(); vi.restoreAllMocks(); + if (fixtures.size) await db.update(heartbeatRuns).set({ status: "failed" }).where(inArray(heartbeatRuns.id, [...fixtures])); + fixtures.clear(); + }); + + async function unchanged(f: Awaited>) { + 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]) + .toMatchObject({ status: "active", cleanupStatus: null, releasedAt: null }); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId))).toHaveLength(0); + 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); + } + + it("reconstructs prepared remote files after an audited lifecycle Stop without a board actor", async () => { + const f = await seed(); + const [historical] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", status: "resolved", ownerType: "board", cause: "native_continuation_requires_reconciliation", + fingerprint: randomUUID(), evidence: { runId: f.runId, originalFailureCode: "native_continuation_requires_reconciliation", + nativeFailureBlock: true, automaticRecovery: { policy: "preserve_without_replay_v1", replay: "blocked", runId: f.runId } }, + nextAction: "The prior execution remains stopped.", resolvedAt: new Date() }).returning(); + const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + 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({ workspaceRestoreFailure: "restore_failed", + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.leaseId] }, + nativeCancellation: f.nativeCancellation }); + expect(retained).toMatchObject({ status: "cancelled", nativePhase: "observed" }); + expect(retained).toMatchObject({ costAccountingPending: true, costAccountedAt: null, + usageJson: f.run.usageJson, finishedAt: f.run.finishedAt }); + expect(retained.runnerProfileJson).toEqual(f.run.runnerProfileJson); + expect(hasAcknowledgedNativeStopIntent(retained)).toBe(false); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + expect(lease).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure", + metadata: { sandboxStopAndRetain: { runId: f.runId, leaseId: f.leaseId } } }); + const actions = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId)); + expect(actions).toHaveLength(2); + expect(actions.find(action => action.id === historical.id)).toEqual(historical); + const preservation = actions.find(action => action.id !== historical.id)!; + expect(preservation.evidence).toMatchObject({ runId: f.runId, workspaceRestoreRecovery: { leaseIds: [f.leaseId] } }); + await preserveCancelledNativeWorkspaces(db); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId))).toEqual(actions); + const source = { companyId: f.companyId, issueId: f.issueId, runId: f.runId, actionId: preservation.id }; + expect((await inspectWorkspaceRecoverySource(db, source)).sources).toHaveLength(1); + expect(worker.nextWakeAt()).toBeNull(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); + reads.mockRestore(); vi.useRealTimers(); + expect(await requestWorkspaceRecovery({ ...source, db, idempotencyKey: randomUUID(), actorId: "operator" })) + .toMatchObject({ actionId: source.actionId, status: "queued" }); + 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((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("in_progress"); + expect((await db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, f.companyId), + eq(nativeRunFinalizations.runId, f.runId))))[0]).toMatchObject({ phase: "terminal_failure", resultId: null, + leaseOwner: null, failureCode: "native_continuation_requires_reconciliation" }); + } finally { await worker.stop(); } + }, 120_000); + + it.each(["pending", "undispatched", "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) + ("does not admit preservation for %s", async invalid => { + const f = await seed(); + const intent: Record = { ...f.nativeCancellation }; + if (invalid === "pending") intent.dispatchState = "pending"; + if (invalid === "undispatched") intent.dispatched = false; + if (invalid === "foreign-company") intent.companyId = randomUUID(); + if (invalid === "foreign-issue") intent.issueId = randomUUID(); + if (invalid === "foreign-run") intent.runId = randomUUID(); + if (invalid === "turn-scope") intent.scope = "turn"; + if (invalid === "empty-intent-audit") intent.intentAuditId = ""; + if (invalid === "missing-ack-audit") delete intent.acknowledgementAuditId; + if (invalid === "empty-ack-audit") intent.acknowledgementAuditId = ""; + const reference = { ...f.reference }; + if (invalid === "foreign-allocation") reference.providerLeaseId = randomUUID(); + if (invalid === "foreign-workspace") reference.workspaceId = randomUUID(); + if (invalid === "foreign-cwd") reference.remoteCwd = "/another-workspace"; + if (invalid === "saved-reference") reference.state = "saved"; + if (invalid === "unknown-reference") reference.schema = "unknown"; + await db.update(heartbeatRuns).set({ resultJson: { nativeCancellation: intent }, + runnerProfileJson: { remoteWorkspace: reference }, + ...(invalid === "running" ? { status: "running" } : {}), + ...(invalid === "live-controller" ? { controllerLeaseExpiresAt: new Date(Date.now() + 60_000) } : {}), + }).where(eq(heartbeatRuns.id, f.runId)); + if (invalid === "live-owner") await db.update(nativeRunFinalizations).set({ leaseOwner: randomUUID(), + leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (invalid === "accepted-result") { + const [contract] = await db.insert(completionContracts).values({ companyId: f.companyId, issueId: f.issueId, + revision: 1, schemaVersion: "v1", policyVersion: "v1", risk: "low", completionAuthority: "agent", + incompleteCriteriaPolicy: "block", contractJson: {}, canonicalSha256: "a".repeat(64), + createdByActorType: "system", createdByActorId: "test" }).returning(); + await db.update(heartbeatRuns).set({ completionContractId: contract.id }).where(eq(heartbeatRuns.id, f.runId)); + const [result] = await db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, runId: f.runId, + completionContractId: contract.id, serverFingerprint: "b".repeat(64), schemaStatus: "accepted", + resultJson: {}, canonicalSha256: "c".repeat(64) }).returning(); + await db.update(nativeRunFinalizations).set({ resultId: result.id }).where(eq(nativeRunFinalizations.runId, f.runId)); + } + const errors: unknown[] = []; + const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { errors.push(error); } }); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + expect(errors).toEqual([]); + await unchanged(f); + } finally { await worker.stop(); } + }); + + it("rolls preservation back with its outer transaction", async () => { + const f = await seed(); + await expect(db.transaction(async tx => { + await preserveNativeFailedWorkspace(tx as unknown as typeof db, f.run, + { allowPreterminal: true, requireCancelledStop: true }); + throw new Error("outer transaction rollback"); + })).rejects.toThrow("outer transaction rollback"); + await unchanged(f); + }); + + it.each(["existing", "renewed-under-lock"] as const)("waits for the actual %s controller lease deadline", async kind => { + const f = await seed(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const expiresAt = new Date(Date.now() + 600_000); + let release = () => {}; + let blocker: Promise | undefined; + if (kind === "existing") { + await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: expiresAt }).where(eq(heartbeatRuns.id, f.runId)); + } else { + let ready!: (pid: number) => void; + const acquired = new Promise(resolve => { ready = resolve; }); + const released = new Promise(resolve => { release = resolve; }); + blocker = db.transaction(async tx => { + await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, f.issueId)).for("update"); + const [row] = await tx.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`); + ready(row.pid); await released; + }); + const pid = await acquired; + const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + const registration = registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true); + await vi.waitFor(async () => { + const [row] = await db.execute<{ waiting: number }>(sql`select count(*)::int as waiting from pg_stat_activity + where datname = current_database() and ${pid} = any(pg_blocking_pids(pid))`); + expect(row.waiting).toBe(1); + }); + await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: expiresAt }).where(eq(heartbeatRuns.id, f.runId)); + release(); await blocker; await registration.ready; + expect(worker.nextWakeAt()).toBe(expiresAt.getTime()); + const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(300_000); + expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); + reads.mockRestore(); transactions.mockRestore(); + await vi.advanceTimersByTimeAsync(300_001); + await vi.waitFor(async () => expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].resultJson) + .toHaveProperty("workspaceRestoreRecovery")); + expect(worker.nextWakeAt()).toBeNull(); + } finally { release(); await blocker; await worker.stop(); } + return; + } + const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + expect(worker.nextWakeAt()).toBe(expiresAt.getTime()); + const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(300_000); + expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); + reads.mockRestore(); transactions.mockRestore(); + await vi.advanceTimersByTimeAsync(300_001); + await vi.waitFor(async () => expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].resultJson) + .toHaveProperty("workspaceRestoreRecovery")); + expect(worker.nextWakeAt()).toBeNull(); + } finally { await worker.stop(); } + }); +}); 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 1ad5ab0ecf5..a3e0216da61 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -11,7 +11,7 @@ import { logActivity } from "../activity-log.js"; import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js"; import { NativeWorkspaceFinalizationBusyError } from "./native-workspace-finalization-ownership.js"; -import { hasAcknowledgedNativeStopIntent } from "../acknowledged-native-stop.js"; +import { hasAcknowledgedNativeRunStopReceipt } from "../acknowledged-native-stop.js"; import { logger } from "../../middleware/logger.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { preserveRemoteWorkspacePreparationFailure } from "./native-workspace-finalization-state.js"; @@ -54,7 +54,8 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick Date.now())) return; const currentReference = workspaceRecoveryReference(run.runnerProfileJson); if (!currentReference || currentReference.state !== "prepared" || (claim?.kind === "reattach_remote_runner" && (claim.remote.kind === "plugin_task" || claim.remote.providerLeaseId !== currentReference.providerLeaseId @@ -166,7 +167,7 @@ export async function preserveCancelledNativeWorkspaces(db: Db) { for (const candidate of candidates) { cancelledWorkspaceCursors.set(db, candidate.runId); const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, candidate.runId)); - if (run && hasAcknowledgedNativeStopIntent(run)) { + if (run && hasAcknowledgedNativeRunStopReceipt(run)) { try { await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true, requireCancelledStop: true }); } catch (error) { if (error instanceof NativeWorkspaceFinalizationBusyError) continue; @@ -192,7 +193,7 @@ export function registerCancelledWorkspacePreservationWork(db: Db, type Entry = { id: string; companyId: string; dueAt: number }; type Candidate = Pick; - const eligible = (run: Candidate) => run.nativeIssueId !== null && hasAcknowledgedNativeStopIntent(run) + const eligible = (run: Candidate) => run.nativeIssueId !== null && hasAcknowledgedNativeRunStopReceipt(run) && workspaceRecoveryReference(run.runnerProfileJson)?.state === "prepared" && !hasRequiredWorkspaceRecovery(run.resultJson) && !nativeRunHasNoWorkspaceAccess(run); const scopeCondition = (scope?: CancelledWorkspacePreservationScope) => scope ? and( @@ -219,6 +220,8 @@ export function registerCancelledWorkspacePreservationWork(db: Db, .where(and(pendingCancelledWorkspace(), scopeCondition({ companyId: entry.companyId, runId: entry.id }))); if (!run || !eligible(run)) return null; if (!canRun() || signal?.aborted) throw new Error("Cancelled workspace preservation stopped"); + const controllerDeadline = run.controllerLeaseExpiresAt?.getTime(); + if (controllerDeadline && controllerDeadline > Date.now()) return { ...entry, dueAt: controllerDeadline }; const [finalization] = await db.select().from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.runId, run.id), eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.issueId, run.nativeIssueId!), @@ -234,7 +237,9 @@ export function registerCancelledWorkspacePreservationWork(db: Db, if (source) await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true, requireCancelledStop: true }); const [current] = await db.select().from(heartbeatRuns) .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId))); - return current && current.status === "cancelled" && eligible(current) ? { ...entry, dueAt: Date.now() + 30_000 } : null; + const currentControllerDeadline = current?.controllerLeaseExpiresAt?.getTime(); + return current && current.status === "cancelled" && eligible(current) ? { ...entry, + dueAt: currentControllerDeadline && currentControllerDeadline > Date.now() ? currentControllerDeadline : Date.now() + 30_000 } : null; } return registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.cancelledWorkspacePreservation, batchSize: 5,