diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 1a631958412..e9dd8187545 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1227,18 +1227,27 @@ Shutdown, process loss, and provider failure use the existing durable failure re Graceful shutdown interrupts an owned legacy adapter through its cancellation control, then waits within one shared 30-second window for provider stop and -adapter settlement across all affected runs. Before interruption becomes visible, -it commits a settlement fence on the exact sandbox lease. Cleanup on this server -or another server defers while that fence is pending. If the controller dies, -cleanup records exact stop-only retention and a repair hold after the deadline, -without claiming a successful restore. The shipped Compose services allow 90 -seconds for settlement and subsequent shutdown cleanup. +adapter settlement across all affected runs. A run that exceeds that wait stays +in the server's finalizer drain, which has a further shared 30-second budget. +The wait ending does not prove the adapter or its file export stopped. +Before interruption becomes visible, shutdown commits a settlement fence on +the exact sandbox lease. The owning finalizer renews its 30-second deadline +every 10 seconds, with one renewal in flight per run. Renewal uses a one-second +database lock timeout and a five-second statement timeout. It cannot revive an +expired, settled, or foreign-owned fence. Cleanup on this server defers while +the adapter still owns settlement; another server reads the renewed fence. +If the controller dies, renewal stops. Cleanup records exact stop-only retention +and a repair hold after the deadline, without claiming a successful restore. +The shipped Compose services allow 90 seconds for settlement and subsequent +shutdown cleanup. The sandbox lease remains available during workspace export and instruction collection. Lease release and recovery classification follow settlement and use -the current execution evidence. If settlement exceeds the deadline with an -active sandbox source, shutdown records a restore failure and pins the exact -source for stop and retention before cleanup. A later run must pass the normal -workspace repair gate. Native runner ownership keeps its existing restart path. +the current execution evidence. A late successful export can finish cleanup +and queue one bounded continuation without a false workspace repair hold. +An actual restore failure, or loss of the controller before settlement, keeps +the exact source retained. A later run must pass the normal workspace repair +gate. This shutdown join does not reconstruct an interrupted legacy export +after process death. Native runner ownership keeps its existing restart path. Real gates still apply: company and task ownership, active provider ownership, budget limits, agent availability, dependencies, pending approval/review paths, and explicit pause holds. Native runner reattachment and finalization retain their existing ownership protocol. Process, HTTP, and gateway adapters retain their recovery rules because invoking those adapters can itself repeat an external action rather than start a conversation turn. diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index 9eb68f7ef32..9bb221e6ca2 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -24,6 +24,11 @@ Agent access: `available` — REST / native call_api; CLI issue diagnostics. Policy: Task/run visibility, run ownership and source/action policy; workspace export retry and some operator controls require board. +Graceful legacy shutdown retains the exact sandbox until its owned adapter +settles, including after the first shutdown wait. A renewed settlement fence +protects cleanup by other controllers. Loss of the owner retains the source +for repair after expiry. This automatic protection grants no agent operation, +execution permission, or bypass of workspace repair gates. The server can recover a prepared remote bootstrap only with a durable never-claimed executor receipt and no contradictory execution evidence. This automatic classification grants no new agent action or permission; normal diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 677f40fe39f..8f3d23b61c6 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -211,6 +211,20 @@ repair, record the compared branches/revisions and preserve uncommitted work; confirm the repair's actual result before resuming. Check the denied and stale action paths. A changed card label is insufficient proof of workspace integrity. +For graceful legacy shutdown, interrupt a disposable sandbox run while file +export is in flight. Delay settlement past the first 30-second wait. Confirm +the lease stays available, no repair hold or successor appears before export, +and the finalizer drain waits for the same owner. After export, confirm saved +files, successful lease cleanup, and one bounded continuation. Reload before +checking the successor. Repeat with controller loss: the expired fence must +retain the exact source and require repair, never claim restored files or start +a second executor. The database-backed +[shutdown regression](../../server/src/__tests__/heartbeat-process-recovery.test.ts) +checks finalizer joining, release order, durable retry uniqueness, and reload. +The [settlement regressions](../../server/src/services/heartbeat/recovery.test.ts) +cover other-controller fencing and owner loss. Provider boundaries are controlled +in these tests; they do not qualify a live sandbox export. + For an abandoned native controller, follow the [stop-verification procedure](../../doc/native-workspace-finalization-recovery.md) with a disposable deployment. A new platform-bound owner recovers automatically only after the harness signs proof that the exact deployment fully stopped. diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 4b8a38f3e3b..a9b0b766164 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1,7 +1,7 @@ import { enableRunFailureReporting } from "./helpers/run-failure-reporting.js"; import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; import * as conversationContinuation from "../services/conversation-continuation.js"; -import { environmentRuntimeService } from "../services/environment-runtime.js"; +import { classifyEnvironmentCapabilities, environmentRuntimeService } from "../services/environment-runtime.js"; import * as workspaceRuntime from "../services/workspace-runtime.js"; import { nativeWorkspaceSyncInternals } from "../services/native-runtime/native-workspace-sync.js"; import { ensureNativeCompletionContract } from "../services/native-runtime/completion-contracts.js"; @@ -4081,6 +4081,63 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("drains late legacy settlement before releasing its source and scheduling one continuation", async () => { + const f = await seedRunFixture({ runtimeMode: "legacy", adapterType: "codex_local", agentStatus: "running" }); + await db.update(heartbeatRuns).set({ scopeKind: "issue", issueId: f.issueId }).where(eq(heartbeatRuns.id, f.runId)); + const { leaseId } = await seedEnvironmentLeaseFixture({ ...f, provider: "daytona", driver: "sandbox" }); + await db.update(environmentLeases).set({ providerLeaseId: randomUUID(), metadata: { driver: "sandbox" } }) + .where(eq(environmentLeases.id, leaseId)); + const release = vi.fn(async () => { + const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(saved.resultJson).toMatchObject({ workspaceRestored: true }); + return (await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date() }) + .where(eq(environmentLeases.id, leaseId)).returning())[0]; + }); + const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", + resolveCapabilities: async () => classifyEnvironmentCapabilities({}), + acquireRunLease: async () => { throw new Error("Unexpected acquire"); }, releaseRunLease: release }] }); + const heartbeat = heartbeatService(db, { environmentRuntime: runtime }); + const control = createAdapterExecutionControl(); + adapterExecutionControls.set(f.runId, control); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const shutdown = heartbeat.drainRunningRunsForShutdown("SIGTERM", new Date(), [f.runId]); + let drained = false; + let finalizers: Promise | undefined; + try { + await vi.waitFor(() => expect(control.controller.signal.aborted).toBe(true)); + await vi.advanceTimersByTimeAsync(30_000); + expect(await shutdown).toMatchObject({ interruptedRunIds: [f.runId], retryRunIds: [] }); + finalizers = heartbeat.drainActiveRunExecutions().then(() => { drained = true; }); + await Promise.resolve(); + expect(drained).toBe(false); + expect(release).not.toHaveBeenCalled(); + expect(await selectRecoveryRunsByAgent(f.agentId)).toHaveLength(1); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId))).toHaveLength(0); + await db.update(heartbeatRuns).set({ resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || '{"workspaceRestored":true}'::jsonb` }) + .where(eq(heartbeatRuns.id, f.runId)); + control.finish(); + await finalizers; + expect(release).toHaveBeenCalledOnce(); + const runs = await selectRecoveryRunsByAgent(f.agentId); + expect(runs).toHaveLength(2); + expect(runs.find(run => run.id !== f.runId)).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: 1, + scopeKind: "issue", issueId: f.issueId }); + const reloaded = heartbeatService(db, { environmentRuntime: runtime }); + await reloaded.drainRunningRunsForShutdown("SIGTERM", new Date(), [f.runId]); + expect(await selectRecoveryRunsByAgent(f.agentId)).toHaveLength(2); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)))[0]) + .toMatchObject({ status: "released", cleanupStatus: "success", metadata: { legacyShutdownWorkspaceSettlement: { state: "settled" } } }); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId))).toHaveLength(0); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + control.finish(); + adapterExecutionControls.delete(f.runId); + await shutdown; + await finalizers; + } + }); + it("reports a selectively drained ACP run as lost when terminal persistence fails", async () => { const child = spawnAliveProcess(); childProcesses.add(child); diff --git a/server/src/index.ts b/server/src/index.ts index 6d33c9742a6..674deb49626 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -2099,6 +2099,7 @@ async function startServerWithDatabaseTeardown( await drainRunExecutionFinalizersForShutdown({ signal, drain: drainHeartbeatExecutionFinalizers, + timeoutMs: 30_000, log: logger, }); } diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index 6d76f2df511..8faa8c7e865 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -10,8 +10,9 @@ import { createHeartbeatRecovery, type HeartbeatRecoveryDependencies } from "./r import type { HotRestartIntent } from "../hot-restart.js"; import type { NativeRestartRecoveryClaim, NativeRestartRecoveryDisposition } from "../native-runtime/index.js"; import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "../native-runtime/native-runner-ownership.js"; -import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, legacyShutdownWorkspaceResourceProtected, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js"; +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"; type Run = typeof heartbeatRuns.$inferSelect; @@ -290,6 +291,103 @@ describePostgres("heartbeat recovery module ownership", () => { adapterExecutionControls.delete(f.run.id); } + it("does not let an elapsed shutdown wait stop a sandbox whose exporter still owns settlement", async () => { + const f = await seedShutdownSource(); + const control = createAdapterExecutionControl(); + control.controller.abort(); + adapterExecutionControls.set(f.run.id, control); + const stop = vi.fn(async () => ({ providerLeaseId: f.lease.providerLeaseId, state: "stopped" })); + const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", + resolveCapabilities: async () => classifyEnvironmentCapabilities({}), + acquireRunLease: async () => { throw new Error("Unexpected acquire"); }, + releaseRunLease: vi.fn(async () => null), retryPendingSandboxTeardown: stop }] }); + try { + await db.update(heartbeatRuns).set({ status: "interrupted", errorCode: "server_shutdown_interrupted", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, f.run.id)); + const [source] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + expect(await allowLegacyShutdownWorkspaceCleanup(db, source, new Date(Date.now() + 60_000))).toBe(false); + await runtime.releaseRunLeases(f.run.id); + expect(stop).not.toHaveBeenCalled(); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].status).toBe("active"); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.run.issueId!))).toHaveLength(0); + } finally { control.finish(); await deleteShutdownSource(f); } + }); + + it("renews the durable export fence for other controllers and expires it after owner loss", async () => { + const f = await seedShutdownSource(); + const [initial] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + const originalDeadline = (initial.metadata!.legacyShutdownWorkspaceSettlement as { deadlineAt: string }).deadlineAt; + vi.useFakeTimers({ toFake: ["Date", "setInterval", "clearInterval"] }); + const stopRenewing = maintainLegacyShutdownWorkspaceSettlement(db, f.run); + try { + await vi.advanceTimersByTimeAsync(10_000); + await vi.waitFor(async () => { + const [current] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + expect(Date.parse((current.metadata!.legacyShutdownWorkspaceSettlement as { deadlineAt: string }).deadlineAt)) + .toBeGreaterThan(Date.parse(originalDeadline)); + }); + expect(await allowLegacyShutdownWorkspaceCleanup(db, initial, new Date(Date.parse(originalDeadline) + 1))).toBe(false); + await stopRenewing(); + await vi.advanceTimersByTimeAsync(60_000); + await db.update(heartbeatRuns).set({ status: "interrupted", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + await sweepLegacyShutdownWorkspaceSettlements(db); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]) + .toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure", + metadata: { legacyShutdownWorkspaceSettlement: { state: "expired" }, sandboxStopAndRetain: { leaseId: f.lease.id } } }); + } finally { await stopRenewing(); vi.useRealTimers(); await deleteShutdownSource(f); } + }); + + it("bounds a contended settlement renewal without blocking unrelated database work", async () => { + const f = await seedShutdownSource(); + const warning = vi.spyOn(logger, "warn"); + let unlock!: () => void; + let locked!: () => void; + const lockReady = new Promise(resolve => { locked = resolve; }); + const releaseLock = new Promise(resolve => { unlock = resolve; }); + const lock = db.transaction(async tx => { + await tx.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)).for("update"); + locked(); + await releaseLock; + }); + await lockReady; + vi.useFakeTimers({ toFake: ["Date", "setInterval", "clearInterval"] }); + const stopRenewing = maintainLegacyShutdownWorkspaceSettlement(db, f.run); + try { + await vi.advanceTimersByTimeAsync(10_000); + expect(await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id))) + .toEqual([{ id: f.run.id }]); + await stopRenewing(); + expect(warning).toHaveBeenCalledWith({ runId: f.run.id, errorKind: "lock_timeout" }, + "Shutdown workspace settlement lease renewal failed; source remains protected locally"); + } finally { + unlock(); + await lock; + await stopRenewing(); + vi.useRealTimers(); + warning.mockRestore(); + await deleteShutdownSource(f); + } + }); + + it.each([["57014", "statement_cancelled"], ["08006", "connection_failed"], ["private-credential", "unknown"]])( + "reports renewal failure %s without raw database diagnostics", async (code, errorKind) => { + const run = { id: randomUUID(), companyId: randomUUID() }; + const privateValue = "postgres://private-credential@example.invalid/database"; + const failure = new Error(privateValue, { cause: { code, query: privateValue, params: [privateValue] } }); + const dbBoundary = { transaction: vi.fn().mockRejectedValue(failure) } as unknown as Db; + const warning = vi.spyOn(logger, "warn").mockImplementation(() => undefined); + vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] }); + const stopRenewing = maintainLegacyShutdownWorkspaceSettlement(dbBoundary, run); + try { + await vi.advanceTimersByTimeAsync(10_000); + await stopRenewing(); + expect(warning).toHaveBeenCalledExactlyOnceWith({ runId: run.id, errorKind }, + "Shutdown workspace settlement lease renewal failed; source remains protected locally"); + expect(JSON.stringify(warning.mock.calls)).not.toContain("private-credential"); + } finally { await stopRenewing(); vi.useRealTimers(); warning.mockRestore(); } + }, + ); + it("does not turn a stale pending snapshot into a repair hold after settlement", async () => { const f = await seedShutdownSource(); try { @@ -425,12 +523,13 @@ describePostgres("heartbeat recovery module ownership", () => { } finally { control.finish(); await deleteShutdownSource(f); } }); - it("pins an unjoined sandbox restore for repair before bounded shutdown cleanup", async () => { + it("joins a late successful export before cleanup and continuation after the shutdown wait", async () => { const run = await seedRun(); await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: run.issueId } }).where(eq(heartbeatRuns.id, run.id)); const [lease] = await db.insert(environmentLeases).values({ companyId: run.companyId, heartbeatRunId: run.id, issueId: run.issueId, status: "active", provider: "daytona", providerLeaseId: randomUUID(), leasePolicy: "ephemeral", metadata: { driver: "sandbox", sandboxProviderPlugin: true, pluginId: "fixture-plugin", remoteCwd: "/home/daytona/workspace" } }).returning(); + await db.update(agents).set({ adapterType: "codex_local" }).where(eq(agents.id, run.agentId)); const deps = callbacks(db); deps.setRunStatusIfRunning.mockImplementation(async (id, status, patch) => { expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0].metadata) @@ -440,10 +539,10 @@ describePostgres("heartbeat recovery module ownership", () => { }); deps.getRun.mockImplementation(async id => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0] ?? null); deps.releaseEnvironmentLeasesForRun.mockImplementation(async () => { - expect((await deps.getRun(run.id))?.resultJson).toMatchObject({ workspaceRestoreFailure: "restore_failed", - workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [lease.id] } }); + expect((await deps.getRun(run.id))?.resultJson).toEqual({ workspaceRestored: true, conversationContinuation: "continue_conversation_v1" }); expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0]).toMatchObject({ - status: "pending_cleanup", leasePolicy: "retain_on_failure", metadata: { sandboxStopAndRetain: { leaseId: lease.id } } }); + status: "active", leasePolicy: "ephemeral", metadata: { legacyShutdownWorkspaceSettlement: { state: "settled" } } }); + await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date() }).where(eq(environmentLeases.id, lease.id)); }); const control = createAdapterExecutionControl(); adapterExecutionControls.set(run.id, control); @@ -452,15 +551,24 @@ describePostgres("heartbeat recovery module ownership", () => { try { await vi.waitFor(() => expect(control.controller.signal.aborted).toBe(true)); await vi.advanceTimersByTimeAsync(30_000); - await drain; + expect(await drain).toMatchObject({ interruptedRunIds: [run.id] }); + expect(deps.activeRunExecutionPromises.size).toBe(1); + expect(deps.releaseEnvironmentLeasesForRun).not.toHaveBeenCalled(); + expect(deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled(); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!))).toHaveLength(0); + await db.update(heartbeatRuns).set({ resultJson: { workspaceRestored: true, conversationContinuation: "continue_conversation_v1" } }).where(eq(heartbeatRuns.id, run.id)); + control.finish(); + await Promise.all([...deps.activeRunExecutionPromises]); expect(deps.releaseEnvironmentLeasesForRun).toHaveBeenCalledOnce(); - expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!))) - .toEqual([expect.objectContaining({ evidence: expect.objectContaining({ workspaceRestoreFailure: "restore_failed" }) })]); + expect(deps.scheduleBoundedRetryForRun).toHaveBeenCalledOnce(); + expect(deps.scheduleBoundedRetryForRun).toHaveBeenCalledAfter(deps.releaseEnvironmentLeasesForRun); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!))).toHaveLength(0); } finally { vi.useRealTimers(); control.finish(); adapterExecutionControls.delete(run.id); await drain; + await Promise.all([...deps.activeRunExecutionPromises]); await db.delete(environmentLeases).where(eq(environmentLeases.id, lease.id)); await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!)); await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 4623b7bae36..5d5bcf6d534 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -75,7 +75,7 @@ import { } from "../hot-restart.js"; import { serverVersion } from "../../version.js"; import { adapterExecutionControls, waitForAdapterStop } from "../adapter-execution-control.js"; -import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js"; +import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js"; import type { environmentService } from "../environments.js"; import type { environmentRuntimeService, ProviderResourceDisposition } from "../environment-runtime.js"; @@ -1207,96 +1207,85 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD }), }, ); - if (!interruptedStatus.updated || !interruptedStatus.run) { - // A competing terminal write does not prove the owned adapter settled. + if (interruptedStatus.updated && interruptedStatus.run) interruptedRunIds.push(run.id); + const finishInterruptedRun = async () => { if (control) { - control.controller.abort(new Error(message)); - try { - await waitForAdapterStop(control.settled, Math.max(1, settlementDeadline.getTime() - Date.now())); - await finishLegacyShutdownWorkspaceSettlement(db, run); + await control.settled; + await finishLegacyShutdownWorkspaceSettlement(db, run); + } + if (!interruptedStatus.updated || !interruptedStatus.run) { + if (control) { const current = await getRun(run.id, { includeExecutionEvidence: true }); if (current && HEARTBEAT_RUN_TERMINAL_STATUSES.some(status => status === current.status)) { await releaseEnvironmentLeasesForRun({ runId: current.id, companyId: current.companyId, agentId: current.agentId, status: current.status, failureReason: current.error ?? undefined }); } - } catch { /* The durable fence retains this source after controller loss. */ } - } - return; - } - let interrupted = interruptedStatus.run; - if (control) { - // Keep the exact lease open while its adapter stops the provider and - // copies files back. Provider release closes command admission and - // must not race the adapter's git export or instruction collection. - control.controller.abort(new Error(message)); - try { - await waitForAdapterStop(control.settled, Math.max(1, settlementDeadline.getTime() - Date.now()), { - runId: run.id, adapterType: agent.adapterType, runtimeMode: run.runtimeMode, - abortRequested: true, - }, control); - await finishLegacyShutdownWorkspaceSettlement(db, run); - } catch { - logger.warn({ runId: run.id }, "Shutdown adapter settlement timed out; retaining the sandbox through normal cleanup"); - const sources = await db.select().from(environmentLeases).where(and( - eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), - sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'state' = 'pending'`, - )); - // Stop-only settlement can already have released the source. Claim - // its current fence and pin recovery before classifying or retrying. - for (const source of sources) await allowLegacyShutdownWorkspaceCleanup(db, source, - new Date(Math.max(Date.now(), settlementDeadline.getTime()))); - if (runningProcesses.get(run.id) === running && running) { - await terminateHeartbeatRunProcess({ pid: running.child.pid, processGroupId: running.processGroupId, - graceMs: Math.max(1, running.graceSec) * 1000 }); - if (runningProcesses.get(run.id) === running) runningProcesses.delete(run.id); } + return; } // Settlement can save restore-failure evidence after the shutdown CAS. - // Classification and retry must use the current row, not its snapshot. - interrupted = (await getRun(run.id, { includeExecutionEvidence: true })) ?? interrupted; + let interrupted = (control && await getRun(run.id, { includeExecutionEvidence: true })) ?? interruptedStatus.run; + await setWakeupStatus(run.wakeupRequestId, "cancelled", { + finishedAt: now, + error: null, + }); + interrupted = + (await classifyAndPersistRunLiveness( + interrupted, + parseObject(interrupted.resultJson), + )) ?? interrupted; + + await releaseEnvironmentLeasesForRun({ + runId: interrupted.id, + companyId: interrupted.companyId, + agentId: interrupted.agentId, + status: interrupted.status, + failureReason: interrupted.error ?? undefined, + }); + + const retry = await enqueueProcessLossRetry(interrupted, agent, now); + if (!retry) { + await releaseIssueExecutionAndPromote(interrupted); + } else { + retryRunIds.push(retry.id); + } + + await appendRunEvent(interrupted, { + eventType: "lifecycle", + stream: "system", + level: "warn", + message, + payload: { + signal, + ...(run.processPid ? { processPid: run.processPid } : {}), + ...(run.processGroupId ? { processGroupId: run.processGroupId } : {}), + ...(retry ? { retryRunId: retry.id } : {}), + }, + }); + + await finalizeAgentStatus(run.agentId, "interrupted", message, { + wasFirstHeartbeat: timerClaimWasFirstHeartbeat(run), + }); + }; + if (!control) { + await finishInterruptedRun(); + return; } - await setWakeupStatus(run.wakeupRequestId, "cancelled", { - finishedAt: now, - error: null, + const stopRenewing = maintainLegacyShutdownWorkspaceSettlement(db, run); + control.controller.abort(new Error(message)); + const settlement = finishInterruptedRun().finally(stopRenewing); + const tracked = settlement.catch(() => { + logger.warn({ runId: run.id, errorKind: "shutdown_settlement_failed" }, "Shutdown workspace settlement failed; durable cleanup protection remains in place"); }); - interrupted = - (await classifyAndPersistRunLiveness( - interrupted, - parseObject(interrupted.resultJson), - )) ?? interrupted; - - await releaseEnvironmentLeasesForRun({ - runId: interrupted.id, - companyId: interrupted.companyId, - agentId: interrupted.agentId, - status: interrupted.status, - failureReason: interrupted.error ?? undefined, - }); - - const retry = await enqueueProcessLossRetry(interrupted, agent, now); - if (!retry) { - await releaseIssueExecutionAndPromote(interrupted); - } else { - retryRunIds.push(retry.id); + activeRunExecutionPromises.add(tracked); + void tracked.finally(() => activeRunExecutionPromises.delete(tracked)); + try { + await waitForAdapterStop(settlement, Math.max(1, settlementDeadline.getTime() - Date.now()), { + runId: run.id, adapterType: agent.adapterType, runtimeMode: run.runtimeMode, abortRequested: true, + }, control); + } catch { + logger.warn({ runId: run.id }, "Shutdown workspace settlement continues in the finalizer drain"); } - - await appendRunEvent(interrupted, { - eventType: "lifecycle", - stream: "system", - level: "warn", - message, - payload: { - signal, - ...(run.processPid ? { processPid: run.processPid } : {}), - ...(run.processGroupId ? { processGroupId: run.processGroupId } : {}), - ...(retry ? { retryRunId: retry.id } : {}), - }, - }); - - await finalizeAgentStatus(run.agentId, "interrupted", message, { - wasFirstHeartbeat: timerClaimWasFirstHeartbeat(run), - }); - interruptedRunIds.push(interrupted.id); })); for (const [index, settlement] of settlements.entries()) { if (settlement.status === "rejected") logger.warn({ runId: activeRuns[index]!.run.id, diff --git a/server/src/services/legacy-shutdown-workspace-settlement.ts b/server/src/services/legacy-shutdown-workspace-settlement.ts index fcf2fbd5448..54f9b1a1c00 100644 --- a/server/src/services/legacy-shutdown-workspace-settlement.ts +++ b/server/src/services/legacy-shutdown-workspace-settlement.ts @@ -3,10 +3,53 @@ import { environmentLeases, heartbeatRuns, issues, type Db } from "@paperclipai/ import { adapterExecutionControls } from "./adapter-execution-control.js"; import { legacyControllerBootId } from "./legacy-controller-lease.js"; import { LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js"; +import { logger } from "../middleware/logger.js"; const SCHEMA = "paperclip.legacy-shutdown-workspace-settlement.v1"; type Lease = Pick; type Run = Pick; +const SETTLEMENT_RENEW_MS = 10_000; +const SETTLEMENT_LEASE_MS = 30_000; + +function settlementRenewalFailureKind(error: unknown) { + for (let depth = 0; depth < 4 && error && typeof error === "object"; depth++) { + const value = error as { code?: unknown; cause?: unknown }; + if (value.code === "55P03") return "lock_timeout"; + if (value.code === "57014") return "statement_cancelled"; + if (typeof value.code === "string" && (/^08[0-9A-Z]{3}$/.test(value.code) + || ["ECONNRESET", "ECONNREFUSED", "ETIMEDOUT", "CONNECT_TIMEOUT", "CONNECTION_CLOSED", "CONNECTION_ENDED", "CONNECTION_DESTROYED"].includes(value.code))) { + return "connection_failed"; + } + error = value.cause; + } + return "unknown"; +} + +export function maintainLegacyShutdownWorkspaceSettlement(db: Db, run: Run) { + let stopped = false; + let pending: Promise | null = null; + const renew = () => { + if (stopped || pending) return; + pending = db.transaction(async tx => { + await tx.execute(sql`SET LOCAL lock_timeout = '1s'`); + await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); + await tx.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata}, + '{legacyShutdownWorkspaceSettlement,deadlineAt}', ${JSON.stringify(new Date(Date.now() + SETTLEMENT_LEASE_MS).toISOString())}::jsonb)` }) + .where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'schema' = ${SCHEMA}`, + sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'runId' = ${run.id}`, + sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'controllerBootId' = ${legacyControllerBootId}`, + sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'state' = 'pending'`, + sql`(${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'deadlineAt')::timestamptz > clock_timestamp()`, + )); + }).catch(error => { + logger.warn({ runId: run.id, errorKind: settlementRenewalFailureKind(error) }, "Shutdown workspace settlement lease renewal failed; source remains protected locally"); + }).finally(() => { pending = null; }); + }; + const timer = setInterval(renew, SETTLEMENT_RENEW_MS); + timer.unref?.(); + return async () => { stopped = true; clearInterval(timer); await pending; }; +} export function pendingLegacyShutdownWorkspaceSettlement(lease: Lease) { const value = lease.metadata?.legacyShutdownWorkspaceSettlement; @@ -90,6 +133,7 @@ export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease, // some adapters to settle. This exception never permits provider destruction. if (options.ownerStopOnly && marker.controllerBootId === legacyControllerBootId && adapterExecutionControls.get(run.id)?.controller.signal.aborted) return true; + if (marker.controllerBootId === legacyControllerBootId && adapterExecutionControls.has(run.id)) return false; const deadline = Date.parse(marker.deadlineAt as string); if (!Number.isFinite(deadline) || deadline > now.getTime()) return false; // A killed container cannot certify export. Preserve source and hold in