From 663fe667d66d3863d0919a8724f1d99500feef0b Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 27 Oct 2026 09:50:50 -0500 Subject: [PATCH] fix(recovery): reconstruct missing native workspace obligations (#17877) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A failed native run can still own prepared files in a remote sandbox. > - Cleanup must preserve those files before another executor can start. > - Older failure producers can omit the preservation record. > - A saved message then waits for cleanup without a recovery owner. > - This change reconstructs that record during user continuation and orphan cleanup. > - The existing worker saves the files, verifies stop, and returns input to normal admission. ## Linked Issues or Issue Description **What happened?** A failed native run retained a prepared workspace and an active sandbox lease. It had no file-preservation obligation. Cleanup protected the files. The next user message waited for a stop receipt that no worker owned. **Expected behavior** A new message registers the missing maintenance work at once. Existing orphan cleanup recovers historical saved input. Ordinary Retry remains available. Saved input can start once, after verified preservation and containment. **Steps to reproduce** 1. Persist a failed native run with an exact prepared remote workspace and its active lease. 2. Leave its finalizer terminal and unowned. Omit the workspace recovery record. 3. Send a user comment or run orphan cleanup. 4. Observe that no preservation worker starts and the comment remains deferred. **Paperclip version or commit** Reproduced on `bd382eee536f4ada680072b1cc19100163c3d04a`. **Deployment mode** Native Runner with a remote sandbox. Local verification uses disposable PostgreSQL 18 and real temporary workspace files. Refs #17788 and #17800. Those fixes preserve future deadline failures and register historical preservation on Retry. This change adds automatic recovery for new and saved messages. Refs #17839 for the separate restart-failure producer repair. ## What Changed - Register missing preservation before an authenticated continuation waits for remote cleanup. - Reconstruct the same exact-source obligation during the existing orphan lease sweep. - Notify the existing preservation worker in the registration transaction. Do not add a timer or history scan. - Test ephemeral and reusable sources, live owners, active executors, mismatched sources, rollback, lock contention, Retry and saved-input delivery. - Document the recovery contract and feature-map behavior. ## Verification - Red: both new orphan-recovery cases fail on unchanged production code. Four ownership and source negative controls pass. - Red: a new comment starts no preservation work without the admission repair. - All 461 test cases across five affected suites pass across the local runs. One existing 30-second timing case timed out under concurrent verification; its isolated rerun passed with the same deadline and assertions. - The three final-head entry journeys pass again: HTTP Retry, HTTP comment creation, and historical saved-input recovery. - `pnpm -r typecheck` and `pnpm build` pass on final head `8fab8b2ba9049f640fd884094a2007a69465e6da`. The affected server typecheck and build also pass. - The integration journey holds the provider stop reply, checks that no successor starts, preserves real files and newer host edits, and checks one saved-input successor. - Module-boundary checks, map checks and `git diff --check` pass. - Native macOS PostgreSQL cannot start because host semaphore capacity is exhausted. Database verification uses separate disposable PostgreSQL databases for each suite. An earlier shared-database attempt caused fixture interference and is not counted as a pass. - Greptile gives 5/5 with no findings or unresolved threads on `8fab8b2ba9049f640fd884094a2007a69465e6da`. All final-head GitHub checks are settled: 55 passed, two Storybook checks skipped, and one neutral Paperclip review. Full CI test, build, typecheck, Runner and browser lanes pass. No full local `pnpm test:run` pass is claimed. The fix is not deployed. This PR makes no claim of a live hosted model turn or recovery of the original incident. ## Risks - Registration grants maintenance authority only. Existing coordinator, descriptor, allocation, stop, file-receipt and execution gates remain effective. - A locked source remains held. Its one-second lock timeout does not prevent an unrelated source from registering recovery. - New preservation writes wake existing work. Settled sources produce no new recurring work. Historical discovery uses the existing orphan cleanup lifecycle and its backoff. - Provider lifecycle is simulated in local integration. PostgreSQL, HTTP routing, file export/merge, the worker and queue admission are real. - No schema, credential, model, trust, retry-budget or assignment change. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, repository inspection, tool use and code execution. The runtime does not expose the exact model ID or context-window size. ## 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 | 9 +++ maps/feature-map/recovery.md | 3 + .../explicit-native-continuation.test.ts | 16 +++- .../services/explicit-native-continuation.ts | 5 ++ .../src/services/heartbeat/recovery.test.ts | 77 ++++++++++++++++++- server/src/services/heartbeat/recovery.ts | 5 +- .../native-failed-workspace-recovery.test.ts | 36 ++++++++- .../native-failed-workspace-recovery.ts | 40 ++++++++-- .../src/services/workspace-recovery.test.ts | 47 +++++++++-- 9 files changed, 213 insertions(+), 25 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 7bc0d578290..b1408b10266 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1402,6 +1402,15 @@ replay: the provider stop receipt, saved-file receipt, current execution authori and effect fences still control successor admission. Repeated clicks retain the same recovery generation and request. +An authenticated new message also registers that missing obligation before it +waits for cleanup. Orphan cleanup reconstructs it for a failed native run that +still owns an exact prepared remote workspace, including historical saved input. +Both paths wake the existing preservation worker at commit instead of waiting +indefinitely for an export that has no owner. A live executor or finalizer, accepted result, or mismatched source +remains protected. Cleanup never interprets this registration as a stop receipt +or permission to replay provider work. Repeated cleanup visits register the source +once; settled sources add no continuing discovery or new maintenance timer. + Every continuation carries the triggering request, ordered user direction, interaction outcomes, completed work, and explicit history coverage. A delivered message remains part of the task's request after its connection or approval resolves. The original title is background; a completed Notion read does not satisfy a later Gmail request. Author and source-trust boundaries survive rendering into both native and legacy prompts. Missing required history must be fetched before dispatch rather than described as complete. ### Remote runner process evidence diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index f56db5f99ae..63983d50e29 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -40,6 +40,9 @@ Agent access: [companion](../agent-access-map/recovery.md). for the same stop-and-save workflow before Retry admits another executor. Retry reconstructs missing preservation for historical failures only from an exact prepared source without a live finalization owner. + New messages and orphan cleanup can register missing preservation automatically + and wake the existing worker. Saved input still waits for verified stop and file + persistence, then enters ordinary admission once. - `unknown-outcome`: `recovery/unknown-outcome` — uncertain external side effects require inspection and an explicit decision, not automatic replay. - `user-continuation`: `recovery/user-continuation` — saved user messages after a stop are reconsidered once diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 361af71a97c..9224609f475 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -16,11 +16,12 @@ import { join } from "node:path"; import { and, eq, sql } from "drizzle-orm"; import { beforeAll, afterAll, describe, it, expect, vi } from "vitest"; import { - approvals, issueApprovals, issueThreadInteractions, chatConversations, chatEndpoints, toolApplications, toolConnections, + applyPendingMigrations, approvals, issueApprovals, issueThreadInteractions, chatConversations, chatEndpoints, toolApplications, toolConnections, agentWakeupRequests, agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, completionContracts, environmentLeases, environments, issueRelations, issueTreeHolds, issueTreeHoldMembers, } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase, getEmbeddedPostgresTestSupport } from "../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js"; import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; import { adapterExecutionControls, createAdapterExecutionControl } from "./adapter-execution-control.js"; import { CONVERSATION_CONTINUATION_POLICY } from "./conversation-continuation.js"; @@ -29,12 +30,19 @@ import { heartbeatService, persistHeartbeatRunProcessMetadata, type HeartbeatEnv import { getExecutionBlocker } from "./execution-blocker.js"; import { createDurableChatWakeupRequest } from "./durable-chat-wakeup.js"; import { recordNativeFinalizationFailure } from "./native-runtime/native-run-finalizer.js"; -const support = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_CONTINUATION_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("explicit native conversation continuation", () => { let database: Awaited>; let db: ReturnType; - beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("explicit-native-message-"); db = createDb(database.connectionString); }, 90_000); - afterAll(async () => { await database?.cleanup(); }); + beforeAll(async () => { + if (externalUrl) { + await applyPendingMigrations(externalUrl); + database = { connectionString: externalUrl, cleanup: async () => {} }; + } else database = await startEmbeddedPostgresTestDatabase("explicit-native-message-"); + db = createDb(database.connectionString); + }, 90_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); async function seed() { const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); const sourceRunId = randomUUID(), successorRunId = randomUUID(); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 0c1ee1ebe21..904bf270097 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -4,6 +4,7 @@ import { readTaskRetryIntent } from "./task-retry.js"; import { readSystemRecovery } from "./system-recovery-state.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; +import { preserveNativeFailedWorkspace } from "./native-runtime/native-failed-workspace-recovery.js"; import { createHash } from "node:crypto"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; import { canContinueCancelledRun } from "./run-cancellation.js"; @@ -298,6 +299,10 @@ export async function admitExplicitNativeContinuation(input: { run.status === "cancelled" && !coordinator.leaseOwner ? "The cancelled run still needs verified cleanup. Your message is saved. Inspect the run and its environment for details." : "Waiting for the previous run to finish recovery. Your message will start automatically."); + const preserved = hasRequiredWorkspaceRecovery(run.resultJson) ? null + : await preserveNativeFailedWorkspace(db, run); + if (preserved && hasRequiredWorkspaceRecovery(preserved.resultJson)) return blocked("workspace_repair_required", + "Paperclip is recovering saved work before continuing. Your message is saved."); const leases = await db.select() .from(environmentLeases).where(and( eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id), diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index 6b5eb940732..01a0af3151b 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -1,6 +1,6 @@ import { randomUUID } from "node:crypto"; import type { Db } from "@paperclipai/db"; -import { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environments, environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { activityLog, agents, applyPendingMigrations, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environments, environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; import { eq } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; @@ -15,6 +15,9 @@ import { claimNativeRestartRecoveries } from "../native-runtime/native-restart-r import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, legacyShutdownWorkspaceResourceProtected, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js"; import { adapterExecutionControls, createAdapterExecutionControl } from "../adapter-execution-control.js"; import { logger } from "../../middleware/logger.js"; +import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../delivery-work-notifications.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js"; type Run = typeof heartbeatRuns.$inferSelect; @@ -169,7 +172,8 @@ describe("heartbeat recovery module callbacks", () => { }); }); -const support = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_RECOVERY_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describePostgres = support.supported ? describe : describe.skip; if (!support.supported) console.warn(`Skipping recovery module database tests: ${support.reason}`); @@ -177,10 +181,13 @@ describePostgres("heartbeat recovery module ownership", () => { let db: Db; let database: Awaited>; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-recovery-module-"); + if (externalUrl) { + await applyPendingMigrations(externalUrl); + database = { connectionString: externalUrl, cleanup: async () => {} }; + } else database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-recovery-module-"); db = createDb(database.connectionString); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); - afterAll(async () => { await database?.cleanup(); }); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); async function seedRun() { const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), issueId = randomUUID(); @@ -190,6 +197,68 @@ describePostgres("heartbeat recovery module ownership", () => { return (await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, invocationSource: "automation", status: "running", runtimeMode: "legacy" }).returning())[0]; } + it.each(["ephemeral", "reuse_by_environment", "live_owner", "live_executor", "mismatched_source", "finalizing"] as const)( + "reconstructs missing native preservation during orphan cleanup without replay: %s", async state => { + const run = await seedRun(); + const [environment] = await db.insert(environments).values({ name: run.id, driver: "sandbox" }).returning(); + const [lease] = await db.insert(environmentLeases).values({ companyId: run.companyId, issueId: run.issueId, + heartbeatRunId: run.id, environmentId: environment.id, provider: "daytona", providerLeaseId: randomUUID(), + status: "active", leasePolicy: state === "reuse_by_environment" ? state : "ephemeral", + metadata: { driver: "sandbox", sandboxProviderPlugin: true, pluginId: randomUUID(), remoteCwd: "/work" }, + }).returning(); + await db.update(heartbeatRuns).set({ status: "failed", runtimeMode: "native", nativeIssueId: run.issueId, + finishedAt: new Date(), errorCode: "execution_finalization_deadline_exceeded", contextSnapshot: { issueId: run.issueId }, + runnerProfileJson: { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared", + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, + workspaceId: run.id, leaseId: lease.id, providerLeaseId: lease.providerLeaseId, + remoteCwd: state === "mismatched_source" ? "/other" : "/work", resourceDisposition: "keep_running" } }, + }).where(eq(heartbeatRuns.id, run.id)); + await db.insert(nativeRunFinalizations).values({ companyId: run.companyId, issueId: run.issueId!, runId: run.id, + phase: state === "finalizing" ? "workspace_finalizing" : "terminal_failure", + leaseOwner: state === "live_owner" ? "another-controller" : null }); + const deps = callbacks(db); + if (state === "live_executor") deps.activeRunExecutions.add(run.id); + const recovery = createHeartbeatRecovery(db, deps); + const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.workspaceRecovery, wake); + const shouldRecover = state === "ephemeral" || state === "reuse_by_environment"; + try { + expect(await recovery.sweepOrphanedActiveLeases({ backoffMs: 0 })).toEqual({ recovered: shouldRecover ? 1 : 0 }); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + const [retained] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)); + const records = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!)); + expect(source.status).toBe("failed"); + if (shouldRecover) { + expect(source.resultJson).toMatchObject({ workspaceRestoreRecovery: { leaseIds: [lease.id] } }); + expect(retained).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure", + metadata: { sandboxStopAndRetain: { leaseId: lease.id, runId: run.id } } }); + expect(records).toHaveLength(1); + expect(records[0].evidence).toMatchObject({ automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } }); + expect(wake).toHaveBeenCalledOnce(); + const registered = source.resultJson; + expect(await recovery.sweepOrphanedActiveLeases({ backoffMs: 0 })).toEqual({ recovered: 0 }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].resultJson).toEqual(registered); + expect(wake).toHaveBeenCalledOnce(); + } else { + expect(source.resultJson).toBeNull(); + expect(retained.status).toBe("active"); + expect(records).toHaveLength(0); + expect(wake).not.toHaveBeenCalled(); + } + expect(hasRemoteTerminationReceipt(retained)).toBe(false); + expect(retained.metadata).not.toHaveProperty("remoteExecutionTermination"); + expect(deps.executeRun).not.toHaveBeenCalled(); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, run.companyId))).toHaveLength(1); + } finally { + unsubscribe(); + await db.delete(environmentLeases).where(eq(environmentLeases.id, lease.id)); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!)); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + await db.delete(activityLog).where(eq(activityLog.runId, run.id)); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + } + }, + ); + it.each(["restored", "restore_failed"])("keeps the sandbox available until adapter workspace settlement: %s", async outcome => { const run = await seedRun(); const deps = callbacks(db); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index af13219764c..ba35fe84d7b 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -3,7 +3,7 @@ import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js import type { WorkspaceWorkers } from "../plugin-workspaces.js"; import { createHeartbeatAgentStatus } from "./agent-status.js"; import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js"; -import { preserveCancelledNativeWorkspaces } from "../native-runtime/native-failed-workspace-recovery.js"; +import { preserveCancelledNativeWorkspaces, preserveOrphanedNativeWorkspace } from "../native-runtime/native-failed-workspace-recovery.js"; import { nativeWorkspaceExportPendingForLease, pendingNativeWorkspaceFinalizationCondition } from "../native-runtime/native-workspace-finalization-state.js"; import { CHAT_COMPLETION_WAKE_REASON } from "../chat-completion-delivery.js"; import { @@ -1486,6 +1486,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let recovered = 0; for (const { lease } of rows) { if (await nativeWorkspaceExportPendingForLease(db, lease)) { + if (lease.heartbeatRunId && !activeRunExecutions.has(lease.heartbeatRunId) + && !adapterExecutionControls.has(lease.heartbeatRunId) + && await preserveOrphanedNativeWorkspace(db, lease)) recovered += 1; await deferOrphanedActiveLease(lease.id); continue; } diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts index 3d5f155bd14..6aa2f9c01c0 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts @@ -6,7 +6,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro import { getExecutionBlocker } from "../execution-blocker.js"; import { markExecutionReconciliation, settleUnrecoverableExecutions, validateExecutionReconciliation } from "../execution-recovery-resolution.js"; import { prepareSandboxStopAndRetain, settleStopOnlyCleanup } from "../sandbox-stop-and-retain.js"; -import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace } from "./native-failed-workspace-recovery.js"; +import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace, preserveOrphanedNativeWorkspace } from "./native-failed-workspace-recovery.js"; import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js"; import { NativeWorkspaceFinalizationBusyError } from "./native-workspace-finalization-ownership.js"; import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js"; @@ -51,6 +51,40 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes const readLease = async (id: string) => (await db.select().from(environmentLeases).where(eq(environmentLeases.id, id)))[0]!; const actions = async (issueId: string) => db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)); + it("keeps another orphan recoverable while a task lock delays preservation", async () => { + const busy = await seed(), free = await seed(); + let unlock!: () => void, ready!: () => void; + const released = new Promise(resolve => { unlock = resolve; }); + const acquired = new Promise(resolve => { ready = resolve; }); + const holding = db.transaction(async tx => { + await tx.select().from(issues).where(eq(issues.id, busy.issueId)).for("update"); + ready(); await released; + }); + await acquired; + try { + expect(await preserveOrphanedNativeWorkspace(db, busy.lease)).toBe(false); + expect(await preserveOrphanedNativeWorkspace(db, free.lease)).toBe(true); + expect((await readLease(busy.lease.id)).status).toBe("active"); + expect(await readLease(free.lease.id)).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure" }); + expect(await actions(busy.issueId)).toHaveLength(0); + } finally { unlock(); await holding; } + expect(await preserveOrphanedNativeWorkspace(db, busy.lease)).toBe(true); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, busy.companyId))).toHaveLength(1); + }); + + it("rolls back orphan registration with its enclosing transaction", async () => { + const f = await seed(); + await expect(db.transaction(async tx => { + expect(await preserveOrphanedNativeWorkspace(tx as unknown as typeof db, f.lease)).toBe(true); + throw new Error("rollback preservation"); + })).rejects.toThrow("rollback preservation"); + expect(await readRun(f.runId)).toEqual(f.run); + expect(await readLease(f.lease.id)).toEqual(f.lease); + expect(await actions(f.issueId)).toHaveLength(0); + expect(await preserveOrphanedNativeWorkspace(db, f.lease)).toBe(true); + expect(await actions(f.issueId)).toHaveLength(1); + }); + it.each(["settled", "historical", "live_owner", "retry", "nonterminal", "wrong_wake", "wrong_hold_run", "no_hold"])("retires only an exact superseded native resume obligation: %s", async scenario => { const f = await seed(), actionId = randomUUID(); await db.insert(issueRecoveryActions).values({ id: actionId, companyId: f.companyId, sourceIssueId: f.issueId, 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 c8f31024920..6e6792d6b72 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -9,6 +9,7 @@ import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js"; import { NativeWorkspaceFinalizationBusyError } from "./native-workspace-finalization-ownership.js"; import { hasAcknowledgedNativeStopIntent } from "../acknowledged-native-stop.js"; import { logger } from "../../middleware/logger.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; export async function preserveNativeFailedWorkspace(db: Db, selected: typeof heartbeatRuns.$inferSelect, options: { allowPreterminal?: boolean; restartRecovery?: NativeRestartRecoveryClaim } = {}) { if (readNativeWorkspaceSyncReference(selected.runnerProfileJson?.nativeWorkspaceSync)?.state !== "prepared") return; @@ -53,6 +54,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: typeof hea // Their Runner descriptor is not evidence that user files need exporting. const retainFiles = !nativeRunHasNoWorkspaceAccess(run); const alreadyRetained = hasRequiredWorkspaceRecovery(run.resultJson); + if (retainFiles && !alreadyRetained) await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery); let retained = !retainFiles || alreadyRetained ? run : await recordRequiredWorkspaceRestoreFailure(tx as unknown as Db, run, { workspaceRestoreFailure: "restore_failed" }, source); if (!retained || retainFiles && !hasRequiredWorkspaceRecovery(retained.resultJson)) return; if (options.allowPreterminal && coordinator.phase === "observed") { @@ -81,6 +83,36 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: typeof hea }); } +export async function preserveOrphanedNativeWorkspace(db: Db, lease: Pick): Promise { + if (!lease.heartbeatRunId) return false; + const [run] = await db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, lease.companyId), eq(heartbeatRuns.id, lease.heartbeatRunId), + eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "failed"), + )).limit(1); + if (!run || hasRequiredWorkspaceRecovery(run.resultJson)) return false; + try { + const retained = await preserveNativeFailedWorkspace(db, run); + return Boolean(retained && hasRequiredWorkspaceRecovery(retained.resultJson)); + } catch (error) { + if (error instanceof NativeWorkspaceFinalizationBusyError) return false; + const code = retentionContentionCode(error); + if (!code) throw error; + logger.warn({ runId: run.id, code }, "Native workspace retention is waiting for its current owner"); + return false; + } +} + +function retentionContentionCode(error: unknown): string | null { + let current = error; + for (let depth = 0; depth < 4 && current && typeof current === "object"; depth++) { + const value = current as { code?: unknown; cause?: unknown }; + if (value.code === "55P03" || value.code === "57014") return value.code; + current = value.cause; + } + return null; +} + const cancelledWorkspaceCursors = new WeakMap(); /** Older controllers could acknowledge Stop without pinning prepared files. @@ -113,13 +145,7 @@ export async function preserveCancelledNativeWorkspaces(db: Db) { try { await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true }); } catch (error) { if (error instanceof NativeWorkspaceFinalizationBusyError) continue; - let current: unknown = error; - let code: string | null = null; - for (let depth = 0; depth < 4 && current && typeof current === "object"; depth++) { - const value = current as { code?: unknown; cause?: unknown }; - if (value.code === "55P03" || value.code === "57014") { code = value.code; break; } - current = value.cause; - } + const code = retentionContentionCode(error); if (!code) throw error; logger.warn({ runId: run.id, code }, "Cancelled native workspace retention is waiting for its current owner"); } diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 4eb74906cb3..b5ec6de2a25 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -137,7 +137,7 @@ describe("automatic retained workspace recovery", () => { const action = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!; const run = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; - it("repairs an unregistered historical deadline failure through HTTP Retry without admitting work before verified stop", async () => { + it.each(["HTTP Retry", "new comment", "automatic cleanup"] as const)("repairs an unregistered historical deadline failure through %s without admitting work before verified stop", async entry => { const f = await seed("native", true, "keep_running"); const source = await run(f.runId); await db.update(heartbeatRuns).set({ resultJson: null, errorCode: "execution_finalization_deadline_exceeded" }) @@ -155,7 +155,7 @@ describe("automatic retained workspace recovery", () => { await stopped; return { providerLeaseId: f.providerLeaseId, state: "stopped" }; }); - const errors = vi.fn(), heartbeat = heartbeatService(db); + const errors = vi.fn(), heartbeat = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: errors }); const worker = registerWorkspaceRecoveryWork(db, coordinator, f.environmentRuntime, heartbeat.resumeRemoteStopComments); const successors = () => db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); @@ -166,11 +166,39 @@ describe("automatic retained workspace recovery", () => { app.use((req, _res, next) => { req.actor = { type: "board", source: "local_implicit", userId: "board" }; next(); }); app.use("/api", issueRoutes(db, {} as Parameters[1])); app.use(errorHandler); - const body = { idempotencyKey: randomUUID(), observedRunId: f.runId }; - const retry = await request(app).post(`/api/issues/${f.issueId}/retry`).send(body).expect(202); - expect((await request(app).post(`/api/issues/${f.issueId}/retry`).send(body).expect(202)).body.requestId).toBe(retry.body.requestId); + let requestId: string | undefined; + let commentId: string | undefined; + if (entry === "HTTP Retry") { + const body = { idempotencyKey: randomUUID(), observedRunId: f.runId }; + const retry = await request(app).post(`/api/issues/${f.issueId}/retry`).send(body).expect(202); + requestId = retry.body.requestId; + expect((await request(app).post(`/api/issues/${f.issueId}/retry`).send(body).expect(202)).body.requestId).toBe(requestId); + } else if (entry === "new comment") { + const response = await request(app).post(`/api/issues/${f.issueId}/comments`) + .send({ body: "Is this fixed?" }).expect(201); + commentId = response.body.id; + expect(await successors()).toHaveLength(0); + } else { + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "board", body: "Is this fixed?" }).returning(); + commentId = comment.id; + const wake = { source: "automation" as const, triggerDetail: "system" as const, + reason: "issue_reopened_via_comment", requestedByActorType: "user" as const, requestedByActorId: "board", + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" }, + contextSnapshot: { source: "issue.comment.reopen", issueId: f.issueId, wakeCommentId: comment.id }, + }; + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: source.agentId, + source: wake.source, triggerDetail: wake.triggerDetail, reason: wake.reason, + requestedByActorType: "user", requestedByActorId: "board", status: "deferred_issue_execution", + payload: { ...wake.payload, _paperclipWakeContext: { ...wake.contextSnapshot, wakeReason: wake.reason, + wakeSource: wake.source, wakeTriggerDetail: wake.triggerDetail, wakeCommentIds: [comment.id] } }, + }); + expect(await successors()).toHaveLength(0); + if (entry === "automatic cleanup") await heartbeat.sweepOrphanedActiveLeases({ backoffMs: 0 }); + } await expect.poll(() => f.runtime.retryPendingSandboxTeardown.mock.calls.length, { timeout: 20_000 }).toBe(1); - await heartbeat.resumeTaskRetry(f.companyId, retry.body.requestId); + if (requestId) await heartbeat.resumeTaskRetry(f.companyId, requestId); + else await heartbeat.resumeQueuedRuns(); expect(await successors()).toHaveLength(0); expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].metadata) @@ -179,11 +207,14 @@ describe("automatic retained workspace recovery", () => { releaseStop(); await expect.poll(async () => (await successors()).length, { timeout: 20_000 }).toBe(1); const [successor] = await successors(); + if (commentId) expect(successor.contextSnapshot?.wakeCommentIds).toContain(commentId); expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n"); expect(await readFile(path.join(f.host, "newer.txt"), "utf8")).toBe("newer local work\n"); - expect(await readFreshWorkspaceReadiness(db, successor)).toMatchObject({ mode: "resume_preserved_workspace" }); - await Promise.all([heartbeat.resumeTaskRetry(f.companyId, retry.body.requestId), heartbeat.resumeRemoteStopComments(await run(f.runId))]); + if (requestId) expect(await readFreshWorkspaceReadiness(db, successor)).toMatchObject({ mode: "resume_preserved_workspace" }); + else expect(successor.contextSnapshot?.previousRunId).toBe(f.runId); + await Promise.all([requestId ? heartbeat.resumeTaskRetry(f.companyId, requestId) : heartbeat.resumeQueuedRuns(), + heartbeat.resumeRemoteStopComments(await run(f.runId))]); expect(await successors()).toHaveLength(1); expect(f.runtime.resumeRunLease).toHaveBeenCalledOnce(); expect((await run(f.runId)).status).toBe("failed");