diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 98aab13805e..5e806dff873 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1834,6 +1834,19 @@ An interrupted conversation does not permanently block its task. For local conve Shutdown, process loss, and provider failure use the existing durable failure retry counter and delay. Ordinary failure recovery permits at most two automatic retries in a failure chain. Accepted-interaction infrastructure recovery retains its existing bounded policy. Repeated scheduler visits reuse the same successor; restarting the server does not reset the counter. After exhaustion, automatic attempts stop. A new explicit user message can start a fresh run and failure budget. Productive max-turn continuation and confirmed workspace waits keep their separate existing semantics. +If shutdown interrupts Paperclip Runner preparation before native execution was +selected, recovery verifies that no provider, process launch, native result or +coordinator exists. The previous controller must have settled or lost its lease. +Required workspace recovery must have an exact saved-file receipt and stopped +leases before the task can continue. Once these checks pass, the server retires +its own unchanged bootstrap-failure block and schedules one bounded retry. +Startup also discovers this obligation when file recovery completed before the +restart. A pending Retry or saved message owns its continuation instead of +starting a competing automatic attempt. Retry uses the same preparation proof +and preserves the recovered workspace. Current task ownership, human blocks, +pauses, approvals, budgets and the existing retry limit remain effective. +Operator Stops retain their explicit continuation requirements. + 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. A run that exceeds that wait stays diff --git a/doc/observability.md b/doc/observability.md index c04aef97125..3c0ba327c29 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -1009,10 +1009,15 @@ a new terminal run transition. The event keeps the actual observed run status, the adapter and the company, agent, task and run references. It uses fixed error text and includes no message bodies, prompts or raw run diagnostics. Working, scheduled-retry, resolved and user-action states do not produce this event. -Repeated reads of the same company/task/run/intent are suppressed by a bounded -process-local cache of the latest 1,000 reports. A server restart or cache eviction -can report an old failure again. This task-state event can accompany a separate -terminal run-failure report; the `failure_source` tag distinguishes them. +Repeated reads of the same company/task/run/intent are suppressed after the SDK +receives a successful HTTP delivery response. The process retains the latest +1,000 confirmed reports. An SDK capture ID alone does not confirm delivery: +processing backpressure, network errors and rate limits can discard the event. +Unconfirmed captures wait 60 seconds before a later task read can try again; +this pending cache also holds at most 1,000 entries and starts no background +timer. A server restart or cache eviction can report an old failure again. +This task-state event can accompany a separate terminal run-failure report; +the `failure_source` tag distinguishes them. The versioned `paperclip` context supplies company and agent IDs and relative API paths for the saved run, events, log, and optional task. A valid HTTP(S) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 7e198cb6a72..ae692d574fd 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -7118,7 +7118,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { agentId, status: "scheduled_retry", scheduledRetryAttempt: 1, - scheduledRetryReason: "transient_failure", + scheduledRetryReason: "interrupted_preparation", }); // The retry keeps the handoff contract: it is still the corrective run, // so a finished attempt that leaves no disposition still escalates. @@ -7329,7 +7329,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { agentId, status: "scheduled_retry", scheduledRetryAttempt: 2, - scheduledRetryReason: "transient_failure", + scheduledRetryReason: "interrupted_preparation", }); const sourceIssue = await db .select({ status: issues.status }) diff --git a/server/src/__tests__/idle-completed-history.test.ts b/server/src/__tests__/idle-completed-history.test.ts index 389422f543c..23b5a4615c5 100644 --- a/server/src/__tests__/idle-completed-history.test.ts +++ b/server/src/__tests__/idle-completed-history.test.ts @@ -6,6 +6,7 @@ import net from "node:net"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; import * as tables from "@paperclipai/db"; +import { isRuntimeExposureAppPort } from "@paperclipai/shared"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { completedHistoryChecks, readIdleSleepSafety } from "../services/idle-sleep-safety.js"; @@ -2122,10 +2123,15 @@ const support = externalDatabaseUrl ? { supported: true } : await getEmbeddedPos expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("unknown"); await runtime.reconcilePersistedRuntimeServicesOnStartup(db); expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("none"); - const server = net.createServer(); - await new Promise((resolve, reject) => { server.once("error", reject); server.listen(0, "127.0.0.1", resolve); }); - const port = (server.address() as net.AddressInfo).port; - await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); + let port = 0; + for (let attempt = 0; attempt < 32; attempt++) { + const server = net.createServer(); + await new Promise((resolve, reject) => { server.once("error", reject); server.listen(0, "127.0.0.1", resolve); }); + port = (server.address() as net.AddressInfo).port; + await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); + if (!isRuntimeExposureAppPort(port)) break; + } + expect(isRuntimeExposureAppPort(port)).toBe(false); const command = "node -e \"require('node:http').createServer((req,res)=>res.end('history')).listen(Number(process.env.PORT),'127.0.0.1')\""; const refs = await runtime.startRuntimeServicesForWorkspaceControl({ db, actor: { id: null, name: "History operator", companyId: f.companyId }, issue: null, diff --git a/server/src/__tests__/run-failure-sentry-real-sdk.test.ts b/server/src/__tests__/run-failure-sentry-real-sdk.test.ts index d4b119e3b35..d7f2d03b90b 100644 --- a/server/src/__tests__/run-failure-sentry-real-sdk.test.ts +++ b/server/src/__tests__/run-failure-sentry-real-sdk.test.ts @@ -50,6 +50,67 @@ afterEach(async () => { }); describe.skipIf(!sentryPackage)("run failure context with the real Sentry SDK", () => { + it("reports a task again after the SDK drops its first capture under processing backpressure", async () => { + const Sentry = sentryPackage!; + const events: Array> = []; + let responseStatus = 200; + vi.stubEnv("SENTRY_DSN_BACKEND", "https://public@example.invalid/1"); + enableRunFailureReporting(); + vi.doMock("../peer-version-check.js", () => ({ checkExactPeerVersions: () => ({ ok: true }) })); + vi.doMock("@sentry/node", () => ({ ...Sentry, + init: (options: Record) => Sentry.init({ ...options, transportOptions: { bufferSize: 1 }, + transport: () => ({ send: async (envelope: [unknown, Array<[{ type: string }, Record]>]) => { + for (const [header, payload] of envelope[1]) if (header.type === "event") events.push(payload); + return { statusCode: responseStatus }; + }, flush: async () => true }), + }), + })); + const { setSentryRunLinkResolver } = await import("../sentry-links.js"); + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + let entered!: () => void; + const started = new Promise(resolve => { entered = resolve; }); + setSentryRunLinkResolver(async () => { entered(); await held; return null; }); + const { captureTaskInterruptionFailure, sentryReady } = await import("../sentry.js"); + await sentryReady; + const event = { companyId: "55555555-5555-4555-8555-555555555555", agentId: "66666666-6666-4666-8666-666666666666", + taskId: "11111111-1111-4111-8111-111111111111", runId: "22222222-2222-4222-8222-222222222222", + agentAdapter: "paperclip_runner", runStatus: "interrupted" as const, intentId: null }; + const dropped = { ...event, taskId: "33333333-3333-4333-8333-333333333333", runId: "44444444-4444-4444-8444-444444444444" }; + try { + captureTaskInterruptionFailure(event); + await started; + captureTaskInterruptionFailure(dropped); + release(); + await Sentry.flush(2000); + expect(events).toHaveLength(1); + expect(events[0].tags).toMatchObject({ run_id: event.runId }); + setSentryRunLinkResolver(null); + const later = Date.now() + 120_000; + const clock = vi.spyOn(Date, "now").mockReturnValue(later); + captureTaskInterruptionFailure(event); + captureTaskInterruptionFailure(dropped); + await Sentry.flush(2000); + expect(events.map(item => (item.tags as Record).run_id)).toEqual([event.runId, dropped.runId]); + const rejected = { ...event, taskId: "77777777-7777-4777-8777-777777777777", runId: "88888888-8888-4888-8888-888888888888" }; + responseStatus = 503; + captureTaskInterruptionFailure(rejected); + await Sentry.flush(2000); + captureTaskInterruptionFailure(rejected); + await Sentry.flush(2000); + expect(events).toHaveLength(3); + clock.mockReturnValue(later + 120_000); + responseStatus = 200; + captureTaskInterruptionFailure(rejected); + await Sentry.flush(2000); + expect(events).toHaveLength(4); + clock.mockReturnValue(later + 240_000); + captureTaskInterruptionFailure(rejected); + await Sentry.flush(2000); + expect(events).toHaveLength(4); + } finally { release(); setSentryRunLinkResolver(null); vi.restoreAllMocks(); } + }); + it("exports event-local routine ancestry for terminal and saved-message failures without caller credentials", async () => { const Sentry = sentryPackage!; const events: Array> = []; diff --git a/server/src/__tests__/run-failure-sentry.test.ts b/server/src/__tests__/run-failure-sentry.test.ts index 85de34731a1..ccea8a5550a 100644 --- a/server/src/__tests__/run-failure-sentry.test.ts +++ b/server/src/__tests__/run-failure-sentry.test.ts @@ -49,6 +49,7 @@ function mockSentryPackage() { vi.doMock("@sentry/node", () => ({ init, + getClient: () => ({ on: vi.fn() }), captureException, close, httpIntegration, diff --git a/server/src/__tests__/sentry.test.ts b/server/src/__tests__/sentry.test.ts index 8771bf694d1..eefc5d42a97 100644 --- a/server/src/__tests__/sentry.test.ts +++ b/server/src/__tests__/sentry.test.ts @@ -55,6 +55,7 @@ function mockSentryPackage() { })); vi.doMock("@sentry/node", () => ({ + getClient: () => ({ on: vi.fn() }), init, captureException, close, diff --git a/server/src/modules/run-dispatch/adapters/postgres.test.ts b/server/src/modules/run-dispatch/adapters/postgres.test.ts index 5b494545f79..3d894ecc45a 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.test.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.test.ts @@ -355,7 +355,7 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => { expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, competingId)))[0]?.status).toBe("running"); }); - it.each(["native_safe_replacement", "native_provider_overloaded"])("does not dispatch %s when the task becomes blocked after scheduling", async retryReason => { + it.each(["native_safe_replacement", "native_provider_overloaded", "interrupted_preparation"])("does not dispatch %s when the task becomes blocked after scheduling", async retryReason => { const { companyId, agentId } = await seedCompanyAndAgent(); const issueId = randomUUID(); await seedIssue({ companyId, issueId, assigneeAgentId: agentId, status: "in_progress" }); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 26266084275..32f2c60c82d 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -153,7 +153,7 @@ function classifyRetryReasonKind(retryReason: string | null): RetryReasonKind { if (retryReason === MAX_TURN_CONTINUATION_RETRY_REASON) return "max_turn_continuation"; if (retryReason === ISSUE_DISPOSITION_REPAIR_RETRY_REASON) return "disposition_repair"; if ((retryReason === "ai_connection_busy" || retryReason === "ai_connection_pool_wait")) return "ai_connection_wait"; - if (retryReason === "native_safe_replacement" || retryReason === "native_provider_overloaded") return "native_safe_replacement"; + if (["native_safe_replacement", "native_provider_overloaded", "interrupted_preparation"].includes(retryReason ?? "")) return "native_safe_replacement"; return "other"; } diff --git a/server/src/sentry.ts b/server/src/sentry.ts index 991023cc81d..8cc63533bbd 100644 --- a/server/src/sentry.ts +++ b/server/src/sentry.ts @@ -209,7 +209,11 @@ function normalizeRunSignal(value: unknown): string | null { * */ export function captureRunFailure(event: RunFailureEvent): boolean { - if (!sentryHandle || readRunFailureReportingMode() === "off") return false; + return captureRunFailureEvent(event) !== null; +} + +function captureRunFailureEvent(event: RunFailureEvent): string | null { + if (!sentryHandle || readRunFailureReportingMode() === "off") return null; const handle = sentryHandle; try { const errorCode = event.errorCode ?? "unknown"; @@ -269,7 +273,7 @@ export function captureRunFailure(event: RunFailureEvent): boolean { exception.stack = typeof diagnostics?.adapter.stackPreview === "string" ? diagnostics.adapter.stackPreview : undefined; } - handle.captureException(exception, { + return handle.captureException(exception, { tags: { event_kind: "run_failure", ...(event.failureSource ? { failure_source: event.failureSource } : {}), @@ -297,22 +301,48 @@ export function captureRunFailure(event: RunFailureEvent): boolean { }, fingerprint: [errorCode, event.agentAdapter, ...(failurePhase ? [failurePhase] : []), ...(rootErrorCode ? [rootErrorCode] : [])], }); - return true; } catch (err) { // eslint-disable-next-line no-console console.error("[paperclip] Sentry captureRunFailure failed", err); - return false; + return null; } } const reportedTaskInterruptions = new Set(); const MAX_REPORTED_TASK_INTERRUPTIONS = 1_000; +const TASK_INTERRUPTION_DELIVERY_RETRY_MS = 60_000; +const pendingTaskInterruptions = new Map(); +const pendingTaskInterruptionEvents = new Map(); + +function forgetPendingTaskInterruption(key: string) { + const pending = pendingTaskInterruptions.get(key); + if (pending && pendingTaskInterruptionEvents.get(pending.eventId) === key) { + pendingTaskInterruptionEvents.delete(pending.eventId); + } + pendingTaskInterruptions.delete(key); +} + +function confirmTaskInterruptionDelivery(eventId: string | undefined, statusCode: number | undefined) { + if (!eventId || statusCode === undefined || statusCode < 200 || statusCode >= 300) return; + const key = pendingTaskInterruptionEvents.get(eventId); + if (!key) return; + forgetPendingTaskInterruption(key); + reportedTaskInterruptions.add(key); + if (reportedTaskInterruptions.size > MAX_REPORTED_TASK_INTERRUPTIONS) { + reportedTaskInterruptions.delete(reportedTaskInterruptions.values().next().value!); + } +} export function captureTaskInterruptionFailure(event: Pick & { intentId: string | null }): void { const key = JSON.stringify([event.companyId, event.taskId, event.runId, event.intentId]); if (reportedTaskInterruptions.has(key)) return; - if (!captureRunFailure({ + const now = Date.now(); + if ((pendingTaskInterruptions.get(key)?.retryAt ?? 0) > now) return; + for (const [pendingKey, pending] of pendingTaskInterruptions) { + if (pending.retryAt <= now) forgetPendingTaskInterruption(pendingKey); + } + const eventId = captureRunFailureEvent({ companyId: event.companyId, agentId: event.agentId, taskId: event.taskId, @@ -322,10 +352,12 @@ export function captureTaskInterruptionFailure(event: Pick MAX_REPORTED_TASK_INTERRUPTIONS) { - reportedTaskInterruptions.delete(reportedTaskInterruptions.values().next().value!); + }); + if (!eventId) return; + pendingTaskInterruptions.set(key, { eventId, retryAt: now + TASK_INTERRUPTION_DELIVERY_RETRY_MS }); + pendingTaskInterruptionEvents.set(eventId, key); + if (pendingTaskInterruptions.size > MAX_REPORTED_TASK_INTERRUPTIONS) { + forgetPendingTaskInterruption(pendingTaskInterruptions.keys().next().value!); } } @@ -439,6 +471,10 @@ async function bootstrapSentry(dsn: string): Promise { Sentry.init(buildSentryInitOptions(dsn, Sentry)); + Sentry.getClient?.()?.on("afterSendEvent", (event: { event_id?: string }, response?: { statusCode?: number }) => { + confirmTaskInterruptionDelivery(event.event_id, response?.statusCode); + }); + sentryHandle = { captureException: (...args) => Sentry.captureException(...args), close: (timeout) => Sentry.close(timeout), diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index 8467918f2ec..e742a2133e6 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -6,16 +6,19 @@ import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-loc import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { canContinueCancelledRun } from "./run-cancellation.js"; import { legacyControllerBootId } from "./legacy-controller-lease.js"; +import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; +import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; type Run = Pick; type Coordinator = typeof nativeRunFinalizations.$inferSelect; export async function isFailedNativePreparation(db: Db, run: Run, coordinator: Coordinator | undefined) { + if (run.status === "interrupted") return !coordinator && isInterruptedNativePreparation(db, run); const recovery = run.resultJson?.executionRecovery as Record | undefined; if (run.status !== "failed" || !run.finishedAt || run.runtimeMode !== "legacy" || run.runtimeModeResolvedAt || run.errorCode !== "setup_failed" || run.executionStage !== "preparing" @@ -34,6 +37,34 @@ export async function isFailedNativePreparation(db: Db, run: Run, coordinator: C return !lease && !result && !await nativeStartupHasExecutionEvidence(db, run); } +export async function isInterruptedNativePreparation(db: Db, run: Run): Promise { + const recovery = run.resultJson?.executionRecovery as Record | undefined; + const settled = recovery?.kind === "bootstrap" && recovery.providerWorkStarted === false; + if (run.status !== "interrupted" || run.errorCode !== "server_shutdown_interrupted" || !run.finishedAt + || run.runtimeMode !== "legacy" || run.runtimeModeResolvedAt || run.executionStage !== "preparing" + || claimedAdapterType(run) !== "paperclip_runner" || run.processPid || run.processGroupId + || run.processStartedAt || run.sessionIdAfter || run.nativeIssueId || run.nativeSessionId + || run.runnerProfileJson?.nativeExecutionInput != null || recovery?.providerWorkStarted === true + || run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date() + || !settled && (!run.controllerBootId || run.controllerBootId === legacyControllerBootId)) return false; + const [coordinator] = await db.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id), + )).limit(1); + const [result] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( + eq(nativeRunResults.companyId, run.companyId), eq(nativeRunResults.runId, run.id), + )).limit(1); + if (coordinator || result || await nativeStartupHasExecutionEvidence(db, run)) return false; + const leases = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + )); + if (leases.some(lease => !lease.releasedAt || lease.status === "pending_cleanup" || lease.cleanupStatus === "failed" + || lease.metadata?.pendingCleanupInFlight === true + || lease.provider !== "local" && !hasRemoteTerminationReceipt(lease))) return false; + if (hasRequiredWorkspaceRecovery(run.resultJson) || run.runnerProfileJson?.remoteWorkspace != null + || run.runnerProfileJson?.nativeWorkspaceSync != null) return hasVerifiedWorkspaceRepair(db, run); + return run.resultJson?.workspaceRestoreFailure == null; +} + /** Candidate for explicit user Retry only, never automatic replay after Stop. */ export async function canRetryStoppedRun(db: Db, run: Run): Promise { if (run.errorCode === "native_session_cleanup_quarantined") return false; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 68bf5fd630a..8215d222fc8 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1439,6 +1439,7 @@ export function heartbeatService( } const runControl = createHeartbeatRunControl(db, { + scheduleBoundedRetry, verifyStartupCleanup: async run => { await verifyNativeStartupCleanup(db, run, () => envOrchestrator.releaseForRun({ heartbeatRunId: run.id, companyId: run.companyId, agentId: run.agentId, status: "expired", diff --git a/server/src/services/heartbeat/__tests__/run-control-fixtures.ts b/server/src/services/heartbeat/__tests__/run-control-fixtures.ts index 561e21a6e5e..781a4b574cf 100644 --- a/server/src/services/heartbeat/__tests__/run-control-fixtures.ts +++ b/server/src/services/heartbeat/__tests__/run-control-fixtures.ts @@ -58,6 +58,7 @@ export function callbacks(db: Db): TestCallbacks { startNextQueuedRunForAgent: vi.fn(), sweepPendingCleanupLeases: vi.fn(), timerClaimWasFirstHeartbeat: vi.fn(() => undefined), + scheduleBoundedRetry: vi.fn(), wakeQueue: { releaseIssueExecution: vi.fn(async () => ({ outcome: { kind: "released" }, postCommitEffects: [] })) }, applyWakeQueuePostCommitEffects: vi.fn(async () => undefined), getSchedulingSuppression: vi.fn(async () => ({ suppressed: false })), diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 60e80818bea..2ea6de7c323 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -428,8 +428,11 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen errorCode: "chat_completion_outbox_owns_retry" as const, issueId: readNonEmptyString(run.contextSnapshot.issueId) }; } const now = opts?.now ?? new Date(); - const retryReason = - opts?.retryReason ?? BOUNDED_TRANSIENT_HEARTBEAT_RETRY_REASON; + const preparationRecovery = parseObject(run.resultJson?.executionRecovery); + const retryReason = opts?.retryReason ?? + (run.status === "interrupted" && run.errorCode === "server_shutdown_interrupted" && + preparationRecovery.kind === "bootstrap" && preparationRecovery.providerWorkStarted === false + ? "interrupted_preparation" : BOUNDED_TRANSIENT_HEARTBEAT_RETRY_REASON); const wakeReason = opts?.wakeReason ?? BOUNDED_TRANSIENT_HEARTBEAT_RETRY_WAKE_REASON; const maxAttempts = Math.max( @@ -567,6 +570,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen : baseSchedule; const requiresIssueGate = + (run.status === "interrupted" && run.errorCode === "server_shutdown_interrupted") || isTransientWorkspacePreparationCode(run.errorCode) || hasConversationContinuationPolicy(run.resultJson) || (retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON) || diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 7555e16fe92..ead808b6623 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -53,7 +53,8 @@ import { isUserCommentContinuationReason, undeliveredLegacyUserCommentIds, } from "../explicit-native-continuation.js"; -import { isCancelledNativeStartup } from "../cancelled-native-startup.js"; +import { isCancelledNativeStartup, isInterruptedNativePreparation } from "../cancelled-native-startup.js"; +import { settleInterruptedNativeBootstrap } from "../legacy-execution-recovery.js"; import { executionBlockerPredicate, getExecutionBlocker, @@ -193,6 +194,7 @@ export interface HeartbeatRunControlDependencies extends Pick["startNextQueuedRunForAgent"]; sweepPendingCleanupLeases: ReturnType["sweepPendingCleanupLeases"]; timerClaimWasFirstHeartbeat: ReturnType["timerClaimWasFirstHeartbeat"]; + scheduleBoundedRetry: ReturnType["scheduleBoundedRetry"]; wakeQueue: Pick, "releaseIssueExecution">; applyWakeQueuePostCommitEffects: (effects: WakeQueuePostCommitEffect[]) => Promise; getSchedulingSuppression: () => Promise<{ suppressed: boolean }>; @@ -569,13 +571,21 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont async function resumeRemoteStopComments(run: NonNullable>>, requestId?: string) { if (!isHeartbeatRunTerminalStatus(run.status) || adapterExecutionControls.has(run.id)) return; + let interruptedPreparation = false; + if (run.status === "interrupted" && run.errorCode === "server_shutdown_interrupted") { + const current = await getRun(run.id, { includeExecutionEvidence: true }); + if (current && await isInterruptedNativePreparation(db, current)) { + const settled = await settleInterruptedNativeBootstrap(db, { run: current, providerDispatchStarted: false }); + if (settled) { run = settled; interruptedPreparation = true; } + } + } const [preparationCoordinator] = run.runtimeMode === "legacy" && run.status === "cancelled" && !run.runtimeModeResolvedAt ? await db.select().from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id), )) : []; const cancelledPreparation = run.runtimeMode === "legacy" && await isCancelledNativeStartup(db, run, preparationCoordinator); - if (run.runtimeMode !== "native" && !cancelledPreparation && + if (run.runtimeMode !== "native" && !cancelledPreparation && !interruptedPreparation && parseObject(run.resultJson?.startupCancellation).beforeNativeSelection !== true && !(await remoteExecutionHasStopped(db, run.companyId, run.id)) && !(await recordedLocalExecutionStopped(db, run))) return; @@ -605,6 +615,14 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont requestId ? eq(agentWakeupRequests.id, requestId) : undefined, sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, )).orderBy(asc(agentWakeupRequests.requestedAt)).limit(50); + if (interruptedPreparation && !pending.length && !requestId && !(await getSchedulingSuppression()).suppressed) { + const [task] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId }).from(issues).where(and( + eq(issues.companyId, run.companyId), eq(issues.id, issueId), + )); + if (task?.assigneeAgentId === run.agentId && ["todo", "in_progress", "in_review"].includes(task.status)) { + await dependencies.scheduleBoundedRetry(run.id); + } + } for (const wake of pending) { if (wake.idempotencyKey?.startsWith("chat-inbound:")) { await resumeDeferredChatWake(db, { companyId: run.companyId, agentId: run.agentId, diff --git a/server/src/services/heartbeat/run-state.ts b/server/src/services/heartbeat/run-state.ts index c2030afa7d1..adc914c07c9 100644 --- a/server/src/services/heartbeat/run-state.ts +++ b/server/src/services/heartbeat/run-state.ts @@ -318,6 +318,9 @@ const heartbeatRunExecutionEvidenceColumn = sql | null>` 'timeoutSource', ${heartbeatRuns.resultJson} -> 'timeoutSource', 'workspaceRestoreFailure', ${heartbeatRuns.resultJson} -> 'workspaceRestoreFailure', 'workspaceRestoreRecovery', ${heartbeatRuns.resultJson} -> 'workspaceRestoreRecovery', + 'executionRecovery', jsonb_build_object( + 'kind', ${heartbeatRuns.resultJson} #> '{executionRecovery,kind}', + 'providerWorkStarted', ${heartbeatRuns.resultJson} #> '{executionRecovery,providerWorkStarted}'), 'executionCancellation', ${heartbeatRuns.resultJson} -> 'executionCancellation', 'nativeCancellation', ${heartbeatRuns.resultJson} -> 'nativeCancellation', 'cancelledByActorType', ${heartbeatRuns.resultJson} -> 'cancelledByActorType', @@ -1144,6 +1147,7 @@ export function createHeartbeatRunState(db: Db) { finishedAt: heartbeatRuns.finishedAt, errorCode: heartbeatRuns.errorCode, error: safeForLegacyEncoding ? heartbeatRunSqlAsciiSafeColumns.error : heartbeatRuns.error, + issueId: heartbeatRuns.issueId, nativeIssueId: heartbeatRuns.nativeIssueId, nativeSessionId: heartbeatRuns.nativeSessionId, processPid: heartbeatRuns.processPid, diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 68e61f9b02a..68138666771 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -20,6 +20,8 @@ import { executionFailureRetryCount } from "./execution-recovery-attempt.js"; import { logActivity } from "./activity-log.js"; import { isSupersededConversationRun } from "./agent-conversations.js"; import { issueService } from "./issues.js"; +import { isInterruptedNativePreparation } from "./cancelled-native-startup.js"; +import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; type Run = typeof heartbeatRuns.$inferSelect; export const LEGACY_RECOVERY_CAUSE = "legacy_execution_requires_reconciliation"; @@ -85,6 +87,7 @@ export function legacyExecutionNeedsReconciliation( /** Admission-wait receipts are only exempt after retained execution evidence agrees. * The synchronous classifier stays conservative for callers without a DB proof. */ export async function legacyExecutionNeedsReconciliationWithEvidence(db: Db, run: Run): Promise { + if (run.status === "interrupted" && await isInterruptedNativePreparation(db, run)) return false; return await hasUnrestoredRemoteWorkspace(db, run) || (legacyExecutionNeedsReconciliation(run) && !(await isPreDispatchReviewWaitVerified(db, run)) && !(await isPreDispatchExecutionWaitVerified(db, run))); @@ -306,9 +309,6 @@ export async function terminalizeLegacyExecution(input: { }); } -/** Called only by the exact executor's finally block after preparation/cleanup - * settles. A shutdown can win the terminal-status race before the setup catch - * records that no provider started. Retire only that mistaken recovery hold. */ export async function settleInterruptedNativeBootstrap( db: Db, input: { run: Run; providerDispatchStarted: boolean }, ): Promise { @@ -322,7 +322,8 @@ export async function settleInterruptedNativeBootstrap( run.contextSnapshot?.issueId !== task.id || run.status !== "interrupted" || run.errorCode !== "server_shutdown_interrupted" || claimedAdapterType(run) !== "paperclip_runner" || run.processPid || run.processGroupId || - hasWorkspaceRestoreFailure(run.resultJson) || executionFailureRetryCount(run) >= 2) return null; + executionFailureRetryCount(run) >= 2) return null; + if (hasWorkspaceRestoreFailure(run.resultJson) && !await hasVerifiedWorkspaceRepair(tx as unknown as Db, run)) return null; const [native] = await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)).limit(1); const [lease] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), @@ -357,7 +358,7 @@ export async function settleInterruptedNativeBootstrap( }, updatedAt: now }).where(eq(heartbeatRuns.id, run.id)).returning(); if (holds.length) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: task.companyId, issueId: task.id }); const resolved = holds.length ? await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "false_positive", - resolutionNote: "The interrupted executor finished preparation and cleanup without starting a provider.", + resolutionNote: "The interrupted preparation has no provider execution and its cleanup is complete.", resolvedAt: now, updatedAt: now, // An automatic disposition may already have resolved the incident but // left replay blocked. Preserve its history while correcting that verdict. diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 4b7f5390a5c..32b57491fa3 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -39,7 +39,7 @@ import { createSystemRecovery, failSystemRecovery, SystemRecoveryFailure } from import { requestTaskRetry } from "./task-retry.js"; import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; import { execFile } from "node:child_process"; -import { generateKeyPairSync, randomUUID, sign } from "node:crypto"; +import { generateKeyPairSync, randomBytes, randomUUID, sign } from "node:crypto"; import { cp, mkdir, mkdtemp, readFile, realpath, rm, writeFile } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -248,6 +248,154 @@ describe("automatic retained workspace recovery", () => { return { ...f, source, workspaceId, resultId, git, commit }; } + async function interruptedRemotePreparation() { + const f = await acceptedRemoteExport(); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.delete(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)); + const recovery = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] }; + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null, completionContractId: null, + status: "interrupted", executionStage: "preparing", errorCode: "server_shutdown_interrupted", + controllerBootId: randomUUID(), + error: "Interrupted by graceful server shutdown (SIGTERM)", + resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: recovery }, + }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(issues).set({ statusVersion: 2 }).where(eq(issues.id, f.issueId)); + await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation", + fingerprint: `legacy-execution:${f.runId}`, status: "resolved", outcome: "blocked", + evidence: { runId: f.runId, workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: recovery, + automaticRecovery: { policy: "preserve_without_replay_v1", runId: f.runId, replay: "blocked", actionOutcome: "unknown" }, + nativeBootstrapFailureBlock: { runId: f.runId, previousStatus: "in_progress", statusVersion: 2 } }, + }).where(eq(issueRecoveryActions.id, f.actionId)); + await adoptSystemRecovery(db, f); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.source.agentId, status: "running" }); + return f; + } + + it("admits HTTP Retry once after an interrupted Runner preparation recovers its remote workspace", async () => { + const f = await interruptedRemotePreparation(); + await recoverRetainedWorkspace(f.input); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + const intent = { observedRunId: f.runId, idempotencyKey: randomUUID() }; + const app = boardRecoveryApp(); + const receipt = (await request(app).post(`/api/issues/${f.issueId}/retry`).send(intent).expect(202)).body; + const heartbeat = heartbeatService(db); + await Promise.all([heartbeat.resumeTaskRetry(f.companyId, receipt.requestId), + heartbeat.resumeTaskRetry(f.companyId, receipt.requestId)]); + expect((await request(app).post(`/api/issues/${f.issueId}/retry`).send(intent).expect(202)).body.requestId).toBe(receipt.requestId); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + eq(heartbeatRuns.status, "queued"))); + expect(successors).toHaveLength(1); + expect(successors[0].contextSnapshot).toMatchObject({ previousRunId: f.runId }); + expect(await readFreshWorkspaceReadiness(db, successors[0])).toMatchObject({ mode: "resume_preserved_workspace" }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + expect((await run(f.runId)).status).toBe("interrupted"); + const remote = (await run(f.runId)).runnerProfileJson!.remoteWorkspace as { snapshotId: string }; + const saved = await createSavedWorkspaceLoader(db)({ companyId: f.companyId, workspaceId: f.workspaceId, + snapshotId: remote.snapshotId }); + try { expect(await readFile(path.join(saved.root, "saved.txt"), "utf8")).toBe("finished work\n"); } + finally { await saved.cleanup(); } + }, 30_000); + + it.each([false, true])("automatically retries interrupted preparation after preservation, including recovered history at startup=%s", async alreadyRecovered => { + const f = await interruptedRemotePreparation(); + if (alreadyRecovered) await recoverRetainedWorkspace(f.input); + const restarted = createDb(connectionString); + try { + const heartbeat = heartbeatService(restarted); + await Promise.all(await reconcileRetainedWorkspaces(restarted, f.environmentRuntime, + heartbeat.resumeRemoteStopComments, true) ?? []); + await Promise.all([heartbeat.resumeRemoteStopComments(await run(f.runId)), + heartbeat.resumeRemoteStopComments(await run(f.runId))]); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + eq(heartbeatRuns.retryOfRunId, f.runId))); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: 1 }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("in_progress"); + expect((await run(f.runId)).status).toBe("interrupted"); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + } finally { await restarted.$client.end(); } + }, 30_000); + + it.each(["local", "none"])("continues verified interrupted preparation with %s leases", async provider => { + const f = await interruptedRemotePreparation(); + await db.update(heartbeatRuns).set({ runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + resultJson: null }).where(eq(heartbeatRuns.id, f.runId)); + if (provider === "none") await db.delete(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + else await db.update(environmentLeases).set({ provider: "local", metadata: {} }).where(eq(environmentLeases.id, f.lease.id)); + const heartbeat = heartbeatService(db); + await heartbeat.resumeRemoteStopComments(await run(f.runId)); + const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.runId)); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: 1 }); + }, 30_000); + + it.each(["provider_started", "provider_dispatch_recorded", "runtime_selected", "live_controller", "wrong_snapshot", "operator_stop", + "retry_exhausted", "human_reblock", "paused", "reassigned"] as const)( + "keeps the interrupted preparation continuation behind %s", async gate => { + const f = await interruptedRemotePreparation(); + await recoverRetainedWorkspace(f.input); + const source = await run(f.runId); + if (gate === "provider_started") await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, + runId: f.runId, agentId: source.agentId, seq: 1, eventType: "native.process_start_requested", stream: "system" }); + if (gate === "provider_dispatch_recorded") await db.update(heartbeatRuns).set({ resultJson: { ...source.resultJson, + executionRecovery: { kind: "bootstrap", providerWorkStarted: true }, output: randomBytes(128 * 1024).toString("base64"), + } }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "runtime_selected") await db.update(heartbeatRuns).set({ runtimeModeResolvedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "live_controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "wrong_snapshot") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...source.runnerProfileJson, + remoteWorkspace: { ...(source.runnerProfileJson!.remoteWorkspace as Record), snapshotId: randomUUID() }, + } }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "operator_stop") await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "operator_interrupted" }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "retry_exhausted") await db.update(heartbeatRuns).set({ scheduledRetryAttempt: 2 }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "human_reblock") await db.update(issues).set({ statusVersion: 3 }).where(eq(issues.id, f.issueId)); + if (gate === "paused") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, source.agentId)); + if (gate === "reassigned") await db.update(issues).set({ assigneeAgentId: null, assigneeUserId: "board" }).where(eq(issues.id, f.issueId)); + const heartbeat = heartbeatService(db); + await heartbeat.resumeRemoteStopComments(await run(f.runId)); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + eq(heartbeatRuns.retryOfRunId, f.runId)))).toHaveLength(0); + if (["provider_started", "provider_dispatch_recorded", "runtime_selected", "live_controller", "wrong_snapshot", "operator_stop"].includes(gate)) { + const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, + actorId: "local-board", actorSource: "local_implicit", observedRunId: f.runId, idempotencyKey: randomUUID() }); + await heartbeat.resumeTaskRetry(f.companyId, receipt.requestId!); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + eq(heartbeatRuns.status, "queued")))).toHaveLength(0); + } + if (gate === "human_reblock" || gate === "reassigned") { + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("blocked"); + } + expect((await run(f.runId)).error).toBe(source.error); + }, 30_000); + + it("uses a pending Retry as the sole successor after interrupted preparation recovers", async () => { + const f = await interruptedRemotePreparation(); + const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, + actorId: "local-board", actorSource: "local_implicit", observedRunId: f.runId, idempotencyKey: randomUUID() }); + await recoverRetainedWorkspace(f.input); + const heartbeat = heartbeatService(db); + await Promise.all([heartbeat.resumeRemoteStopComments(await run(f.runId)), + heartbeat.resumeTaskRetry(f.companyId, receipt.requestId!)]); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + inArray(heartbeatRuns.status, ["queued", "scheduled_retry"]))); + expect(successors).toHaveLength(1); + expect(successors[0].status).toBe("queued"); + expect(await readFreshWorkspaceReadiness(db, successors[0])).toMatchObject({ mode: "resume_preserved_workspace" }); + }, 30_000); + + it("cancels the interrupted preparation retry when a user blocks the task before it starts", async () => { + const f = await interruptedRemotePreparation(); + await recoverRetainedWorkspace(f.input); + const heartbeat = heartbeatService(db); + await heartbeat.resumeRemoteStopComments(await run(f.runId)); + const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.runId)); + expect(successor.status).toBe("scheduled_retry"); + await db.update(issues).set({ status: "blocked", statusVersion: 4 }).where(eq(issues.id, f.issueId)); + await heartbeat.promoteDueScheduledRetries(new Date(Date.now() + 300_000)); + expect(await run(successor.id)).toMatchObject({ status: "cancelled", errorCode: "issue_blocked" }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("blocked"); + }, 30_000); + it.each([false, true])("keeps Retry behind an accepted remote export and resumes only its saved result (review pending: %s)", async reviewPending => { const f = await acceptedRemoteExport(); const { source, workspaceId, resultId, git, commit } = f; diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index 83a1b1cc674..cdadb3efba4 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -446,6 +446,17 @@ export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedO where ${agentWakeupRequests.companyId} = ${issueRecoveryActions.companyId} and ${agentWakeupRequests.status} = 'deferred_issue_execution' and ${agentWakeupRequests.requestedByActorType} = 'user' and ${agentWakeupRequests.payload}->>'issueId' = ${issueRecoveryActions.sourceIssueId}::text)`; + const interruptedPreparation = sql`(${heartbeatRuns.runtimeMode} = 'legacy' + and ${heartbeatRuns.status} = 'interrupted' and ${heartbeatRuns.errorCode} = 'server_shutdown_interrupted' + and ${heartbeatRuns.runtimeModeResolvedAt} is null and ${heartbeatRuns.executionStage} = 'preparing' + and ${heartbeatRuns.runnerProfileJson}#>>'{adapterDispatch,adapterType}' = 'paperclip_runner' + and ${issueRecoveryActions.evidence}#>>'{automaticRecovery,replay}' = 'blocked' + and exists (select 1 from ${issues} where ${issues.companyId} = ${heartbeatRuns.companyId} + and ${issues.id} = ${issueRecoveryActions.sourceIssueId} and ${issues.assigneeAgentId} = ${heartbeatRuns.agentId} + and ${issues.status} in ('todo', 'in_progress', 'in_review', 'blocked') + and ${issues.executionRunId} is null and ${issues.checkoutRunId} is null + and (${issues.status} <> 'blocked' or + ${issueRecoveryActions.evidence}#>>'{nativeBootstrapFailureBlock,statusVersion}' = ${issues.statusVersion}::text)))`; return db.select({ action: issueRecoveryActions, run: heartbeatRuns }).from(issueRecoveryActions) .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, issueRecoveryActions.companyId), eq(heartbeatRuns.id, sql`case when ${issueRecoveryActions.evidence}->>'runId' ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' @@ -454,12 +465,12 @@ export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedO workspaceRecoverySourceEligible(true), unfinishedOnly ? sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence')` : undefined, obligationsOnly ? sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence') - or (${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' = 'complete' and ${cleanup}) or ${waitingInput})` : undefined, + or (${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' = 'complete' and (${cleanup} or ${interruptedPreparation})) or ${waitingInput})` : undefined, sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence', 'complete', 'exhausted') and (${issueRecoveryActions.evidence}#>'{systemRecovery,nextAttemptAt}' = 'null'::jsonb or ${issueRecoveryActions.evidence}#>'{systemRecovery,nextAttemptAt}' <= to_jsonb(${Date.now()}::bigint)))`, sql`(${issueRecoveryActions.evidence}#>>'{workspaceRecovery,status}' is distinct from 'recovered' or - ${cleanup} or ${waitingInput})`, + ${cleanup} or ${waitingInput} or ${interruptedPreparation})`, cursor ? gt(issueRecoveryActions.id, cursor) : undefined, inFlightRunIds.length ? notInArray(heartbeatRuns.id, inFlightRunIds) : undefined, )).orderBy(asc(issueRecoveryActions.id)).limit(5);