diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 326593a88f4..c25c4e1449d 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1644,6 +1644,45 @@ recording the proof and successor lineage. Retained provider files are not edite Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. Exact-checkpoint resumes retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates. +A due failed remote retry can retain its provider PID after heartbeat completion +releases the task execution lock. Both remote workspace reference formats and +exact company/task/run plugin task leases route that retry to remote restart +recovery. Plugin leases do not require a workspace reference. A PID from another +host is never checked or cleared through the local process table. Before remote inspection, recovery +can reclaim the original task execution only for an explicit exact-checkpoint +retry with a fenced previous controller, matching durable authority and ordinary +resumption eligibility. A newer run for the task, another execution or checkout, +Stop, pause, changed assignment or an unresolved dependency prevents the claim. +The task lock, controller claim, run status and recovery event commit together. +A new failed retry consumes the existing provider attempt budget; replacing an +active controller does not. The claim permits only inspection and attachment to +the recorded remote execution. Missing live provider evidence +still prevents replay or replacement. Admission holds use the existing system +recovery policy: five backoff retries within a 30-minute window, with persisted +jitter and a budget separate from provider attempts. A newer source or an ended +task cannot become eligible and exhausts this automatic policy immediately. +Exhaustion gives the board a manual reconciliation action and stops retry +deadlines. It preserves execution and checkout bindings, the provider PID and +checkpoint, and does not request environment settlement or file preservation: +none of those facts prove the old provider stopped. Live leases stay read-only +until their own deadline. Stale-lock sweeps, terminal checkout cleanup, stale +adoption and queue release share this manual ownership hold. Terminal status +alone cannot clear it; the existing receipt contract must confirm every exact +remote lease stopped before ordinary cleanup and checkout can progress. Missing +lease records do not establish termination. An explicit Stop can still reserve +and fence this exact held execution through the existing audited cancellation +path. The hold survives that Stop's intent and acknowledgement until remote +termination is confirmed. Normal Retry can register the exact source's existing +stop-and-preserve recovery. It waits for verified termination and file recovery +before admitting one continuation from the preserved workspace. The run log +records a changed hold reason once instead +of repeating the same warning on every sweep. A historical provider-budget-exhausted retry can +release its task lock only after the same admission, controller fence and exact +remote authority checks pass. Existing terminal recovery then preserves the +provider evidence and gives the board a reconciliation action; it authorizes no +new provider work. A completed turn without a semantic result retains its +existing bounded disposition-only recovery policy. + A bootstrap retry requires durable evidence that no provider session, provider event, or checkpoint exists. A runner report that `session.open` failed or was rejected qualifies, because that command runs before any prompt. An diff --git a/doc/run-log-events.md b/doc/run-log-events.md index bd935192835..d878213654d 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -146,8 +146,18 @@ that restores a released task execution binding. The claim locks the task, coordinator, and original run in that order. It preserves the assignee and checks current execution/checkout owners, other live work, dependencies, pause holds, and stop/successor records before restoring the binding. Attachment authorization -keeps its existing requirement for that exact binding. Skipped retry candidates -wait 30 seconds before recheck so they cannot fill every bounded claim batch. +keeps its existing requirement for that exact binding. Remote retry claims also +require the exact saved controller authority and reject a newer task run. The +event does not prove that a remote provider survived or permit provider replay. +Skipped local retry candidates wait 30 seconds before recheck. Remote admission +holds use the existing bounded system-recovery backoff independently of provider +attempts. Permanent ineligibility or policy exhaustion records the terminal +automatic recovery disposition and board action without releasing the retained +execution binding, provider, or checkpoint. These settled holds schedule no +further retry. Repeated blocked visits record only a changed hold reason. +The same durable hold prevents background stale-lock cleanup and queue release +from clearing either task binding until the remote termination receipt contract +is satisfied. A terminal heartbeat status is not a provider-stop receipt. Undated retries sort first. A failed candidate cannot discard earlier committed claims. Queued waiters do not block the original execution owner from recovery; they still wait for that owner to settle before normal queue promotion. @@ -290,8 +300,13 @@ create a task warning, recovery action, Telemetry event, or OpenTelemetry export ## Native Restart Recovery Run-Log Event -Paperclip writes a `native.recovery.transition` event for every native restart -classification and for graceful restart suspension. This immutable run-log +Paperclip writes a `native.recovery.transition` event for native restart +classifications and for graceful restart suspension. A blocked remote retry +records a changed hold reason once; repeated visits with the same hold reason, +an existing future deadline, or a live lease do not append another transition. +Exhausting the automatic admission policy records its terminal disposition even +when the underlying hold reason remains the same. +This immutable run-log record lets operators reconstruct recovery decisions without exporting data to Paperclip Telemetry or OpenTelemetry. diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 8cfe44af184..121befc8345 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -14,6 +14,7 @@ import { isCompletedOnboardingHandoffWake } from "../../../services/chat-complet import { instanceSettingsService } from "../../../services/instance-settings.js"; import { currentConversationCommentCondition } from "../../../services/agent-conversations.js"; import { getExecutionBlocker, hasPendingRunCleanup } from "../../../services/execution-blocker.js"; +import { nativeManualAdmissionHoldsExecution } from "../../../services/native-runtime/native-runner-ownership.js"; import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; import { readReleaseRun, type ReleaseRun } from "./release-run-read.js"; import type { Db } from "@paperclipai/db"; @@ -1189,6 +1190,9 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd : []; if (!finalized) return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; } + if (await nativeManualAdmissionHoldsExecution(tx, { companyId: input.companyId, runId: run.id })) { + return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; + } if (input.resumeStrandedQueue && await hasPendingRunCleanup(tx, input.companyId, run.id)) { return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; } diff --git a/server/src/services/agent-name-key.ts b/server/src/services/agent-name-key.ts new file mode 100644 index 00000000000..561a0de1ecf --- /dev/null +++ b/server/src/services/agent-name-key.ts @@ -0,0 +1,5 @@ +export function normalizeAgentNameKey(value: string | null | undefined) { + if (typeof value !== "string") return null; + const normalized = value.trim().toLowerCase(); + return normalized.length > 0 ? normalized : null; +} diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index aa703738606..ba949595d33 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -14,6 +14,7 @@ import { unreconciledNativeExportPredicate } from "./native-runtime/native-works import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; import { hasVerifiedWorkspaceRepair, workspaceRecoveryDisplay } from "./workspace-recovery-receipt.js"; +import { nativeManualAdmissionHoldsExecution } from "./native-runtime/native-runner-ownership.js"; const blockerContextKeys = ["issueId", "taskId", "source", "connectionIntentResolution", "interactionKind", "interactionStatus", "mutation", "interactionId"] as const; @@ -67,6 +68,7 @@ const blockerRunFields = { }; export async function hasPendingRunCleanup(db: Db, companyId: string, runId: string): Promise { + if (await nativeManualAdmissionHoldsExecution(db, { companyId, runId })) return true; const [state] = await db.execute<{ pending: boolean }>(sql`SELECT ${heartbeatRuns.runnerProfileJson}->>'nativeWorkspaceFinalizationOwner' IS NOT NULL OR EXISTS (SELECT 1 FROM ${environmentLeases} diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index dc84e83692c..78812e8c239 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -50,6 +50,7 @@ import { } from "../explicit-native-continuation.js"; import { canRetryStoppedRun } from "../cancelled-native-startup.js"; import { getExecutionBlocker, hasPendingRunCleanup } from "../execution-blocker.js"; +import { nativeManualAdmissionHoldsExecution } from "../native-runtime/native-runner-ownership.js"; import { validateNativeWorkspaceReconstruction } from "../native-runtime/native-workspace-reconstruction.js"; import { isConversationAdapter } from "../conversation-continuation.js"; import { recordExecutionWait } from "../execution-wait.js"; @@ -3493,9 +3494,12 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende .where(eq(heartbeatRuns.id, issue.executionRunId)) .then((rows) => rows[0] ?? null) : null; + const manualExecutionHold = activeExecutionRun && await nativeManualAdmissionHoldsExecution(tx as unknown as Db, + { companyId: issue.companyId, runId: activeExecutionRun.id, issueId: issue.id }); if ( activeExecutionRun && + !manualExecutionHold && !EXECUTION_PATH_HEARTBEAT_RUN_STATUSES.includes( activeExecutionRun.status as (typeof EXECUTION_PATH_HEARTBEAT_RUN_STATUSES)[number], ) @@ -3527,6 +3531,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // normally against the now-running holder. if ( activeExecutionRun && + !manualExecutionHold && activeExecutionRun.status !== "running" && issue.assigneeAgentId && activeExecutionRun.agentId !== issue.assigneeAgentId diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index 3c92e5ed84b..ddc034f7259 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -555,7 +555,7 @@ describePostgres("heartbeat recovery module ownership", () => { 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", + status: "active", leasePolicy: state === "reuse_by_environment" ? state : "ephemeral", updatedAt: new Date(0), metadata: { driver: "sandbox", sandboxProviderPlugin: true, pluginId: randomUUID(), remoteCwd: "/work" }, }).returning(); await db.update(heartbeatRuns).set({ status: "failed", runtimeMode: "native", nativeIssueId: run.issueId, @@ -1266,6 +1266,53 @@ describePostgres("heartbeat recovery module ownership", () => { await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); }); + it("schedules deferred remote retries without repeating unchanged transition events", async () => { + const run = await seedRun(); + const deps = callbacks(db), retryAt = new Date(Date.now() + 30_000); + deps.getRun.mockResolvedValue(run); + const held = { kind: "blocked" as const, runId: run.id, reason: "native_retry_admission_ineligible", retryAt }; + native.claims.mockResolvedValue([{ ...held, transitionUnchanged: false }]); + const recovery = createHeartbeatRecovery(db, deps); + await recovery.recoverNativeRunsAfterRestart(new Date(), [run.id], { skipDelegatedWork: true }); + expect(deps.scheduleNativeSessionResumeDispatch).toHaveBeenCalledWith(run.id, retryAt, "restart"); + expect(deps.appendRunEvent).toHaveBeenCalledTimes(1); + native.claims.mockResolvedValue([{ ...held, transitionUnchanged: true }]); + await recovery.recoverNativeRunsAfterRestart(new Date(), [run.id], { skipDelegatedWork: true }); + expect(deps.appendRunEvent).toHaveBeenCalledTimes(1); + expect(deps.executeRun).not.toHaveBeenCalled(); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + }); + + it.each(["nativeWorkspaceSync", "remoteWorkspace"])("routes a due %s retry to remote ownership verification", async reference => { + const run = await seedRun(); + const leaseId = randomUUID(), workspaceId = randomUUID(); + await db.update(agents).set({ adapterType: "paperclip_runner", status: "active" }).where(eq(agents.id, run.agentId)); + await db.update(issues).set({ status: "in_progress", assigneeAgentId: run.agentId, + executionRunId: null, checkoutRunId: null }).where(eq(issues.id, run.issueId!)); + const location = { workspaceId, leaseId, providerLeaseId: "retained-sandbox", remoteCwd: "/workspace", state: "prepared" }; + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId, status: "failed", + processPid: process.pid, processStartedAt: new Date(), runnerProfileJson: { + [reference]: reference === "remoteWorkspace" + ? { ...location, schema: "paperclip.remote-workspace-run.v1", epoch: 1, snapshotId: null } + : { ...location, schema: "paperclip.native-workspace-sync/v3", descriptorSha256: "a".repeat(64), + baselineSha256: "b".repeat(64), finalHostSha256: null, resourceDisposition: "destroy" }, + } }).where(eq(heartbeatRuns.id, run.id)); + await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!, + phase: "retryable_failure", nextAttemptAt: new Date(0), + failureDetail: { recoveryMode: "exact_checkpoint_resume" } }); + const deps = callbacks(db); + deps.getRun.mockImplementation(async id => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0] ?? null); + native.claims.mockImplementation(claimNativeRestartRecoveries); + await createHeartbeatRecovery(db, deps).resumeRetryableNativeProcessSessions({ runId: run.id }); + expect(native.claims).toHaveBeenCalledWith(expect.objectContaining({ runIds: [run.id] })); + expect(deps.executeRun).not.toHaveBeenCalled(); + expect((await deps.getRun(run.id))?.processPid).toBe(process.pid); + expect((await db.select().from(issues).where(eq(issues.id, run.issueId!)))[0].executionRunId).toBeNull(); + await Promise.all([...deps.activeRunExecutionPromises]); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + }); + it("retains a live remote owner's actual lease deadline without dispatching competing work", async () => { const deps = callbacks(db); const runId = randomUUID(), retryAt = new Date(Date.now() + 60_000); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index bf701c8d55d..d6ccf33bdc6 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1013,12 +1013,13 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD disposition.kind === "bootstrap_incomplete", ); for (const disposition of dispositions) { - if (disposition.kind === "awaiting_evidence" && disposition.retryAt) { + if ((disposition.kind === "awaiting_evidence" || disposition.kind === "blocked") && disposition.retryAt) { scheduleNativeSessionResumeDispatch(disposition.runId, disposition.retryAt, "restart"); } // No ownership decision was made for a locked row. Writing an event // would wait on that same row and defeat the non-blocking startup scan. if (disposition.kind === "awaiting_evidence" && disposition.reason === "recovery_rows_busy") continue; + if (disposition.kind === "blocked" && disposition.transitionUnchanged) continue; // Claims are already committed and must still dispatch, so one failed // audit write cannot stop the rest of the batch. const auditFailed = (error: unknown) => { diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 07bf4659789..011f87b588c 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -1,4 +1,5 @@ import { notifyAttentionRetentionWork } from "../retention-work-notifications.js"; +import { normalizeAgentNameKey } from "../agent-name-key.js"; import { notifyStatusCardIssueChange } from "../status-card-notifications.js"; import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js"; @@ -368,11 +369,7 @@ export function computeBoundedTransientHeartbeatRetrySchedule( }; } -export function normalizeAgentNameKey(value: string | null | undefined) { - if (typeof value !== "string") return null; - const normalized = value.trim().toLowerCase(); - return normalized.length > 0 ? normalized : null; -} +export { normalizeAgentNameKey }; function readNonEmptyString(value: unknown): string | null { return typeof value === "string" && value.trim().length > 0 ? value : null; diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 461fcffb616..11f25b30457 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -127,7 +127,7 @@ import { } from "../native-runtime/index.js"; import { runningProcesses } from "../../adapters/index.js"; import { parseObject } from "../../adapters/utils.js"; -import { isNativeRunnerOwnershipHeld, nativeRunnerOwnershipNotHeldCondition } from "../native-runtime/native-runner-ownership.js"; +import { isNativeRunnerOwnershipHeld, nativeManualAdmissionOwnerColumns, nativeRunnerOwnershipNotHeldCondition } from "../native-runtime/native-runner-ownership.js"; import { readNativeStartupCleanup, acknowledgeNativeStartupCleanup } from "../native-runtime/native-startup-cleanup.js"; import { withCurrentBudgetEnforcement, @@ -1314,7 +1314,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont const pendingNativeRetry = run.runtimeMode === "native" && run.status === "failed" && Boolean(run.nativeIssueId) && ( caseHandoffRetryClaimed || Boolean(options.cancellationRequestId) || await db - .select({ phase: nativeRunFinalizations.phase, failureCode: nativeRunFinalizations.failureCode }) + .select(nativeManualAdmissionOwnerColumns) .from(nativeRunFinalizations) .where( and( diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index c70d4d8047d..2b5fe72ab2b 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -57,6 +57,7 @@ import { Buffer } from "node:buffer"; import { isDeepStrictEqual } from "node:util"; import { createHash, randomUUID } from "node:crypto"; import { isExplicitContinuationRetryClaim } from "./explicit-continuation-retry-claim.js"; +import { nativeManualAdmissionHoldsExecution } from "./native-runtime/native-runner-ownership.js"; import { markdownToPlainText, parseMarkdown } from "chat"; import { listExternalReviewWaits } from "./native-runtime/external-review-wait.js"; import { @@ -7788,14 +7789,16 @@ export function issueService(db: Db) { ]); const stale = !existingRun || TERMINAL_HEARTBEAT_RUN_STATUSES.has(existingRun.status); - if (isExplicitContinuationRetryClaim(lockedIssue, existingRun)) { + if (isExplicitContinuationRetryClaim(lockedIssue, existingRun) || await nativeManualAdmissionHoldsExecution(tx as unknown as Db, + { companyId: lockedIssue.companyId, runId: input.expectedCheckoutRunId, issueId: input.issueId })) { return { adopted: null, latest: lockedIssue }; } if (lockedIssue.executionRunId && lockedIssue.executionRunId !== input.expectedCheckoutRunId) { const executionRun = await tx.select().from(heartbeatRuns) .where(eq(heartbeatRuns.id, lockedIssue.executionRunId)).for("update") .then(rows => rows[0] ?? null); - if (isExplicitContinuationRetryClaim(lockedIssue, executionRun)) { + if (isExplicitContinuationRetryClaim(lockedIssue, executionRun) || await nativeManualAdmissionHoldsExecution(tx as unknown as Db, + { companyId: lockedIssue.companyId, runId: lockedIssue.executionRunId, issueId: input.issueId })) { return { adopted: null, latest: lockedIssue }; } } @@ -7925,7 +7928,8 @@ export function issueService(db: Db) { .where(eq(heartbeatRuns.id, issue.executionRunId)) .then((rows) => rows[0] ?? null); if (run && !TERMINAL_HEARTBEAT_RUN_STATUSES.has(run.status)) return false; - if (isExplicitContinuationRetryClaim(issue, run)) return false; + if (isExplicitContinuationRetryClaim(issue, run) || await nativeManualAdmissionHoldsExecution(tx as unknown as Db, + { companyId: issue.companyId, runId: issue.executionRunId, issueId })) return false; await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId }); await notifyNativeStatusReassessment(tx, { companyId: issue.companyId, issueId }); @@ -7980,7 +7984,8 @@ export function issueService(db: Db) { .where(eq(heartbeatRuns.id, issue.checkoutRunId)) .then((rows) => rows[0] ?? null); if (run && !TERMINAL_HEARTBEAT_RUN_STATUSES.has(run.status)) return false; - if (isExplicitContinuationRetryClaim(issue, run)) return false; + if (isExplicitContinuationRetryClaim(issue, run) || await nativeManualAdmissionHoldsExecution(tx as unknown as Db, + { companyId: issue.companyId, runId: issue.checkoutRunId, issueId })) return false; if ( issue.executionRunId && @@ -7999,7 +8004,8 @@ export function issueService(db: Db) { !TERMINAL_HEARTBEAT_RUN_STATUSES.has(executionRun.status) ) return false; - if (isExplicitContinuationRetryClaim(issue, executionRun)) return false; + if (isExplicitContinuationRetryClaim(issue, executionRun) || await nativeManualAdmissionHoldsExecution(tx as unknown as Db, + { companyId: issue.companyId, runId: issue.executionRunId, issueId })) return false; } await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId }); @@ -12375,6 +12381,10 @@ export function issueService(db: Db) { } const staleExecutionRunId = current.executionRunId; const adopted = await db.transaction(async tx => { + await tx.execute(sql`select ${issues.id} from ${issues} where ${issues.id} = ${id} + and ${issues.companyId} = ${issueCompany.companyId} for update`); + if (await nativeManualAdmissionHoldsExecution(tx as unknown as Db, + { companyId: issueCompany.companyId, runId: staleExecutionRunId, issueId: id })) return null; await notifyStatusCardIssueChange(tx, issueCompany.companyId); await notifyIssueAttentionRetentionWork(tx, { companyId: issueCompany.companyId, issueId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); @@ -12618,7 +12628,8 @@ export function issueService(db: Db) { existing.checkoutRunId, tx, ); - if (!stale) { + if (!stale || await nativeManualAdmissionHoldsExecution(tx as unknown as Db, + { companyId: existing.companyId, runId: existing.checkoutRunId, issueId: id })) { throw conflict("Only checkout run can release issue", { issueId: existing.id, assigneeAgentId: existing.assigneeAgentId, diff --git a/server/src/services/native-runtime/native-cancellation-request.ts b/server/src/services/native-runtime/native-cancellation-request.ts index 6150f030c26..3b38abfa5c4 100644 --- a/server/src/services/native-runtime/native-cancellation-request.ts +++ b/server/src/services/native-runtime/native-cancellation-request.ts @@ -2,6 +2,7 @@ import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellatio import { and, eq, sql } from "drizzle-orm"; import { activityLog, budgetReservations, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { badRequest, conflict } from "../../errors.js"; +import { nativeManualAdmissionHoldMatches } from "./native-runner-ownership.js"; const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; const record = (value: unknown): Record => value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; @@ -78,8 +79,10 @@ export function assertCancellationRequest(result: unknown, requestId: string, al * is not retry authority; a resumed Stop needs its exact durable native intent. */ export function nativeRetryCancellationEligible(input: { runId: string; companyId: string; issueId: string; resultJson: unknown; requestId?: string; scope?: string; -}, coordinator: { phase?: string; failureCode?: string | null } | null | undefined) { +}, coordinator: Partial> | null | undefined) { if (coordinator?.phase === "retryable_failure") return true; + if (coordinator?.failureCode === "native_restart_recovery_blocked" && nativeManualAdmissionHoldMatches(input, coordinator)) return true; if (coordinator?.phase !== "terminal_failure" || coordinator.failureCode !== "native_retry_cancelled") return false; const result = record(input.resultJson), native = record(result.nativeCancellation); return (!input.requestId || (record(result.startupCancellation).cancellationRequestId === input.requestId diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 6a8b1d43ca2..493551935cc 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,5 +1,4 @@ import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; -import { notifyStatusCardIssueChange } from "../status-card-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { pendingNativeRunProjectionCondition } from "./native-run-projection.js"; import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js"; @@ -15,10 +14,9 @@ import { dismissAutomaticCompletionReviews } from "./automatic-completion-review import { logger } from "../../middleware/logger.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { createHash, randomUUID } from "node:crypto"; -import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lt, lte, ne, not, notInArray, or, sql, type SQL } from "drizzle-orm"; +import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lt, lte, not, notInArray, or, sql, type SQL } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { - agents, heartbeatRunEvents, heartbeatRuns, issues, @@ -57,11 +55,7 @@ import { type NativeAuthoritativeIssueStatus, } from "./status-arbiter.js"; -import { getNativeReviewAssignment } from "./native-review-participant.js"; -import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js"; -import { isAutomaticRecoverySuppressedByPauseHold } from "../recovery/pause-hold-guard.js"; -import { readRunCancellation } from "../run-cancellation.js"; -import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; +import { readNativeSessionResumptionAgent, restoreNativeSessionExecution } from "./native-session-resumption-admission.js"; import { retireSupersededNativeRetry } from "./native-superseded-retry.js"; import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js"; @@ -180,37 +174,13 @@ export async function claimNativeSessionResumptions(input: { ) return false; - if ((issue.executionRunId && issue.executionRunId !== run.id) || - (issue.checkoutRunId && issue.checkoutRunId !== run.id)) return false; - const [agent] = await tx.select().from(agents).where(and( - eq(agents.companyId, candidate.companyId), eq(agents.id, run.agentId), - )).limit(1); - if (!agent || !(await evaluateAgentInvokabilityFromDb(tx as unknown as Db, agent)).invokable) return false; - if (issue.assigneeAgentId !== run.agentId && !await getNativeReviewAssignment(tx as unknown as Db, { - companyId: run.companyId, issueId: issue.id, agentId: run.agentId, - contextSnapshot: run.contextSnapshot, - })) return false; - if (await isAutomaticRecoverySuppressedByPauseHold(tx as unknown as Db, run.companyId, issue.id)) return false; - if (!(await issueService(tx as unknown as Db).getDependencyReadiness(issue.id, tx)).isDependencyReady) return false; - const [competing] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, run.companyId), ne(heartbeatRuns.id, run.id), - inArray(heartbeatRuns.status, issue.executionRunId === run.id - ? ["running", "scheduled_retry"] : ["queued", "running", "scheduled_retry"]), - or(eq(heartbeatRuns.issueId, issue.id), eq(heartbeatRuns.nativeIssueId, issue.id), - sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`), - )).limit(1); - if (competing) return false; + const agent = await readNativeSessionResumptionAgent(tx as unknown as Db, { issue, run, coordinator }); + if (!agent) return false; const profile = row.run.runnerProfileJson ?? {}; const persistedInput = profile.nativeExecutionInput; const checkpoint = profile.sessionCheckpoint; const failureDetail = record(row.coordinator.failureDetail); - const cancellation = readRunCancellation(run.resultJson); - const nativeCancellation = record(record(run.resultJson).nativeCancellation); - if ((cancellation?.source === "operator" && cancellation.expected) || - (["run", "issue"].includes(String(nativeCancellation.scope)) && - ["pending", "acknowledged"].includes(String(nativeCancellation.dispatchState))) || - typeof failureDetail.successorRunId === "string") return false; const originalFailureCode = typeof failureDetail.originalFailureCode === "string" ? failureDetail.originalFailureCode @@ -345,19 +315,7 @@ export async function claimNativeSessionResumptions(input: { nativePhaseUpdatedAt: now, updatedAt: now, }).where(eq(heartbeatRuns.id, row.run.id)); - await notifyStatusCardIssueChange(tx, run.companyId); - await tx.update(issues).set({ - executionRunId: run.id, - executionAgentNameKey: agent.name.trim().toLowerCase() || null, - executionLockedAt: now, - updatedAt: now, - }).where(and(eq(issues.id, issue.id), eq(issues.companyId, run.companyId))); - if (issue.executionRunId !== run.id) await appendHeartbeatRunEvent(tx as unknown as Db, { - companyId: run.companyId, runId: run.id, agentId: run.agentId, - eventType: "native.recovery.execution_reclaimed", stream: "system", level: "info", - message: "Reclaimed the original task execution for native session recovery.", - payload: { issueId: issue.id }, - }); + await restoreNativeSessionExecution(tx as unknown as Db, { issue, run, agentName: agent.name, now }); return true; }).catch((error: unknown) => { claimFailed = true; diff --git a/server/src/services/native-runtime/native-remote-retry-admission.test.ts b/server/src/services/native-runtime/native-remote-retry-admission.test.ts new file mode 100644 index 00000000000..69227e908e5 --- /dev/null +++ b/server/src/services/native-runtime/native-remote-retry-admission.test.ts @@ -0,0 +1,785 @@ +import { randomUUID } from "node:crypto"; +import { resolve } from "node:path"; +import { and, eq, sql } from "drizzle-orm"; +import { agents, companies, completionContracts, databaseWorkPending, environmentLeases, reconcileDatabaseWork, heartbeatRunEvents, heartbeatRuns, issues, issueRelations, + issueTreeHolds, issueTreeHoldMembers, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints } from "@paperclipai/db"; +import { expect, it, vi } from "vitest"; +import type { NativeExecutionInputV1 } from "@paperclipai/paperclip-runner"; + +const provider = vi.hoisted(() => ({ execute: vi.fn() })); +vi.mock("../../vendor/paperclip-runner/index.js", async importOriginal => ({ + ...await importOriginal(), + executeNativeSession: provider.execute, +})); +import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js"; +import { describeEmbeddedPostgres, nativeRestartFixture, remotePreparedWorkspace } from "../../__tests__/helpers/native-restart.js"; +import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; +import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../delivery-work-notifications.js"; +import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; +import { cancelNativeSession, executePaperclipNativeSession } from "./native-session-executor.js"; +import { buildNativeCompletionContract } from "./completion-contracts.js"; +import { claimNativeRestartRecoveries, currentNativeControllerIdentity, NATIVE_RESTART_RECOVERY_FAILED } from "./native-restart-recovery.js"; +import { readSystemRecovery, SYSTEM_RECOVERY_MAX_RETRIES } from "../system-recovery-state.js"; +import { recoveryService } from "../recovery/service.js"; +import { issueService } from "../issues.js"; +import { hasPendingRunCleanup } from "../execution-blocker.js"; +import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js"; +import { nativeManualAdmissionHoldsExecution } from "./native-runner-ownership.js"; +import { remoteTerminationReceipt } from "../remote-execution-termination.js"; +import { releaseSettledNativeRunProcess } from "../native-local-process-stop.js"; +import { createHeartbeatQueue } from "../heartbeat/queue.js"; +import { createHeartbeatQueueTestDependencies } from "../../__tests__/helpers/heartbeat-queue.js"; +import { instanceSettingsService } from "../instance-settings.js"; +import { claimCancellationRequest } from "./native-cancellation-request.js"; + +describeEmbeddedPostgres("remote native retry execution admission", () => { + const environment = nativeRestartFixture(); + + async function failedRemoteRun(storage: "classic" | "remote" | "plugin") { + const { companyId, agentId, runtimeRoot } = environment.current; + const fixture = await environment.seedRun(`REMOTE-RETRY-${storage}-${randomUUID()}`); + await fixture.db.update(agents).set({ status: "active" }).where(eq(agents.id, agentId)); + const controller = { bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() }; + await fixture.db.update(nativeRunFinalizations).set({ + controllerBootId: controller.bootId, controllerGeneration: controller.generation, + leaseOwner: controller.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000), + }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + const authority = new DurablePrpControlPlane({ + stateDirectory: resolve(runtimeRoot, fixture.runId), + identity: { runId: fixture.runId, normalizedSessionId: fixture.native.normalizedSessionId, + runnerInstanceId: fixture.native.runnerInstanceId, environmentLeaseId: fixture.native.environmentLeaseId, + turnId: fixture.native.turnId, itemId: fixture.native.itemId }, + expectedRunnerVersion: "retry-admission-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}`, + }); + authority.setCheckpointPersistence(nativeRunnerCheckpointWriter(fixture.db, { + companyId, issueId: fixture.issueId, runId: fixture.runId, authorityRunId: fixture.runId, + ...controller, + })); + await authority.flushCheckpoint(); + await authority.stop(); + const classic = remotePreparedWorkspace(); + const remote = { schema: "paperclip.remote-workspace-run.v1", workspaceId: classic.workspaceId, + leaseId: classic.leaseId, providerLeaseId: classic.providerLeaseId, remoteCwd: classic.remoteCwd, + epoch: 1, state: "prepared", snapshotId: null }; + const checkpoint = { backendKind: "acpx", driverKind: "acpx_runtime", sessionId: "provider-session", + providerSessionId: "provider-session", activeTurnId: "retained-turn", semanticResult: null, + pendingRuntimeRequests: [], terminalTurns: [], identity: { companyId, agentId, + issueId: fixture.issueId, runId: fixture.runId, sessionId: fixture.native.normalizedSessionId } }; + const [pluginLease] = storage === "plugin" ? await fixture.db.insert(environmentLeases).values({ companyId, + issueId: fixture.issueId, heartbeatRunId: fixture.runId, provider: "daytona", providerLeaseId: randomUUID(), + status: "active", leasePolicy: "ephemeral", metadata: { runnerTaskLease: true }, + }).returning({ id: environmentLeases.id }) : []; + await fixture.db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), + processPid: 1386, processStartedAt: new Date("2026-11-01T12:00:00Z"), + nativePhase: "retryable_failure", errorCode: "native_session_interrupted", + runnerProfileJson: { ...fixture.preparedProfile, nativeExecutionInput: { binding: { + companyId, agentId, issueId: fixture.issueId, runId: fixture.runId } }, sessionCheckpoint: checkpoint, + ...(storage === "plugin" ? { nativeWorkspaceSync: null, remoteWorkspace: null } + : storage === "classic" ? { nativeWorkspaceSync: classic } : { remoteWorkspace: remote }) }, + }).where(eq(heartbeatRuns.id, fixture.runId)); + await fixture.db.update(nativeRunFinalizations).set({ phase: "retryable_failure", attempt: 1, + leaseOwner: null, leaseExpiresAt: null, recoveryState: "resuming_session", + failureCode: "native_session_interrupted", nextAttemptAt: new Date(0), + failureDetail: { recoveryMode: "exact_checkpoint_resume", providerSessionEstablished: true, + checkpointExists: true, providerEventsExist: true }, + }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + await fixture.db.update(issues).set({ executionRunId: null, checkoutRunId: null }) + .where(eq(issues.id, fixture.issueId)); + return { ...fixture, companyId, agentId, checkpoint, pluginLeaseId: pluginLease?.id }; + } + + async function manuallyHeldRemoteRun() { + const f = await failedRemoteRun("remote"); + await f.db.update(issues).set({ executionRunId: f.runId, checkoutRunId: f.runId }).where(eq(issues.id, f.issueId)); + await f.db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, + status: "succeeded", createdAt: new Date(Date.now() + 1_000) }); + await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId] }); + return f; + } + + it.each(["background sweep", "terminal cleanup", "pending cleanup", "queue release", "agent release"] as const) + ("preserves a manual native execution hold through %s", async entry => { + const f = await manuallyHeldRemoteRun(); + const [before] = await f.db.select().from(issues).where(eq(issues.id, f.issueId)); + if (entry === "background sweep") expect(await recoveryService(f.db, { enqueueWakeup: vi.fn() }) + .sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId })).toMatchObject({ cleared: 0 }); + if (entry === "terminal cleanup") { + await expect(issueService(f.db).assertCheckoutOwner(f.issueId, f.agentId, randomUUID())) + .rejects.toMatchObject({ status: 409 }); + } + if (entry === "agent release") await expect(issueService(f.db).release(f.issueId, f.agentId, randomUUID())) + .rejects.toMatchObject({ status: 409 }); + if (entry === "pending cleanup") expect(await hasPendingRunCleanup(f.db, f.companyId, f.runId)).toBe(true); + if (entry === "queue release") await createPostgresWakeQueueAdapter(f.db, { + resolveResponsibleUserId: async () => "operator", getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }), + resolveSessionBeforeForWakeup: async () => null, + }).withIssueExecutionLock({ companyId: f.companyId, runId: f.runId, now: new Date() }, + async () => ({ outcome: { kind: "released" }, postCommitEffects: [] })); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(before); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } }); + }); + + it.each(["execution", "checkout"] as const)("preserves a manual hold referenced only by the %s binding", async pointer => { + const f = await manuallyHeldRemoteRun(); + await f.db.update(issues).set(pointer === "execution" ? { checkoutRunId: null } : { executionRunId: null }) + .where(eq(issues.id, f.issueId)); + expect(await recoveryService(f.db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId })) + .toMatchObject({ cleared: 0 }); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]) + .toMatchObject(pointer === "execution" ? { executionRunId: f.runId, checkoutRunId: null } : { executionRunId: null, checkoutRunId: f.runId }); + }); + + it("rechecks a newly committed manual hold under the issue lock before background cleanup", async () => { + const f = await manuallyHeldRemoteRun(); + const [hold] = await f.db.select({ failureCode: nativeRunFinalizations.failureCode, + failureDetail: nativeRunFinalizations.failureDetail }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, f.runId)); + await f.db.update(nativeRunFinalizations).set({ failureCode: null, failureDetail: null }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + const beforeStaleLockClear = vi.fn(async (issueId: string) => { + if (issueId === f.issueId) await f.db.update(nativeRunFinalizations).set(hold) + .where(eq(nativeRunFinalizations.runId, f.runId)); + }); + expect(await recoveryService(f.db, { enqueueWakeup: vi.fn(), beforeStaleLockClear }) + .sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId })).toMatchObject({ cleared: 0 }); + expect(beforeStaleLockClear).toHaveBeenCalledExactlyOnceWith(f.issueId); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]) + .toMatchObject({ executionRunId: f.runId, checkoutRunId: f.runId }); + }); + + it("defers a new wake behind the manual owner without clearing it or dispatching", async () => { + const f = await manuallyHeldRemoteRun(); + await f.db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + const [agent] = await f.db.select().from(agents).where(eq(agents.id, f.agentId)); + const dependencies = createHeartbeatQueueTestDependencies(f.db); + dependencies.getAgent.mockResolvedValue(agent); + dependencies.instanceSettings.getExperimental.mockImplementation(() => instanceSettingsService(f.db).getExperimental()); + dependencies.wakeQueue.admitWakeBehindIssueExecution.mockResolvedValue({ kind: "deferred" }); + await createHeartbeatQueue(f.db, dependencies).enqueueWakeup(f.agentId, { source: "assignment", reason: "issue_assigned", + requestedByActorType: "system", payload: { issueId: f.issueId }, contextSnapshot: { issueId: f.issueId } }); + expect(dependencies.executeRun).not.toHaveBeenCalled(); + expect(dependencies.wakeQueue.admitWakeBehindIssueExecution).toHaveBeenCalledOnce(); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]) + .toMatchObject({ executionRunId: f.runId, checkoutRunId: f.runId }); + }); + + it("retains the hold until every exact remote lease is confirmed stopped, then permits ordinary checkout", async () => { + const f = await manuallyHeldRemoteRun(); + const scope = { companyId: f.companyId, runId: f.runId, issueId: f.issueId }; + expect(await nativeManualAdmissionHoldsExecution(f.db, scope)).toBe(true); + expect(await nativeManualAdmissionHoldsExecution(f.db, { ...scope, companyId: randomUUID() })).toBe(false); + expect(await nativeManualAdmissionHoldsExecution(f.db, { ...scope, runId: randomUUID() })).toBe(false); + expect(await nativeManualAdmissionHoldsExecution(f.db, { ...scope, issueId: randomUUID() })).toBe(false); + const leases = await f.db.insert(environmentLeases).values([1, 2].map(index => ({ companyId: f.companyId, + issueId: f.issueId, heartbeatRunId: f.runId, provider: "daytona", providerLeaseId: `fixture-${index}-${randomUUID()}`, + status: "released", releasedAt: new Date(), cleanupStatus: "success", metadata: {}, + }))).returning(); + expect(await nativeManualAdmissionHoldsExecution(f.db, scope)).toBe(true); + for (let index = 0; index < leases.length; index += 1) { + const lease = leases[index]; + await f.db.update(environmentLeases).set({ metadata: { remoteExecutionTermination: remoteTerminationReceipt(lease, + { providerLeaseId: lease.providerLeaseId, state: "stopped" }) } }).where(eq(environmentLeases.id, lease.id)); + expect(await nativeManualAdmissionHoldsExecution(f.db, scope)).toBe(index < leases.length - 1); + } + expect(await releaseSettledNativeRunProcess(f.db, f.runId)).toBe("released"); + expect(await hasPendingRunCleanup(f.db, f.companyId, f.runId)).toBe(false); + expect(await recoveryService(f.db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId })) + .toMatchObject({ cleared: 1 }); + const actorRunId = randomUUID(); + await f.db.insert(heartbeatRuns).values({ id: actorRunId, companyId: f.companyId, agentId: f.agentId, + status: "running", invocationSource: "manual", contextSnapshot: { issueId: f.issueId } }); + expect(await issueService(f.db).assertCheckoutOwner(f.issueId, f.agentId, actorRunId)) + .toMatchObject({ executionRunId: actorRunId, checkoutRunId: actorRunId }); + }); + + it.each(["ordinary terminal", "accepted result"] as const)("does not preserve unrelated %s ownership", async state => { + const f = await manuallyHeldRemoteRun(); + if (state === "ordinary terminal") await f.db.update(nativeRunFinalizations).set({ failureDetail: null }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + else { + const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [result] = await f.db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, + runId: f.runId, completionContractId: run.completionContractId!, serverFingerprint: randomUUID(), + canonicalSha256: "a".repeat(64), schemaStatus: "accepted", resultJson: {} }).returning(); + await f.db.update(nativeRunFinalizations).set({ resultId: result.id }).where(eq(nativeRunFinalizations.runId, f.runId)); + } + expect(await nativeManualAdmissionHoldsExecution(f.db, { companyId: f.companyId, runId: f.runId, issueId: f.issueId })).toBe(false); + expect(await recoveryService(f.db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId })) + .toMatchObject({ cleared: 1 }); + }); + + it("admits an explicit correlated Stop for the preserved manual execution", async () => { + const f = await manuallyHeldRemoteRun(); + const requestId = randomUUID(); + expect(await claimCancellationRequest(f.db, f.runId, f.companyId, requestId, "operator")) + .toMatchObject({ resultJson: { startupCancellation: { cancellationRequestId: requestId, retryCancellation: true } } }); + const stop = await cancelNativeSession(f.runId, "Operator Stop", { db: f.db, scope: "run", cancellationRequestId: requestId }); + expect(stop).toMatchObject({ dispatched: false, auditId: expect.any(String) }); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "terminal_failure", failureCode: "native_retry_cancelled" }); + expect(await nativeManualAdmissionHoldsExecution(f.db, { companyId: f.companyId, runId: f.runId, issueId: f.issueId })).toBe(true); + expect(await recoveryService(f.db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks({ companyId: f.companyId, issueId: f.issueId })) + .toMatchObject({ cleared: 0 }); + expect(await cancelNativeSession(f.runId, "Repeat Operator Stop", { db: f.db, scope: "run", cancellationRequestId: requestId })) + .toMatchObject({ auditId: stop.auditId }); + await f.db.update(nativeRunFinalizations).set({ failureCode: "native_restart_recovery_blocked" }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await cancelNativeSession(f.runId, "Resume recorded Stop", { db: f.db, scope: "run", cancellationRequestId: requestId })) + .toMatchObject({ auditId: stop.auditId }); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ failureCode: "native_retry_cancelled" }); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]) + .toMatchObject({ executionRunId: f.runId, checkoutRunId: f.runId }); + }); + + it.each(["classic", "remote", "plugin"] as const)("reclaims the exact failed %s execution before remote inspection", async storage => { + const fixture = await failedRemoteRun(storage); + expect(await claimNativeSessionResumptions({ db: fixture.db, runnerInstanceId: "ordinary-resumer", + runIds: [fixture.runId] })).toEqual([]); + const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, + controller: environment.current.successor, restartKind: "hard", runIds: [fixture.runId] }); + expect(claim).toMatchObject({ kind: "reattach_remote_runner", runId: fixture.runId, providerAttempt: 2 }); + expect((await fixture.db.select().from(issues).where(eq(issues.id, fixture.issueId)))[0]) + .toMatchObject({ executionRunId: fixture.runId, status: "in_progress", assigneeAgentId: fixture.agentId }); + expect((await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)))[0]) + .toMatchObject({ status: "running", processPid: 1386, + runnerProfileJson: { sessionCheckpoint: fixture.checkpoint } }); + expect((await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)))[0]) + .toMatchObject({ attempt: 2, phase: "observed", recoveryState: "awaiting_runner_reattach" }); + }); + + it.each(["company", "run", "issue"] as const)("does not use a plugin lease from another %s", async scope => { + const f = await failedRemoteRun("plugin"); + if (!f.pluginLeaseId) throw new Error("Expected a plugin allocation"); + const foreignId = randomUUID(); + if (scope === "company") { + await f.db.insert(companies).values({ id: foreignId, name: "Other company", issuePrefix: foreignId.slice(0, 8) }); + await f.db.update(environmentLeases).set({ companyId: foreignId }).where(eq(environmentLeases.id, f.pluginLeaseId)); + } else if (scope === "run") { + await f.db.insert(heartbeatRuns).values({ id: foreignId, companyId: f.companyId, agentId: f.agentId, status: "failed" }); + await f.db.update(environmentLeases).set({ heartbeatRunId: foreignId }).where(eq(environmentLeases.id, f.pluginLeaseId)); + } else { + await f.db.insert(issues).values({ id: foreignId, companyId: f.companyId, title: "Other task" }); + await f.db.update(environmentLeases).set({ issueId: foreignId }).where(eq(environmentLeases.id, f.pluginLeaseId)); + } + const onClaim = vi.fn(); + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], onClaim })) + .toMatchObject([{ kind: "blocked", runId: f.runId, reason: "issue_execution_lock_changed" }]); + expect(onClaim).not.toHaveBeenCalled(); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]) + .toMatchObject({ executionRunId: null, checkoutRunId: null }); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "retryable_failure", attempt: 1 }); + }); + + it.each((["remote", "plugin"] as const).flatMap(storage => (["done", "cancelled", "blocked", "paused-agent", "pending-agent-approval", "reassigned", + "dependency", "tree-pause", "native-stop", "operator-stop", "execution-owner", "checkout-owner", + "queued-input", "newer-completed-input", "wrong-checkpoint", "missing-authority"] as const).map(gate => [storage, gate] as const))) + ("preserves a %s retry held by %s", async (storage, gate) => { + const f = await failedRemoteRun(storage); + const { db } = f; + if (["done", "cancelled", "blocked"].includes(gate)) { + await db.update(issues).set({ status: gate }).where(eq(issues.id, f.issueId)); + } else if (gate === "paused-agent" || gate === "pending-agent-approval") { + await db.update(agents).set({ status: gate === "paused-agent" ? "paused" : "pending_approval" }).where(eq(agents.id, f.agentId)); + } else if (gate === "reassigned") { + const nextAgent = randomUUID(); + await db.insert(agents).values({ id: nextAgent, companyId: f.companyId, name: "Next owner", status: "active" }); + await db.update(issues).set({ assigneeAgentId: nextAgent }).where(eq(issues.id, f.issueId)); + } else if (gate === "dependency") { + const dependency = randomUUID(); + await db.insert(issues).values({ id: dependency, companyId: f.companyId, title: "Required input", status: "todo" }); + await db.insert(issueRelations).values({ companyId: f.companyId, issueId: dependency, relatedIssueId: f.issueId, type: "blocks" }); + } else if (gate === "tree-pause") { + const holdId = randomUUID(); + await db.insert(issueTreeHolds).values({ id: holdId, companyId: f.companyId, rootIssueId: f.issueId, mode: "pause" }); + await db.insert(issueTreeHoldMembers).values({ holdId, companyId: f.companyId, issueId: f.issueId, + issueTitle: "Paused work", issueStatus: "in_progress" }); + } else if (gate === "native-stop" || gate === "operator-stop") { + await db.update(heartbeatRuns).set({ resultJson: gate === "native-stop" + ? { nativeCancellation: { scope: "run", dispatchState: "pending" } } + : { cancellation: { source: "operator", expected: true, initiator: { type: "user" }, + reason: "Stop", recordedAt: new Date().toISOString() } } }).where(eq(heartbeatRuns.id, f.runId)); + } else if (["execution-owner", "checkout-owner", "queued-input", "newer-completed-input"].includes(gate)) { + const nextRun = randomUUID(); + await db.insert(heartbeatRuns).values({ id: nextRun, companyId: f.companyId, agentId: f.agentId, + issueId: f.issueId, contextSnapshot: { issueId: f.issueId, wakeReason: "issue_comment" }, + invocationSource: "manual", createdAt: new Date(Date.now() + 1_000), + status: gate === "queued-input" ? "queued" : gate === "newer-completed-input" ? "succeeded" : "failed" }); + if (gate === "execution-owner" || gate === "checkout-owner") await db.update(issues) + .set(gate === "execution-owner" ? { executionRunId: nextRun } : { checkoutRunId: nextRun }).where(eq(issues.id, f.issueId)); + } else if (gate === "wrong-checkpoint") { + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, + sessionCheckpoint: { ...f.checkpoint, identity: { ...f.checkpoint.identity, runId: randomUUID() } }, + } }).where(eq(heartbeatRuns.id, f.runId)); + } else if (gate === "missing-authority") { + await db.delete(nativeRunnerCheckpoints).where(and(eq(nativeRunnerCheckpoints.companyId, f.companyId), + eq(nativeRunnerCheckpoints.runId, f.runId))); + } + const beforeTask = (await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]; + const beforeRun = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]; + const beforeOwner = (await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]; + const dispositions = await claimNativeRestartRecoveries({ db, + controller: environment.current.successor, restartKind: "hard", runIds: [f.runId] }); + expect(dispositions.every(value => value.kind === "blocked")).toBe(true); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(beforeTask); + const [afterRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [afterOwner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + if (["done", "cancelled", "newer-completed-input", "execution-owner", "checkout-owner", "queued-input"].includes(gate)) { + expect(afterOwner).toMatchObject({ phase: "terminal_failure", attempt: 1, nextAttemptAt: null, + failureDetail: { systemRecovery: { state: "exhausted" } } }); + expect(afterRun).toMatchObject({ status: "failed", processPid: beforeRun.processPid, + runnerProfileJson: beforeRun.runnerProfileJson }); + return; + } + expect(afterRun).toEqual(beforeRun); + expect(afterOwner.failureDetail).toMatchObject(beforeOwner.failureDetail!); + expect({ ...afterOwner, nextAttemptAt: beforeOwner.nextAttemptAt, updatedAt: beforeOwner.updatedAt, + recoveryHistory: beforeOwner.recoveryHistory, failureDetail: beforeOwner.failureDetail }).toEqual(beforeOwner); + if (gate === "execution-owner") expect(afterOwner).toEqual(beforeOwner); + else expect(afterOwner.nextAttemptAt!.getTime()).toBeGreaterThan(beforeOwner.nextAttemptAt!.getTime()); + }); + + it.each(["remote", "plugin"] as const)("waits for the exact old %s controller lease and keeps future retries quiet", async storage => { + const f = await failedRemoteRun(storage); + const now = new Date(), deadline = new Date(now.getTime() + 60_000); + await f.db.update(nativeRunFinalizations).set({ nextAttemptAt: deadline }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", now, runIds: [f.runId] })).toEqual([]); + await f.db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(0), leaseOwner: "old-owner", leaseExpiresAt: deadline }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", now, runIds: [f.runId] })).toMatchObject([{ kind: "awaiting_evidence", reason: "remote_controller_lease_live" }]); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBeNull(); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe("failed"); + }); + + it.each(["remote", "plugin"] as const)("grants two competing controllers only one original %s execution", async storage => { + const f = await failedRemoteRun(storage); + const claims = (await Promise.all([environment.current.successor, + { ...environment.current.successor, bootId: randomUUID() }].map(controller => + claimNativeRestartRecoveries({ db: f.db, controller, restartKind: "hard", runIds: [f.runId] })))).flat(); + expect(claims.filter(claim => claim.kind === "reattach_remote_runner")).toHaveLength(1); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ controllerGeneration: 2, attempt: 2 }); + expect(await f.db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.runId), + eq(heartbeatRunEvents.eventType, "native.recovery.execution_reclaimed")))).toHaveLength(1); + }); + + it.each(["remote", "plugin"] as const)("rolls back the %s task, owner, run, and recovery event together", async storage => { + const f = await failedRemoteRun(storage); + await expect(f.db.transaction(async tx => { + const claims = await claimNativeRestartRecoveries({ db: tx as unknown as typeof f.db, + controller: environment.current.successor, restartKind: "hard", runIds: [f.runId] }); + expect(claims).toMatchObject([{ kind: "reattach_remote_runner" }]); + throw new Error("rollback admission"); + })).rejects.toThrow("rollback admission"); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBeNull(); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe("failed"); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "retryable_failure", controllerGeneration: 1, leaseOwner: null, attempt: 1 }); + expect(await f.db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.runId), + eq(heartbeatRunEvents.eventType, "native.recovery.execution_reclaimed")))).toEqual([]); + }); + + it.each([ + ["succeeded", "2026-11-01T12:00:00.000600Z", 1, true], + ["succeeded", "2026-11-01T12:00:00.000000Z", 1, true], + ["failed", "2026-11-01T12:00:00.000000Z", 1, true], + ["failed", "2026-11-01T12:00:00.000600Z", 1, true], + ["succeeded", "2026-11-01T12:00:00.000599Z", 1, false], + ["failed", "2026-11-01T12:00:00.000601Z", -1, true], + ["failed", "2026-11-01T12:00:00.000600Z", -1, false], + ] as const)("uses canonical source order for %s input at %s with ID offset %s", async (status, timestamp, offset, superseded) => { + const f = await failedRemoteRun("remote"); + const idHex = (BigInt(`0x${f.runId.replaceAll("-", "")}`) + BigInt(offset)).toString(16).padStart(32, "0"); + const nextId = `${idHex.slice(0, 8)}-${idHex.slice(8, 12)}-${idHex.slice(12, 16)}-${idHex.slice(16, 20)}-${idHex.slice(20)}`; + const sourceTimestamp = timestamp.endsWith(".000000Z") ? timestamp : "2026-11-01T12:00:00.000600Z"; + await f.db.update(heartbeatRuns).set({ createdAt: sql`${sourceTimestamp}::timestamptz` }) + .where(eq(heartbeatRuns.id, f.runId)); + await f.db.insert(heartbeatRuns).values({ id: nextId, companyId: f.companyId, agentId: f.agentId, + issueId: f.issueId, invocationSource: "manual", status, createdAt: sql`${timestamp}::timestamptz` }); + const [disposition] = await claimNativeRestartRecoveries({ db: f.db, + controller: environment.current.successor, restartKind: "hard", runIds: [f.runId] }); + expect(disposition).toMatchObject(superseded + ? { kind: "blocked", reason: "native_retry_source_superseded" } + : { kind: "reattach_remote_runner", providerAttempt: 2 }); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId) + .toBe(superseded ? null : f.runId); + }); + + it.each((["remote", "plugin"] as const).flatMap(storage => (["released", "retained"] as const).map(lock => [storage, lock] as const))) + ("exhausts the existing attempt budget after repeated %s snapshot failures with a %s task lock", async (storage, lock) => { + const f = await failedRemoteRun(storage); + if (lock === "retained") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + const [issue] = await f.db.select().from(issues).where(eq(issues.id, f.issueId)); + const [contract] = await f.db.select().from(completionContracts).where(eq(completionContracts.issueId, f.issueId)); + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { companyId: f.companyId, runId: f.runId, issueId: f.issueId, agentId: f.agentId, executionWorkspaceId: f.runId }, + provider: { kind: "codex", model: "gpt-5.6-luna" }, + task: { identifier: issue.identifier!, title: issue.title, description: null, prompt: "Continue retained work.", workMode: "standard" }, + workspace: { cwd: environment.current.runtimeRoot, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: f.native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", + contract: buildNativeCompletionContract(issue, { revision: contract.revision }) }, + interactionResponses: [], credentialBindings: [], + }; + await f.db.update(heartbeatRuns).set({ runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) + || ${JSON.stringify({ nativeExecutionInput: execution })}::jsonb` }).where(eq(heartbeatRuns.id, f.runId)); + provider.execute.mockReset().mockRejectedValue(new Error("PRP command session.snapshot failed: original provider unavailable")); + const controller = await currentNativeControllerIdentity(); + for (const attempt of [2, 3]) { + await f.db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId)); + const [claim] = await claimNativeRestartRecoveries({ db: f.db, controller, restartKind: "hard", runIds: [f.runId] }); + expect(claim).toMatchObject({ kind: "reattach_remote_runner", providerAttempt: attempt }); + if (claim.kind !== "reattach_remote_runner") throw new Error("Expected exact remote recovery claim"); + await expect(executePaperclipNativeSession({ db: f.db, execution, runnerInstanceId: f.native.runnerInstanceId, + leaseOwner: claim.leaseOwner, restartRecovery: claim, + backend: { descriptor: async () => { throw new Error("Unexpected descriptor call"); }, + openSession: async () => { throw new Error("Unexpected session launch"); } }, + })).rejects.toThrow("PRP command session.snapshot failed"); + const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(owner).toMatchObject({ attempt, phase: attempt === 3 ? "terminal_failure" : "retryable_failure", + failureCode: attempt === 3 ? "native_session_retry_exhausted" : "native_session_interrupted", + recoveryState: attempt === 3 ? "blocked" : "resuming_session", leaseOwner: null }); + expect(owner.nextAttemptAt === null).toBe(attempt === 3); + if (lock === "released") await f.db.update(issues).set({ executionRunId: null }).where(eq(issues.id, f.issueId)); + } + expect(provider.execute).toHaveBeenCalledTimes(2); + expect(await claimNativeRestartRecoveries({ db: f.db, controller, restartKind: "hard", runIds: [f.runId] })).toEqual([]); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "blocked", executionRunId: null }); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ status: "failed", nativePhase: "terminal_failure", errorCode: "native_session_interrupted", + processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } }); + }); + + it.each((["remote", "plugin"] as const).flatMap(storage => (["released", "retained"] as const).map(lock => [storage, lock] as const))) + ("does not admit another failed %s retry when the persisted provider budget is already exhausted with a %s task lock", async (storage, lock) => { + const f = await failedRemoteRun(storage); + if (lock === "retained") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + await f.db.update(nativeRunFinalizations).set({ attempt: 3 }).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId] })).toMatchObject([{ kind: "blocked", reason: "execution_recovery_budget_exhausted" }]); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBeNull(); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "terminal_failure", attempt: 3, leaseOwner: null, nextAttemptAt: null, + failureCode: "native_restart_recovery_blocked", failureDetail: { reason: "execution_recovery_budget_exhausted" } }); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ status: "failed", nativePhase: "terminal_failure", processPid: 1386, + runnerProfileJson: { sessionCheckpoint: f.checkpoint } }); + expect(await f.db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId), + eq(issueRecoveryActions.kind, "active_run_watchdog")))) + .toMatchObject([{ ownerType: "board", evidence: { runId: f.runId, reason: "execution_recovery_budget_exhausted" } }]); + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId] })).toEqual([]); + }); + + it("persists a blocked retry deadline and records only changed hold reasons", async () => { + const f = await failedRemoteRun("remote"); + await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + const now = new Date(); + const attempt = (at: Date) => claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], now: at }); + const [first] = await attempt(now); + expect(first).toMatchObject({ kind: "blocked", reason: "native_retry_admission_ineligible", transitionUnchanged: false }); + if (!("retryAt" in first) || !first.retryAt) throw new Error("missing first retry deadline"); + expect(first.retryAt.getTime() - now.getTime()).toBeGreaterThanOrEqual(24_000); + expect(first.retryAt.getTime() - now.getTime()).toBeLessThanOrEqual(36_000); + expect(await attempt(new Date(first.retryAt.getTime() - 1))).toEqual([]); + const [second] = await attempt(first.retryAt); + expect(second).toMatchObject({ kind: "blocked", transitionUnchanged: true }); + if (!("retryAt" in second) || !second.retryAt) throw new Error("missing second retry deadline"); + expect(second.retryAt.getTime() - first.retryAt.getTime()).toBeGreaterThanOrEqual(48_000); + let [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(owner).toMatchObject({ phase: "retryable_failure", attempt: 1, + nextAttemptAt: second.retryAt, recoveryHistory: [{ reason: "native_retry_admission_ineligible" }] }); + await f.db.update(agents).set({ status: "active" }).where(eq(agents.id, f.agentId)); + await f.db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, + status: "succeeded", createdAt: new Date(now.getTime() + 1_000) }); + expect(await attempt(second.retryAt)).toMatchObject([{ kind: "blocked", reason: "native_retry_source_superseded" }]); + [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(owner.recoveryHistory).toMatchObject([{ reason: "native_retry_admission_ineligible" }, { reason: "native_retry_source_superseded" }]); + }); + + it("registers recovery work intent before writing a blocked retry deadline", async () => { + const f = await failedRemoteRun("remote"); + await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + const transaction = f.db.transaction.bind(f.db); + let writes = 0; + const interception = vi.spyOn(f.db, "transaction").mockImplementation((async (callback, options) => transaction(async tx => { + const update = tx.update.bind(tx); + const updates = vi.spyOn(tx, "update").mockImplementation((table => { + if (Object.is(table, nativeRunFinalizations)) { + writes += 1; + expect(databaseWorkPending(f.db, DELIVERY_QUEUES.nativeProcessSessions)).toBe(true); + } + return update(table); + }) as typeof tx.update); + try { return await callback(tx); } finally { updates.mockRestore(); } + }, options)) as typeof f.db.transaction); + try { + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId] })).toMatchObject([{ kind: "blocked" }]); + expect(writes).toBe(1); + } finally { interception.mockRestore(); } + }); + + it.each(["done", "cancelled", "superseded"] as const)("settles a permanently %s remote retry without recurring work", async gate => { + for (const lock of ["released", "retained"]) { + const f = await failedRemoteRun("remote"); + if (lock === "retained") await f.db.update(issues).set({ executionRunId: f.runId, checkoutRunId: f.runId }) + .where(eq(issues.id, f.issueId)); + if (gate === "superseded") await f.db.insert(heartbeatRuns).values({ companyId: f.companyId, + agentId: f.agentId, issueId: f.issueId, status: "succeeded", createdAt: new Date(Date.now() + 1_000) }); + else await f.db.update(issues).set({ status: gate }).where(eq(issues.id, f.issueId)); + const onClaim = vi.fn(); + const environmentWake = vi.fn(), stopWatchingEnvironment = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeEnvironmentSettlement, environmentWake); + const [disposition] = await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], onClaim }); + expect(disposition).toMatchObject({ kind: "blocked" }); + expect(disposition).not.toHaveProperty("retryAt"); + expect(onClaim).not.toHaveBeenCalled(); + expect(environmentWake).not.toHaveBeenCalled(); + stopWatchingEnvironment(); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ + executionRunId: lock === "retained" ? f.runId : null, checkoutRunId: lock === "retained" ? f.runId : null }); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "terminal_failure", nextAttemptAt: null, attempt: 1 }); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } }); + const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeProcessSessions, wake); + const transaction = vi.spyOn(f.db, "transaction"); + try { + for (const elapsed of [30_000, 60_000, 3_600_000]) expect(await claimNativeRestartRecoveries({ db: f.db, + controller: environment.current.successor, restartKind: "hard", runIds: [f.runId], + now: new Date(Date.now() + elapsed), onClaim })).toEqual([]); + expect(transaction).not.toHaveBeenCalled(); + expect(wake).not.toHaveBeenCalled(); + } finally { transaction.mockRestore(); unsubscribe(); } + } + }); + + it.each(["paused", "Stop", "missing-authority"] as const)("bounds %s holds and preserves the provider and own task lock for manual reconciliation", async gate => { + const f = await failedRemoteRun("remote"); + if (gate === "paused") await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + if (gate === "Stop") await f.db.update(heartbeatRuns).set({ resultJson: { + nativeCancellation: { scope: "run", dispatchState: "pending" }, + } }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "missing-authority") await f.db.delete(nativeRunnerCheckpoints) + .where(and(eq(nativeRunnerCheckpoints.companyId, f.companyId), eq(nativeRunnerCheckpoints.runId, f.runId))); + await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + await f.db.update(nativeRunFinalizations).set({ attempt: 3 }).where(eq(nativeRunFinalizations.runId, f.runId)); + let now = new Date(); + const onClaim = vi.fn(); + const environmentWake = vi.fn(), stopWatchingEnvironment = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeEnvironmentSettlement, environmentWake); + for (let index = 0; index <= SYSTEM_RECOVERY_MAX_RETRIES; index += 1) { + const [disposition] = await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], now, onClaim }); + const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const policy = readSystemRecovery(owner.failureDetail); + expect(policy).not.toBeNull(); + expect(owner.attempt).toBe(3); + if (index < SYSTEM_RECOVERY_MAX_RETRIES) { + expect(owner.phase).toBe("retryable_failure"); + expect(policy).toMatchObject({ state: "awaiting_evidence" }); + expect(disposition).toHaveProperty("retryAt", owner.nextAttemptAt); + now = owner.nextAttemptAt!; + } else { + expect(owner).toMatchObject({ phase: "terminal_failure", nextAttemptAt: null }); + expect(policy).toMatchObject({ state: "exhausted" }); + expect(disposition).not.toHaveProperty("retryAt"); + } + } + expect(onClaim).not.toHaveBeenCalled(); + expect(environmentWake).not.toHaveBeenCalled(); + stopWatchingEnvironment(); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ executionRunId: f.runId }); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } }); + expect(await f.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId))) + .toMatchObject([{ ownerType: "board", wakePolicy: null, + evidence: { automaticAdmissionExhausted: true, executorStopEvidence: "unverified", providerPid: 1386, checkpointPreserved: true }, + nextAction: expect.stringContaining("original executor stop and state remain unverified") }]); + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], now: new Date(now.getTime() + 3_600_000) })).toEqual([]); + }); + + it("settles the old admission policy while preserving a newer active execution and its binding", async () => { + const f = await failedRemoteRun("remote"); + const newerRunId = randomUUID(); + await f.db.insert(heartbeatRuns).values({ id: newerRunId, companyId: f.companyId, agentId: f.agentId, + issueId: f.issueId, status: "running", invocationSource: "manual", createdAt: new Date(Date.now() + 1_000) }); + await f.db.update(issues).set({ executionRunId: newerRunId, checkoutRunId: newerRunId }).where(eq(issues.id, f.issueId)); + const [task] = await f.db.select().from(issues).where(eq(issues.id, f.issueId)); + const [newer] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, newerRunId)); + const environmentWake = vi.fn(), unsubscribe = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeEnvironmentSettlement, environmentWake); + const onClaim = vi.fn(); + try { + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], onClaim })) + .toMatchObject([{ kind: "blocked", reason: "native_retry_source_superseded" }]); + expect(onClaim).not.toHaveBeenCalled(); + expect(environmentWake).not.toHaveBeenCalled(); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(task); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, newerRunId)))[0]).toEqual(newer); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "terminal_failure", nextAttemptAt: null, failureDetail: { systemRecovery: { state: "exhausted" } } }); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } }); + const transactions = vi.spyOn(f.db, "transaction"); + try { + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], now: new Date(Date.now() + 3_600_000) })).toEqual([]); + expect(transactions).not.toHaveBeenCalled(); + } finally { transactions.mockRestore(); } + } finally { unsubscribe(); } + }); + + it("resumes the original provider when eligibility returns before admission backoff exhausts", async () => { + const f = await failedRemoteRun("remote"); + await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId] }); + const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + await f.db.update(agents).set({ status: "active" }).where(eq(agents.id, f.agentId)); + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], now: owner.nextAttemptAt! })) + .toMatchObject([{ kind: "reattach_remote_runner", providerAttempt: 2, runId: f.runId }]); + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ status: "running", processPid: 1386, runnerProfileJson: { sessionCheckpoint: f.checkpoint } }); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBe(f.runId); + }); + + it("does not settle permanently ineligible work while the original controller is active", async () => { + const f = await failedRemoteRun("remote"); + await f.db.update(issues).set({ status: "done" }).where(eq(issues.id, f.issueId)); + const [before] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeProcessSessions, wake); + try { + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], liveRunExecutions: new Set([f.runId]) })) + .toMatchObject([{ kind: "awaiting_evidence", reason: "current_controller_preparing" }]); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(before); + expect(wake).not.toHaveBeenCalled(); + expect(await f.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId))).toEqual([]); + } finally { unsubscribe(); } + }); + + it("keeps a retry deadline that changed after discovery read-only and quiet", async () => { + const f = await failedRemoteRun("remote"); + await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + const deadline = new Date(Date.now() + 60_000); + const transaction = f.db.transaction.bind(f.db); + const interception = vi.spyOn(f.db, "transaction").mockImplementationOnce((async (callback, options) => { + await f.db.update(nativeRunFinalizations).set({ nextAttemptAt: deadline }).where(eq(nativeRunFinalizations.runId, f.runId)); + return transaction(callback, options); + }) as typeof f.db.transaction); + const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeProcessSessions, wake); + try { + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId] })).toMatchObject([{ kind: "blocked", retryAt: deadline, transitionUnchanged: true }]); + const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(owner).toMatchObject({ nextAttemptAt: deadline, recoveryHistory: [], phase: "retryable_failure", attempt: 1 }); + expect(wake).not.toHaveBeenCalled(); + } finally { interception.mockRestore(); unsubscribe(); } + }); + + it("rolls back work intent when the locked retry deadline update matches no row", async () => { + const f = await failedRemoteRun("remote"); + await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + const name = `retry_noop_${f.runId.replaceAll("-", "")}`; + await f.db.execute(sql.raw(`create function ${name}() returns trigger language plpgsql as $$begin return null; end$$`)); + await f.db.execute(sql.raw(`create trigger ${name} before update on native_run_finalizations for each row + when (old.run_id = '${f.runId}'::uuid) execute function ${name}()`)); + const [before] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + try { + const now = new Date(); + await expect(claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], now })).resolves.toEqual([{ kind: "awaiting_evidence", + runId: f.runId, reason: NATIVE_RESTART_RECOVERY_FAILED, retryAt: new Date(now.getTime() + 30_000) }]); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(before); + await reconcileDatabaseWork(f.db, DELIVERY_QUEUES.nativeProcessSessions); + expect(databaseWorkPending(f.db, DELIVERY_QUEUES.nativeProcessSessions)).toBe(false); + } finally { + await f.db.execute(sql.raw(`drop trigger ${name} on native_run_finalizations`)); + await f.db.execute(sql.raw(`drop function ${name}()`)); + } + }); + + it("does not publish or log a deferred transition when an ineligible retry still has a live lease", async () => { + const f = await failedRemoteRun("remote"); + const now = new Date(), leaseDeadline = new Date(now.getTime() + 60_000); + await f.db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + await f.db.update(nativeRunFinalizations).set({ leaseOwner: "live-controller", leaseExpiresAt: leaseDeadline }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + const [before] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(f.db, DELIVERY_QUEUES.nativeProcessSessions, wake); + try { + for (let index = 0; index < 2; index += 1) { + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], now })).toMatchObject([{ kind: "blocked", + retryAt: leaseDeadline, transitionUnchanged: true }]); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(before); + expect(wake).not.toHaveBeenCalled(); + } + } finally { unsubscribe(); } + }); + + it.each(["released", "retained"] as const)("keeps exhausted %s retries fenced behind live owners, Stops and newer input", async lock => { + for (const gate of ["live-owner", "Stop", "newer-input", "missing-authority"]) { + const f = await failedRemoteRun("remote"); + if (lock === "retained") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + await f.db.update(nativeRunFinalizations).set({ attempt: 3, + ...(gate === "live-owner" ? { leaseOwner: "live-controller", leaseExpiresAt: new Date(Date.now() + 60_000) } : {}), + }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (gate === "Stop") await f.db.update(heartbeatRuns).set({ resultJson: { + nativeCancellation: { scope: "run", dispatchState: "pending" }, + } }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "newer-input") await f.db.insert(heartbeatRuns).values({ companyId: f.companyId, + agentId: f.agentId, issueId: f.issueId, status: "succeeded", createdAt: new Date(Date.now() + 1_000) }); + if (gate === "missing-authority") await f.db.delete(nativeRunnerCheckpoints) + .where(and(eq(nativeRunnerCheckpoints.companyId, f.companyId), eq(nativeRunnerCheckpoints.runId, f.runId))); + const onClaim = vi.fn(); + expect((await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId], onClaim }))[0].kind).toBe(gate === "live-owner" ? "awaiting_evidence" : "blocked"); + expect(onClaim).not.toHaveBeenCalled(); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBe(lock === "retained" ? f.runId : null); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: gate === "newer-input" ? "terminal_failure" : "retryable_failure", attempt: 3 }); + const actions = await f.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + if (gate === "newer-input") expect(actions).toMatchObject([{ ownerType: "board", wakePolicy: null }]); + else expect(actions).toEqual([]); + } + }); + + it("preserves the charged attempt when replacing an active remote controller", async () => { + const f = await failedRemoteRun("remote"); + await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + await f.db.update(heartbeatRuns).set({ status: "running", nativePhase: "observed" }).where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(nativeRunFinalizations).set({ attempt: 2, phase: "observed", failureCode: null, failureDetail: null }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId] })).toMatchObject([{ kind: "reattach_remote_runner", providerAttempt: 2 }]); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "observed", attempt: 2, controllerGeneration: 2 }); + }); + + it("ignores another company's newer input when reclaiming this company's execution", async () => { + const f = await failedRemoteRun("remote"); + const foreignCompany = randomUUID(), foreignAgent = randomUUID(); + await f.db.insert(companies).values({ id: foreignCompany, name: "Other company", issuePrefix: "FOREIGN" }); + await f.db.insert(agents).values({ id: foreignAgent, companyId: foreignCompany, name: "Other agent", status: "active" }); + await f.db.insert(heartbeatRuns).values({ companyId: foreignCompany, agentId: foreignAgent, + status: "queued", contextSnapshot: { issueId: f.issueId }, createdAt: new Date(Date.now() + 1_000) }); + expect(await claimNativeRestartRecoveries({ db: f.db, controller: environment.current.successor, + restartKind: "hard", runIds: [f.runId] })).toMatchObject([{ kind: "reattach_remote_runner" }]); + }); +}); diff --git a/server/src/services/native-runtime/native-restart-recovery-isolation.test.ts b/server/src/services/native-runtime/native-restart-recovery-isolation.test.ts index 22cd7b5d89c..768626afb82 100644 --- a/server/src/services/native-runtime/native-restart-recovery-isolation.test.ts +++ b/server/src/services/native-runtime/native-restart-recovery-isolation.test.ts @@ -1,6 +1,9 @@ import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; import { and, eq, inArray, sql } from "drizzle-orm"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, applyPendingMigrations, companies, createDb, environmentLeases, environments, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, @@ -9,6 +12,8 @@ import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { heartbeatService } from "../heartbeat.js"; import { claimNativeRestartRecoveries, NATIVE_RESTART_RECOVERY_FAILED } from "./native-restart-recovery.js"; +import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js"; +import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TEST_DATABASE_URL"); const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); @@ -19,6 +24,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes // whose claim fails for an unrelated reason. Neither may stop the batch. (support.supported ? describe : describe.skip)("native restart recovery isolates one run's failure", () => { let db: ReturnType; + let runtimeRoot: string; let temporary: Awaited> | undefined; const companyId = randomUUID(); const agentId = randomUUID(); @@ -32,8 +38,14 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes } await db.insert(companies).values({ id: companyId, name: "Restart isolation", issuePrefix: companyId.slice(0, 8) }); await db.insert(agents).values({ id: agentId, companyId, name: "Engineer", adapterType: "paperclip_runner" }); + runtimeRoot = await mkdtemp(resolve(tmpdir(), "restart-isolation-authority-")); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "83".repeat(32)); }, 30_000); - afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); }); + afterAll(async () => { + await db?.$client.end(); await temporary?.cleanup(); + if (runtimeRoot) await rm(runtimeRoot, { recursive: true, force: true }); + vi.unstubAllEnvs(); + }); async function seedScheduledResume(options: { lock: "self" | "none"; routine?: { id: string; fingerprint: string } }) { const issueId = randomUUID(), runId = randomUUID(), sessionId = randomUUID(); @@ -51,6 +63,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes environmentId: environment.id, provider: "daytona", providerLeaseId: randomUUID(), status: "active", leasePolicy: "ephemeral", metadata: { driver: "sandbox", remoteCwd: "/work" } }).returning(); await db.update(heartbeatRuns).set({ runnerProfileJson: { + nativeExecutionInput: { binding: { companyId, issueId, runId, agentId } }, nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared", descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId: runId, @@ -63,9 +76,25 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId, executionLockedAt: failedAt }) .where(eq(issues.id, issueId)); } + const owner = { bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() }; await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "retryable_failure", attempt: 1, + controllerBootId: owner.bootId, controllerGeneration: owner.generation, + leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000), nextAttemptAt: new Date(Date.now() - 60_000), recoveryState: "resuming_session", - failureCode: "native_session_interrupted" }); + failureCode: "native_session_interrupted", + failureDetail: { recoveryMode: "exact_checkpoint_resume", checkpointExists: true, + providerSessionEstablished: true, providerEventsExist: false } }); + const authority = new DurablePrpControlPlane({ stateDirectory: resolve(runtimeRoot, runId), + identity: { runId, normalizedSessionId: sessionId, runnerInstanceId: randomUUID(), + environmentLeaseId: lease.id, turnId: randomUUID(), itemId: randomUUID() }, + expectedRunnerVersion: "restart-isolation-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + authority.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, { + companyId, issueId, runId, authorityRunId: runId, ...owner, + })); + await authority.flushCheckpoint(); + await authority.stop(); + await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, runId)); return { issueId, runId }; } diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index ba6df0e4e02..d8359ffd31f 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,4 +1,5 @@ import { readPendingNativeRunStop } from "./native-cancellation-request.js"; +import { readNativeSessionResumptionAgent, restoreNativeSessionExecution } from "./native-session-resumption-admission.js"; import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { NATIVE_RETAINED_OWNER_WAIT } from "./native-controller-recovery.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; @@ -20,7 +21,8 @@ import { recordNativeLocalProcessStop, } from "../native-local-process-stop.js"; import { randomUUID } from "node:crypto"; -import { and, desc, eq, inArray, isNotNull, isNull, lte, ne, or, sql } from "drizzle-orm"; +import { and, desc, eq, gt, inArray, isNotNull, isNull, lte, ne, or, sql } from "drizzle-orm"; +import { alias } from "drizzle-orm/pg-core"; import type { Db } from "@paperclipai/db"; import { agents, @@ -42,8 +44,9 @@ import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; import { nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; import { retireSupersededNativeRetry } from "./native-superseded-retry.js"; import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js"; -import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { requestNativeStartupCleanup, readNativeStartupContext, promoteNativeStartupCleanup } from "./native-startup-cleanup.js"; +import { createSystemRecovery, failSystemRecovery, readSystemRecovery, reserveSystemRecoveryAttempt, + SystemRecoveryFailure, type SystemRecoveryState } from "../system-recovery-state.js"; export type NativeControllerIdentity = { bootId: string; @@ -110,6 +113,7 @@ export type NativeRestartRecoveryDisposition = runId: string; reason: string; retryAt?: Date; + transitionUnchanged?: boolean; }; export function nextNativeProviderAttempt( @@ -491,11 +495,6 @@ function appendBoundedRecoveryHistory(entry: Record) { )`; } -/** - * Claims abandoned local runner executions without changing the provider retry - * attempt. The transaction locks the run, coordinator, and issue execution - * owner so two successor servers cannot both reconstruct the same authority. - */ export async function claimNativeRestartRecoveries(input: { db: Db; controller?: NativeControllerIdentity; @@ -521,6 +520,7 @@ export async function claimNativeRestartRecoveries(input: { .select({ runId: heartbeatRuns.id, issueId: nativeRunFinalizations.issueId, + companyId: heartbeatRuns.companyId, }) .from(heartbeatRuns) .innerJoin(agents, eq(agents.id, heartbeatRuns.agentId)) @@ -579,9 +579,11 @@ export async function claimNativeRestartRecoveries(input: { // Defer that concrete recovery obligation instead of failing startup // or waiting while holding a lock the serving controller needs. const [issue] = await tx - .select({ id: issues.id }) + .select({ id: issues.id, companyId: issues.companyId, status: issues.status, + executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId, + assigneeAgentId: issues.assigneeAgentId }) .from(issues) - .where(eq(issues.id, candidate.issueId)) + .where(and(eq(issues.id, candidate.issueId), eq(issues.companyId, candidate.companyId))) .for("update", { skipLocked: true }); if (!issue) return busy(); const row = await tx @@ -637,20 +639,91 @@ export async function claimNativeRestartRecoveries(input: { if (!input.liveRunExecutions?.has(row.run.id) && await retireSupersededNativeRetry(tx as unknown as Db, { run: row.run, coordinator: row.coordinator, issueExecutionRunId: row.issueExecutionRunId, now, })) return { kind: "blocked", runId: row.run.id, reason: "native_retry_superseded" } as const; - // A scheduled same-run resume can find its task lock released while no - // other run holds the task. Like the local resume claim, reclaim the - // free lock unless another run is admitted for this task. - const reclaimsExecution = row.issueExecutionRunId === null - && row.coordinator.phase === "retryable_failure" - && (row.issueCheckoutRunId === null || row.issueCheckoutRunId === row.run.id) - && !(await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, row.run.companyId), ne(heartbeatRuns.id, row.run.id), - inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), - or(eq(heartbeatRuns.issueId, row.coordinator.issueId), eq(heartbeatRuns.nativeIssueId, row.coordinator.issueId), - sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${row.coordinator.issueId}`), - )).limit(1)).length; - if (row.issueExecutionRunId !== row.run.id && !(stopCleanup && row.issueExecutionRunId === null) - && !reclaimsExecution) { + const profile = row.run.runnerProfileJson ?? {}; + const storedRemoteWorkspace = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace); + const remoteWorkspace = readNativeWorkspaceSyncReference(profile.nativeWorkspaceSync) + ?? (storedRemoteWorkspace.success ? storedRemoteWorkspace.data : null); + const [pluginTask] = await tx.select({ id: environmentLeases.id, + metadata: sql>`jsonb_build_object('runnerTaskStartup', ${environmentLeases.metadata}->'runnerTaskStartup')`, + }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, row.run.companyId), eq(environmentLeases.issueId, issue.id), + eq(environmentLeases.heartbeatRunId, row.run.id), sql`${environmentLeases.metadata}->>'runnerTaskLease' = 'true'`, + )).orderBy(desc(environmentLeases.createdAt)).limit(1); + const failedRemoteRetry = row.run.status === "failed" && row.coordinator.phase === "retryable_failure" + && row.coordinator.failureDetail?.recoveryMode === "exact_checkpoint_resume" && (remoteWorkspace !== null || pluginTask !== undefined); + const releasedRemoteRetry = failedRemoteRetry && row.issueExecutionRunId === null; + const exhaustedRemoteRetry = failedRemoteRetry && row.coordinator.attempt >= 3; + const needsResumptionAdmission = releasedRemoteRetry || exhaustedRemoteRetry || + (failedRemoteRetry && ["done", "cancelled"].includes(issue.status)); + const admission = { exhausted: null as { reason: string; policy: SystemRecoveryState } | null }; + const deferRemoteRetry = async (reason: string, permanent = false) => { + if (input.liveRunExecutions?.has(row.run.id)) { + return { kind: "awaiting_evidence", runId: row.run.id, reason: "current_controller_preparing" } as const; + } + const existingDeadline = Math.max(row.coordinator.nextAttemptAt?.getTime() ?? 0, + row.coordinator.leaseOwner ? row.coordinator.leaseExpiresAt?.getTime() ?? 0 : 0); + if (existingDeadline > now.getTime()) { + return { kind: "blocked", runId: row.run.id, reason, retryAt: new Date(existingDeadline), + transitionUnchanged: true } as const; + } + const previousPolicy = readSystemRecovery(row.coordinator.failureDetail); + const policy = previousPolicy?.sourceRunId === row.run.id ? previousPolicy : createSystemRecovery(row.run.id); + const reserved = reserveSystemRecoveryAttempt(policy, now.getTime()); + const failed = reserved ? failSystemRecovery(reserved, new SystemRecoveryFailure( + permanent ? "source_binding_invalid" : "allocation_ownership_ambiguous", + permanent ? "user_action_required" : "recheck_evidence"), { now: now.getTime(), evidenceFingerprint: reason }) + : { ...policy, state: "exhausted" as const, nextAttemptAt: null }; + if (failed.state === "exhausted") { + admission.exhausted = { reason, policy: failed }; + return null; + } + const retryAt = new Date(failed.nextAttemptAt!); + const previous = row.coordinator.recoveryHistory.at(-1); + const transitionUnchanged = previous?.disposition === "blocked" && previous.reason === reason; + await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id }); + const updated = await tx.update(nativeRunFinalizations).set({ nextAttemptAt: retryAt, updatedAt: now, + failureDetail: { ...row.coordinator.failureDetail, systemRecovery: failed }, + ...(!transitionUnchanged ? { recoveryHistory: appendBoundedRecoveryHistory({ + at: now.toISOString(), restartKind: input.restartKind, controllerBootId: controller.bootId, + controllerGeneration: row.coordinator.controllerGeneration, disposition: "blocked", reason, + recoveryRequestId: input.recoveryRequestId ?? null, providerAttempt: row.coordinator.attempt, + stateRootAction: "preserve_fail_closed", retryAt: retryAt.toISOString(), + }) } : {}), + }).where(and(eq(nativeRunFinalizations.companyId, row.run.companyId), + eq(nativeRunFinalizations.issueId, issue.id), eq(nativeRunFinalizations.runId, row.run.id), + eq(nativeRunFinalizations.phase, "retryable_failure"), eq(nativeRunFinalizations.attempt, row.coordinator.attempt), + isNull(nativeRunFinalizations.resultId), + or(isNull(nativeRunFinalizations.nextAttemptAt), lte(nativeRunFinalizations.nextAttemptAt, now)), + or(isNull(nativeRunFinalizations.leaseOwner), isNull(nativeRunFinalizations.leaseExpiresAt), lte(nativeRunFinalizations.leaseExpiresAt, now)), + )).returning({ runId: nativeRunFinalizations.runId }); + if (!updated.length) throw new Error("native_retry_deferral_claim_changed"); + return { kind: "blocked", runId: row.run.id, reason, retryAt, transitionUnchanged } as const; + }; + if (failedRemoteRetry) { + const source = alias(heartbeatRuns, "native_retry_source"); + const [newer] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .innerJoin(source, and(eq(source.companyId, row.run.companyId), eq(source.id, row.run.id))) + .where(and( + eq(heartbeatRuns.companyId, row.run.companyId), + or(gt(heartbeatRuns.createdAt, source.createdAt), and( + eq(heartbeatRuns.createdAt, source.createdAt), gt(heartbeatRuns.id, source.id), + )), + or(eq(heartbeatRuns.issueId, issue.id), eq(heartbeatRuns.nativeIssueId, issue.id), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`), + )).limit(1); + if (newer) { + const deferred = await deferRemoteRetry("native_retry_source_superseded", true); + if (deferred) return deferred; + } + } + const resumptionAgent = needsResumptionAdmission + ? await readNativeSessionResumptionAgent(tx as unknown as Db, { issue, run: row.run, coordinator: row.coordinator }) + : null; + if (needsResumptionAdmission && !resumptionAgent && !admission.exhausted) { + const deferred = await deferRemoteRetry("native_retry_admission_ineligible", ["done", "cancelled"].includes(issue.status)); + if (deferred) return deferred; + } + if (row.issueExecutionRunId !== row.run.id && !(stopCleanup && row.issueExecutionRunId === null) && !releasedRemoteRetry && !admission.exhausted) { return { kind: "blocked", runId: row.run.id, @@ -676,14 +749,6 @@ export async function claimNativeRestartRecoveries(input: { return { kind: "awaiting_evidence", runId: row.run.id, reason: "native_startup_cleanup_pending", retryAt: startupRetryAt } as const; } - const profile = row.run.runnerProfileJson ?? {}; - const storedRemoteWorkspace = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace); - const remoteWorkspace = readNativeWorkspaceSyncReference(profile.nativeWorkspaceSync) - ?? (storedRemoteWorkspace.success ? storedRemoteWorkspace.data : null); - const [pluginTask] = await tx.select({ id: environmentLeases.id, metadata: environmentLeases.metadata }).from(environmentLeases).where(and( - eq(environmentLeases.companyId, row.run.companyId), eq(environmentLeases.heartbeatRunId, row.run.id), - sql`${environmentLeases.metadata}->>'runnerTaskLease' = 'true'`, - )).orderBy(desc(environmentLeases.createdAt)).limit(1); const startup = pluginTask?.metadata?.runnerTaskStartup as { contextId?: unknown; deadlineAtMs?: unknown; runId?: unknown } | undefined; if (startup?.runId === row.run.id && typeof startup.contextId === "string" && Number.isSafeInteger(startup.deadlineAtMs) && Number(startup.deadlineAtMs) <= now.getTime() && @@ -961,10 +1026,22 @@ export async function claimNativeRestartRecoveries(input: { const ownershipChanged = row.issueAssigneeAgentId !== row.run.agentId || ["done", "cancelled"].includes(row.issueStatus); - const claimKind = ownershipChanged ? null : classification.claimKind; - const reason = ownershipChanged + if (needsResumptionAdmission && (!durableAuthority || ownershipChanged || classification.claimKind !== "reattach_remote_runner")) { + if (!admission.exhausted) { + const deferred = await deferRemoteRetry("native_retry_remote_authority_unverified"); + if (deferred) return deferred; + } + } + + const hold = admission.exhausted; + const preserveExecutionBinding = hold !== null; + const claimKind = hold || ownershipChanged || exhaustedRemoteRetry ? null : classification.claimKind; + const reason = hold?.reason ?? (ownershipChanged ? "task_ownership_or_status_changed" - : classification.reason; + : exhaustedRemoteRetry ? "execution_recovery_budget_exhausted" : classification.reason); + const nextAction = hold + ? "Automatic admission checks exhausted. The original executor stop and state remain unverified. The provider PID, checkpoint and task bindings are preserved; reconcile that execution before starting fresh work." + : "Inspect the preserved checkpoint and reconcile the previous execution before starting a fresh session."; if (!claimKind) { const generation = row.coordinator.controllerGeneration; @@ -999,9 +1076,9 @@ export async function claimNativeRestartRecoveries(input: { failureCode: "native_restart_recovery_blocked", failureDetail: { ...row.coordinator.failureDetail, + ...(hold ? { systemRecovery: hold.policy } : {}), reason, - nextAction: - "Inspect the preserved checkpoint and reconcile the previous execution before starting a fresh session.", + nextAction, }, recoveryState: "blocked", recoveryRequestId: input.recoveryRequestId ?? null, @@ -1015,11 +1092,13 @@ export async function claimNativeRestartRecoveries(input: { eq(nativeRunFinalizations.phase, row.coordinator.phase), ), ); - await notifyExternalAgentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id }); - await notifyNativeEnvironmentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id }); - await notifyInstructionCollection(tx, { companyId: row.run.companyId, runId: row.run.id }); - await notifyCancelledWorkspacePreservation(tx, { companyId: row.run.companyId, runId: row.run.id }); - await notifyUnsafeWorkspaceExportRunChange(tx, row.run); + if (!preserveExecutionBinding) { + await notifyExternalAgentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id }); + await notifyNativeEnvironmentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id }); + await notifyInstructionCollection(tx, { companyId: row.run.companyId, runId: row.run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: row.run.companyId, runId: row.run.id }); + await notifyUnsafeWorkspaceExportRunChange(tx, row.run); + } const [updatedRun] = await tx .update(heartbeatRuns) .set({ @@ -1038,7 +1117,7 @@ export async function claimNativeRestartRecoveries(input: { terminalRunToReport = updatedRun; } await notifyNativeStatusReassessment(tx, { companyId: row.run.companyId, issueId: row.coordinator.issueId }); - await tx + if (!preserveExecutionBinding) await tx .update(issues) .set({ executionRunId: null, updatedAt: now }) .where( @@ -1048,7 +1127,7 @@ export async function claimNativeRestartRecoveries(input: { eq(issues.executionRunId, row.run.id), ), ); - await tx + if (!preserveExecutionBinding) await tx .update(issues) .set({ checkoutRunId: null, updatedAt: now }) .where( @@ -1059,8 +1138,8 @@ export async function claimNativeRestartRecoveries(input: { ), ); if ( - row.issueAssigneeAgentId === row.run.agentId && - !["done", "cancelled"].includes(row.issueStatus) + preserveExecutionBinding || (row.issueAssigneeAgentId === row.run.agentId && + !["done", "cancelled"].includes(row.issueStatus)) ) await issueRecoveryActionService( tx as unknown as Db, @@ -1076,17 +1155,21 @@ export async function claimNativeRestartRecoveries(input: { runId: row.run.id, reason, providerAttempt: row.coordinator.attempt, + ...(hold ? { automaticAdmissionExhausted: true, executorStopEvidence: "unverified", + providerPid: row.run.processPid, checkpointPreserved: hasCheckpoint } : {}), }, - nextAction: - "Inspect the preserved checkpoint and reconcile the previous execution before starting a fresh session.", + nextAction, maxAttempts: 3, wakePolicy: null, supersedeOnIdentityChange: true, }); - if (updatedRun) await preserveNativeFailedWorkspace(tx as unknown as Db, updatedRun); + if (updatedRun && !preserveExecutionBinding) await preserveNativeFailedWorkspace(tx as unknown as Db, updatedRun); return { kind: "blocked", runId: row.run.id, reason } as const; } + const providerAttempt = failedRemoteRetry + ? nextNativeProviderAttempt(row.coordinator.attempt) + : row.coordinator.attempt; const generation = row.coordinator.controllerGeneration + 1; const leaseOwner = `${controller.bootId}:${generation}:${randomUUID()}`; const event = historyEntry({ @@ -1097,7 +1180,7 @@ export async function claimNativeRestartRecoveries(input: { disposition: claimKind, reason, requestId: input.recoveryRequestId ?? null, - providerAttempt: row.coordinator.attempt, + providerAttempt, processPid: row.run.processPid, processStartedAt: row.run.processStartedAt, hasCheckpoint, @@ -1114,6 +1197,7 @@ export async function claimNativeRestartRecoveries(input: { .update(nativeRunFinalizations) .set({ phase: "observed", + attempt: providerAttempt, leaseOwner, // Cleanup and earlier candidates can outlive the batch observation. // The owner row is locked: grant the full lease at this write. @@ -1185,22 +1269,9 @@ export async function claimNativeRestartRecoveries(input: { updatedAt: now, }) .where(eq(heartbeatRuns.id, row.run.id)); - if (reclaimsExecution) { - const [agent] = await tx.select({ name: agents.name }).from(agents) - .where(eq(agents.id, row.run.agentId)).limit(1); - await tx.update(issues).set({ - executionRunId: row.run.id, - executionAgentNameKey: agent?.name.trim().toLowerCase() || null, - executionLockedAt: now, - updatedAt: now, - }).where(and(eq(issues.id, row.coordinator.issueId), eq(issues.companyId, row.run.companyId), - isNull(issues.executionRunId))); - await appendHeartbeatRunEvent(tx as unknown as Db, { - companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId, - eventType: "native.recovery.execution_reclaimed", stream: "system", level: "info", - message: "Reclaimed the original task execution for native session recovery.", - payload: { issueId: row.coordinator.issueId }, - }); + + if (releasedRemoteRetry && resumptionAgent) { + await restoreNativeSessionExecution(tx as unknown as Db, { issue, run: row.run, agentName: resumptionAgent.name, now }); } claimScope = { companyId: row.run.companyId, issueId: row.coordinator.issueId, controller, remote: Boolean(durableAuthority) }; @@ -1208,7 +1279,7 @@ export async function claimNativeRestartRecoveries(input: { runId: row.run.id, leaseOwner, controllerGeneration: generation, - providerAttempt: row.coordinator.attempt, + providerAttempt, restartKind: input.restartKind, recoveryRequestId: input.recoveryRequestId ?? null, }; diff --git a/server/src/services/native-runtime/native-runner-ownership.ts b/server/src/services/native-runtime/native-runner-ownership.ts index e3895e4a61c..41d54bb5263 100644 --- a/server/src/services/native-runtime/native-runner-ownership.ts +++ b/server/src/services/native-runtime/native-runner-ownership.ts @@ -1,5 +1,39 @@ -import { heartbeatRuns } from "@paperclipai/db"; -import { sql } from "drizzle-orm"; +import { heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { and, eq, inArray, isNull, sql } from "drizzle-orm"; +import { remoteExecutionHasStopped } from "../remote-execution-termination.js"; +import { readSystemRecovery } from "../system-recovery-state.js"; + +export const nativeManualAdmissionOwnerColumns = { + runId: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId, + issueId: nativeRunFinalizations.issueId, phase: nativeRunFinalizations.phase, + resultId: nativeRunFinalizations.resultId, failureCode: nativeRunFinalizations.failureCode, + failureDetail: sql>`jsonb_build_object('systemRecovery', ${nativeRunFinalizations.failureDetail}->'systemRecovery')`, +}; +type ManualAdmissionOwner = Pick; + +export function nativeManualAdmissionHoldMatches(scope: { companyId: string; runId: string; issueId?: string }, + owner: Partial | null | undefined): boolean { + const policy = readSystemRecovery(owner?.failureDetail); + return Boolean(owner?.runId === scope.runId && owner.companyId === scope.companyId && + (!scope.issueId || owner.issueId === scope.issueId) && owner.phase === "terminal_failure" && owner.resultId === null && + ["native_restart_recovery_blocked", "native_retry_cancelled"].includes(owner.failureCode ?? "") && + policy?.state === "exhausted" && policy.sourceRunId === scope.runId); +} + +export async function nativeManualAdmissionHoldsExecution(db: Db, scope: { + companyId: string; runId: string; issueId?: string; +}): Promise { + const [hold] = await db.select(nativeManualAdmissionOwnerColumns).from(heartbeatRuns) + .innerJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId), + eq(nativeRunFinalizations.runId, heartbeatRuns.id), eq(nativeRunFinalizations.issueId, heartbeatRuns.nativeIssueId))) + .where(and(eq(heartbeatRuns.companyId, scope.companyId), eq(heartbeatRuns.id, scope.runId), + eq(heartbeatRuns.runtimeMode, "native"), scope.issueId ? eq(heartbeatRuns.nativeIssueId, scope.issueId) : undefined, + eq(nativeRunFinalizations.phase, "terminal_failure"), inArray(nativeRunFinalizations.failureCode, ["native_restart_recovery_blocked", "native_retry_cancelled"]), + isNull(nativeRunFinalizations.resultId), + )).limit(1); + return nativeManualAdmissionHoldMatches(scope, hold) && !await remoteExecutionHasStopped(db, scope.companyId, scope.runId); +} export const NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE = "native_execution_ownership_unverified"; diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index b7668e5cd08..5d3483ea5ae 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -180,7 +180,9 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( binding: { companyId, issueId: waiting.issueId, runId: waiting.runId }, bootId: successor.bootId, authority }); } finally { await authority.stop(); } await waiting.db.update(nativeRunFinalizations).set({ phase: "retryable_failure", attempt: 1, - leaseOwner: null, leaseExpiresAt: null, nextAttemptAt: new Date(0), failureCode: "provider_transport_interrupted" }) + leaseOwner: null, leaseExpiresAt: null, nextAttemptAt: new Date(0), failureCode: "provider_transport_interrupted", + failureDetail: { recoveryMode: "exact_checkpoint_resume", checkpointExists: true, + providerSessionEstablished: true, providerEventsExist: false } }) .where(eq(nativeRunFinalizations.runId, waiting.runId)); const state = async () => ({ run: await waiting.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, waiting.runId)), @@ -222,7 +224,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( const after = await state(); expect(after.run[0].runnerProfileJson).toEqual(before.run[0].runnerProfileJson); expect(after.checkpoint).toEqual(before.checkpoint); - expect(after.coordinator[0]).toMatchObject({ attempt: 1, controllerGeneration: 1 }); + expect(after.coordinator[0]).toMatchObject({ attempt: 2, controllerGeneration: 1 }); expect(after.task[0].executionRunId).toBe(waiting.runId); expect(await waiting.db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, waiting.runId), eq(heartbeatRunEvents.eventType, "native.recovery.execution_reclaimed")))).toHaveLength(1); diff --git a/server/src/services/native-runtime/native-scheduled-resume-lock.test.ts b/server/src/services/native-runtime/native-scheduled-resume-lock.test.ts index 6ddf261b653..fc2d564786f 100644 --- a/server/src/services/native-runtime/native-scheduled-resume-lock.test.ts +++ b/server/src/services/native-runtime/native-scheduled-resume-lock.test.ts @@ -1,4 +1,7 @@ import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; import { and, eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { @@ -9,6 +12,8 @@ import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { recoveryService } from "../recovery/service.js"; import { claimNativeRestartRecoveries } from "./native-restart-recovery.js"; +import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js"; +import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TEST_DATABASE_URL"); const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); @@ -19,6 +24,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes // run when its retry delay ends. (support.supported ? describe : describe.skip)("scheduled remote native resume and the task execution lock", () => { let db: ReturnType; + let runtimeRoot: string; let temporary: Awaited> | undefined; beforeAll(async () => { if (externalUrl) { @@ -28,10 +34,16 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes temporary = await startEmbeddedPostgresTestDatabase("native-scheduled-resume-lock-"); db = createDb(temporary.connectionString); } + runtimeRoot = await mkdtemp(resolve(tmpdir(), "scheduled-resume-authority-")); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "83".repeat(32)); }, 30_000); - afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); }); + afterAll(async () => { + await db?.$client.end(); await temporary?.cleanup(); + if (runtimeRoot) await rm(runtimeRoot, { recursive: true, force: true }); + vi.unstubAllEnvs(); + }); - async function seed(options: { lockedBy?: "self" | "none" } = {}) { + async function seed(options: { lockedBy?: "self" | "none"; durableAuthority?: boolean } = {}) { const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); const sessionId = randomUUID(); const failedAt = new Date(Date.now() - 5 * 60_000); @@ -64,12 +76,28 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes } }).where(eq(heartbeatRuns.id, runId)); await db.update(issues).set({ executionRunId: lock, checkoutRunId: lock, executionLockedAt: lock ? failedAt : null }) .where(eq(issues.id, issueId)); + const owner = { bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() }; await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "retryable_failure", attempt: 1, + controllerBootId: owner.bootId, controllerGeneration: owner.generation, + leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000), nextAttemptAt: new Date(Date.now() - 60_000), recoveryState: "resuming_session", failureCode: "native_session_interrupted", failureDetail: { recoveryMode: "exact_checkpoint_resume", checkpointExists: true, providerSessionEstablished: true, providerEventsExist: false }, }); + if (options.durableAuthority !== false) { + const authority = new DurablePrpControlPlane({ stateDirectory: resolve(runtimeRoot, runId), + identity: { runId, normalizedSessionId: sessionId, runnerInstanceId: randomUUID(), + environmentLeaseId: lease.id, turnId: randomUUID(), itemId: randomUUID() }, + expectedRunnerVersion: "scheduled-resume-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + authority.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, { + companyId, issueId, runId, authorityRunId: runId, ...owner, + })); + await authority.flushCheckpoint(); + await authority.stop(); + } + await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, runId)); return { companyId, agentId, issueId, runId }; } @@ -112,7 +140,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes const dispositions = await claimNativeRestartRecoveries({ db, restartKind: "hard", runIds: [f.runId] }); expect(dispositions).toEqual([expect.objectContaining({ kind: "reattach_remote_runner", runId: f.runId })]); expect(await readIssue(f.issueId)).toMatchObject({ executionRunId: f.runId }); - expect(await readCoordinator(f.runId)).toMatchObject({ phase: "observed", attempt: 1, nextAttemptAt: null }); + expect(await readCoordinator(f.runId)).toMatchObject({ phase: "observed", attempt: 2, nextAttemptAt: null }); expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) .toMatchObject({ status: "running", nativePhase: "observed" }); const events = await db.select({ eventType: heartbeatRunEvents.eventType }).from(heartbeatRunEvents) @@ -120,6 +148,17 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes expect(events).toHaveLength(1); }); + it("keeps a due remote resume held when only the provider snapshot is present", async () => { + const f = await seed({ lockedBy: "none", durableAuthority: false }); + expect(await claimNativeRestartRecoveries({ db, restartKind: "hard", runIds: [f.runId] })) + .toMatchObject([{ kind: "blocked", runId: f.runId, reason: "native_retry_remote_authority_unverified" }]); + expect(await readIssue(f.issueId)).toMatchObject({ executionRunId: null, checkoutRunId: null }); + expect(await readCoordinator(f.runId)).toMatchObject({ phase: "retryable_failure", attempt: 1 }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toMatchObject({ status: "failed" }); + expect(await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.runId), + eq(heartbeatRunEvents.eventType, "native.recovery.execution_reclaimed")))).toHaveLength(0); + }); + it("does not reclaim a task that another run owns or has admitted", async () => { for (const competitor of ["lock", "queued"] as const) { const f = await seed({ lockedBy: "none" }); @@ -128,10 +167,10 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes issueId: f.issueId, status: competitor === "lock" ? "running" : "queued", contextSnapshot: { issueId: f.issueId } }); if (competitor === "lock") await db.update(issues).set({ executionRunId: otherRunId }).where(eq(issues.id, f.issueId)); expect(await claimNativeRestartRecoveries({ db, restartKind: "hard", runIds: [f.runId] })).toEqual([ - { kind: "blocked", runId: f.runId, reason: "issue_execution_lock_changed" }, + { kind: "blocked", runId: f.runId, reason: "native_retry_source_superseded" }, ]); expect((await readIssue(f.issueId)).executionRunId).toBe(competitor === "lock" ? otherRunId : null); - expect(await readCoordinator(f.runId)).toMatchObject({ phase: "retryable_failure", attempt: 1 }); + expect(await readCoordinator(f.runId)).toMatchObject({ phase: "terminal_failure", attempt: 1 }); } }); }); diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index ce7da62cb9b..316dab9ac5a 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -13,6 +13,7 @@ import { trace } from "@opentelemetry/api"; import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace"; import { configuredEnvironment, configuredEnvironmentProjection } from "../../vendor/paperclip-runner/index.js"; import { afterEach, beforeEach, describe, expect, it, onTestFinished, vi } from "vitest"; +import { createSystemRecovery, failSystemRecovery, reserveSystemRecoveryAttempt, SystemRecoveryFailure } from "../system-recovery-state.js"; import { readFileSync } from "node:fs"; import { PgDialect } from "drizzle-orm/pg-core"; import type { SQL } from "drizzle-orm"; @@ -6702,12 +6703,17 @@ function leaseDb( } function cancellationDb(options?: { + runId?: string; coordinator?: { runId: string; assessmentId: string | null; decisionId?: string | null; phase?: string; failureCode?: string | null; + companyId?: string; + issueId?: string; + resultId?: string | null; + failureDetail?: Record | null; } | null; status?: string; failResultJsonUpdateAt?: number; @@ -6715,7 +6721,7 @@ function cancellationDb(options?: { resultJson?: Record; }) { const initialRun = { - id: execution.binding.runId, + id: options?.runId ?? execution.binding.runId, agentId: execution.binding.agentId, companyId: execution.binding.companyId, nativeIssueId: execution.binding.issueId, @@ -6743,7 +6749,7 @@ function cancellationDb(options?: { const coordinator = options && "coordinator" in options ? options.coordinator - : { runId: execution.binding.runId, assessmentId: null }; + : { runId: initialRun.id, assessmentId: null }; let forUpdateCount = 0; let resultJsonUpdateCount = 0; const updates: Array<{ table: unknown; values: Record }> = @@ -6795,7 +6801,7 @@ function cancellationDb(options?: { result.returning = () => shouldFail ? Promise.reject(new Error("post_dispatch_db_failure")) - : Promise.resolve([{ id: execution.binding.runId }]); + : Promise.resolve([{ id: initialRun.id }]); return result; }, }), @@ -8464,16 +8470,21 @@ describe("native session cancellation", () => { expect(state.cancel).not.toHaveBeenCalled(); }); - it("fences a failed retry at dispatch and preserves the same audited intent on replay", async () => { + it.each([false, true])("fences a failed retry at dispatch and preserves the same audited intent on replay (manual hold: %s)", async manualHold => { const id = "11111111-1111-4111-8111-111111111111"; - const persistence = cancellationDb({ status: "failed", - coordinator: { runId: execution.binding.runId, assessmentId: null, phase: "retryable_failure" }, + const runId = manualHold ? "22222222-2222-4222-8222-222222222222" : execution.binding.runId; + const policy = failSystemRecovery(reserveSystemRecoveryAttempt(createSystemRecovery(runId))!, + new SystemRecoveryFailure("source_binding_invalid", "user_action_required")); + const persistence = cancellationDb({ status: "failed", runId, + coordinator: { runId, assessmentId: null, phase: manualHold ? "terminal_failure" : "retryable_failure", + ...(manualHold ? { companyId: execution.binding.companyId, issueId: execution.binding.issueId, resultId: null, + failureCode: "native_restart_recovery_blocked", failureDetail: { systemRecovery: policy } } : {}) }, resultJson: { startupCancellation: { cancellationRequestId: id, retryCancellation: true } } }); - const first = await cancelNativeSession(execution.binding.runId, "operator Stop", { db: persistence.db, scope: "run", cancellationRequestId: id }); + const first = await cancelNativeSession(runId, "operator Stop", { db: persistence.db, scope: "run", cancellationRequestId: id }); expect(persistence.updates).toContainEqual(expect.objectContaining({ table: nativeRunFinalizations, values: expect.objectContaining({ phase: "terminal_failure", failureCode: "native_retry_cancelled", nextAttemptAt: null }) })); const writes = persistence.getResultJsonUpdateCount(); - await expect(cancelNativeSession(execution.binding.runId, "retry Stop", { db: persistence.db, + await expect(cancelNativeSession(runId, "retry Stop", { db: persistence.db, scope: "run", cancellationRequestId: id })).resolves.toMatchObject({ auditId: first.auditId }); expect(persistence.getResultJsonUpdateCount()).toBe(writes); }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index a38c49122ba..3bd3b3718ea 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -266,6 +266,8 @@ import { readProcessStartedAt } from "../hot-restart.js"; import { NativeRunnerOwnershipUnverifiedError, isNativeRunnerOwnershipHeld, + nativeManualAdmissionHoldMatches, + nativeManualAdmissionOwnerColumns, NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE, NATIVE_ADOPTED_RUNNER_AUTHENTICATION_TIMEOUT, } from "./native-runner-ownership.js"; @@ -7758,7 +7760,7 @@ async function performNativeCancellation( const retryCancellation = lockedRun.status === "failed" || record(record(lockedRun.resultJson).startupCancellation).retryCancellation === true; const coordinatorQuery = tx - .select({ runId: nativeRunFinalizations.runId, phase: nativeRunFinalizations.phase, failureCode: nativeRunFinalizations.failureCode }) + .select(nativeManualAdmissionOwnerColumns) .from(nativeRunFinalizations) .where( and( @@ -7785,8 +7787,11 @@ async function performNativeCancellation( runId, companyId: cancellationContext.companyId, issueId: cancellationContext.issueId, resultJson, requestId: requestedId, scope: options.scope ?? "run", }, coordinator)) throw conflict("Native retry is no longer cancellable"); + const fenceManualAdmission = coordinator.failureCode === "native_restart_recovery_blocked" && nativeManualAdmissionHoldMatches({ + companyId: cancellationContext.companyId, runId, issueId: cancellationContext.issueId, + }, coordinator); const fenceRetryCancellation = async () => { - if (coordinator.phase !== "retryable_failure") return; + if (coordinator.phase !== "retryable_failure" && !fenceManualAdmission) return; await tx .update(nativeRunFinalizations) .set({ @@ -7802,7 +7807,7 @@ async function performNativeCancellation( and( eq(nativeRunFinalizations.runId, runId), eq(nativeRunFinalizations.companyId, cancellationContext.companyId), - eq(nativeRunFinalizations.phase, "retryable_failure"), + eq(nativeRunFinalizations.phase, coordinator.phase), ), ); }; @@ -7829,7 +7834,7 @@ async function performNativeCancellation( : null; if (!existingAuditId) throw new Error("native_cancellation_intent_audit_missing"); - if (retryCancellation && coordinator.phase === "retryable_failure") { + if (retryCancellation && (coordinator.phase === "retryable_failure" || fenceManualAdmission)) { await notifyNativeResultCommit(tx, { companyId: cancellationContext.companyId, runId }); await fenceRetryCancellation(); } diff --git a/server/src/services/native-runtime/native-session-resumption-admission.ts b/server/src/services/native-runtime/native-session-resumption-admission.ts new file mode 100644 index 00000000000..6d27fff6f82 --- /dev/null +++ b/server/src/services/native-runtime/native-session-resumption-admission.ts @@ -0,0 +1,73 @@ +import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm"; +import { agents, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js"; +import { issueService } from "../issues.js"; +import { isAutomaticRecoverySuppressedByPauseHold } from "../recovery/pause-hold-guard.js"; +import { readRunCancellation } from "../run-cancellation.js"; +import { getNativeReviewAssignment } from "./native-review-participant.js"; +import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; +import { notifyStatusCardIssueChange } from "../status-card-notifications.js"; +import { normalizeAgentNameKey } from "../agent-name-key.js"; + +export async function restoreNativeSessionExecution(db: Db, input: { + issue: Pick; + run: Pick; + agentName: string; + now: Date; +}) { + const { issue, run, now } = input; + await notifyStatusCardIssueChange(db, run.companyId); + const restored = await db.update(issues).set({ executionRunId: run.id, + executionAgentNameKey: normalizeAgentNameKey(input.agentName), executionLockedAt: now, updatedAt: now, + }).where(and(eq(issues.id, issue.id), eq(issues.companyId, run.companyId), + or(isNull(issues.executionRunId), eq(issues.executionRunId, run.id)), + )).returning({ id: issues.id }); + if (!restored.length) throw new Error("native_session_execution_binding_changed"); + if (issue.executionRunId !== run.id) await appendHeartbeatRunEvent(db, { + companyId: run.companyId, runId: run.id, agentId: run.agentId, + eventType: "native.recovery.execution_reclaimed", stream: "system", level: "info", + message: "Reclaimed the original task execution for native session recovery.", payload: { issueId: issue.id }, + }); +} + +function nativeSessionResumptionStopped(run: typeof heartbeatRuns.$inferSelect, + coordinator: typeof nativeRunFinalizations.$inferSelect) { + const cancellation = readRunCancellation(run.resultJson); + const nativeCancellation = run.resultJson?.nativeCancellation as { scope?: unknown; dispatchState?: unknown } | undefined; + return (cancellation?.source === "operator" && cancellation.expected) || + (["run", "issue"].includes(String(nativeCancellation?.scope)) && + ["pending", "acknowledged"].includes(String(nativeCancellation?.dispatchState))) || + typeof coordinator.failureDetail?.successorRunId === "string"; +} + +export async function readNativeSessionResumptionAgent(db: Db, input: { + issue: Pick; + run: typeof heartbeatRuns.$inferSelect; + coordinator: typeof nativeRunFinalizations.$inferSelect; +}) { + const { issue, run, coordinator } = input; + if (issue.companyId !== run.companyId || issue.id !== run.nativeIssueId || + coordinator.companyId !== run.companyId || coordinator.issueId !== issue.id || coordinator.runId !== run.id || + !["todo", "in_progress", "in_review"].includes(issue.status) || + (issue.executionRunId && issue.executionRunId !== run.id) || + (issue.checkoutRunId && issue.checkoutRunId !== run.id)) return null; + const [agent] = await db.select().from(agents).where(and( + eq(agents.companyId, run.companyId), eq(agents.id, run.agentId), + )).limit(1); + if (!agent || !(await evaluateAgentInvokabilityFromDb(db, agent)).invokable) return null; + if (issue.assigneeAgentId !== run.agentId && !await getNativeReviewAssignment(db, { + companyId: run.companyId, issueId: issue.id, agentId: run.agentId, contextSnapshot: run.contextSnapshot, + })) return null; + if (await isAutomaticRecoverySuppressedByPauseHold(db, run.companyId, issue.id)) return null; + if (!(await issueService(db).getDependencyReadiness(issue.id, db)).isDependencyReady) return null; + const [competing] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), ne(heartbeatRuns.id, run.id), + inArray(heartbeatRuns.status, issue.executionRunId === run.id + ? ["running", "scheduled_retry"] : ["queued", "running", "scheduled_retry"]), + or(eq(heartbeatRuns.issueId, issue.id), eq(heartbeatRuns.nativeIssueId, issue.id), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`), + )).limit(1); + if (competing) return null; + if (nativeSessionResumptionStopped(run, coordinator)) return null; + return agent; +} diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index b380f431ed5..602e98a23b0 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -92,6 +92,7 @@ import { parseObject, asBoolean, asNumber } from "../../adapters/utils.js"; import { runningProcesses } from "../../adapters/index.js"; import { isNativeRunnerOwnershipHeld, + nativeManualAdmissionHoldsExecution, nativeRunnerOwnershipNotHeldCondition, } from "../native-runtime/native-runner-ownership.js"; import { visibleIssueCondition } from "../issue-visibility.js"; @@ -6526,6 +6527,13 @@ export function recoveryService( await deps.beforeStaleLockClear?.(issue.id); const updated = await db.transaction(async tx => { + const [locked] = await tx.select({ id: issues.id, checkoutRunId: issues.checkoutRunId, executionRunId: issues.executionRunId }) + .from(issues).where(and(eq(issues.companyId, issue.companyId), eq(issues.id, issue.id))).for("update"); + if (!locked || locked.checkoutRunId !== issue.checkoutRunId || locked.executionRunId !== issue.executionRunId) return null; + for (const runId of new Set([locked.checkoutRunId, locked.executionRunId])) { + if (runId && await nativeManualAdmissionHoldsExecution(tx as unknown as Db, + { companyId: issue.companyId, runId, issueId: issue.id })) return null; + } await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id }); await notifyNativeStatusReassessment(tx, { companyId: issue.companyId, issueId: issue.id }); return tx diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index b88e3f7fd19..53e20e9eadc 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -38,7 +38,7 @@ import * as storageBackend from "../storage/index.js"; import { migrateRemoteWorkspace } from "./remote-workspace-migration.js"; import { attachPreservedWorkspaceSnapshots } from "./workspace-recovery-snapshot.js"; import { adoptHistoricalWorkspaceRecoveries, historicalRecoveryCounts } from "./workspace-recovery-backfill.js"; -import { createSystemRecovery, failSystemRecovery, SystemRecoveryFailure } from "./system-recovery-state.js"; +import { createSystemRecovery, failSystemRecovery, reserveSystemRecoveryAttempt, SystemRecoveryFailure } from "./system-recovery-state.js"; import { requestTaskRetry } from "./task-retry.js"; import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; import { materializeIsolatedTaskDirectory } from "./isolated-task-directory.js"; @@ -1654,16 +1654,34 @@ describe("automatic retained workspace recovery", () => { expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.companyId, f.companyId))).toHaveLength(0); }, 30_000); - 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 => { + it.each(["HTTP Retry", "manual admission Retry", "new comment", "automatic cleanup"] as const)("repairs a retained 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" }) + const manual = entry === "manual admission Retry"; + await db.update(heartbeatRuns).set({ resultJson: null, + errorCode: manual ? "native_restart_recovery_blocked" : "execution_finalization_deadline_exceeded", + ...(manual ? { processPid: 1386, nativePhase: "terminal_failure" } : {}) }) .where(eq(heartbeatRuns.id, f.runId)); await db.update(environmentLeases).set({ status: "active", leasePolicy: "reuse_by_environment", releasedAt: null, cleanupStatus: null, metadata: f.lease.metadata }).where(eq(environmentLeases.id, f.lease.id)); - await db.update(issueRecoveryActions).set({ status: "resolved", outcome: "blocked", - cause: "execution_finalization_deadline_exceeded", evidence: { runId: f.runId, providerOwnership: "unverified", - automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }).where(eq(issueRecoveryActions.id, f.actionId)); + if (manual) { + const systemRecovery = failSystemRecovery(reserveSystemRecoveryAttempt(createSystemRecovery(f.runId))!, + new SystemRecoveryFailure("allocation_ownership_ambiguous", "user_action_required")); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", resultId: null, + failureCode: "native_restart_recovery_blocked", failureDetail: { recoveryMode: "exact_checkpoint_resume", systemRecovery } }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(issues).set({ executionRunId: f.runId, checkoutRunId: f.runId }).where(eq(issues.id, f.issueId)); + await db.update(issueRecoveryActions).set({ status: "active", outcome: "blocked", cause: "native_restart_recovery_blocked", + fingerprint: `native-restart:${f.runId}`, returnOwnerAgentId: source.agentId, wakePolicy: null, maxAttempts: 3, + evidence: { runId: f.runId, automaticAdmissionExhausted: true, executorStopEvidence: "unverified", + providerPid: 1386, checkpointPreserved: true }, + nextAction: "Automatic admission checks exhausted; original executor stop and state remain unverified" }) + .where(eq(issueRecoveryActions.id, f.actionId)); + } else { + await db.update(issueRecoveryActions).set({ status: "resolved", outcome: "blocked", + cause: "execution_finalization_deadline_exceeded", evidence: { runId: f.runId, providerOwnership: "unverified", + automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }).where(eq(issueRecoveryActions.id, f.actionId)); + } // Leave admission observable without substituting a fake model turn. await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId, status: "running" }); let releaseStop!: () => void; @@ -1685,7 +1703,9 @@ describe("automatic retained workspace recovery", () => { app.use(errorHandler); let requestId: string | undefined; let commentId: string | undefined; - if (entry === "HTTP Retry") { + if (manual) expect((await request(app).get(`/api/issues/${f.issueId}`).expect(200)).body.interruption) + .toMatchObject({ state: "failed", canRetry: true, observedRunId: f.runId }); + if (entry === "HTTP Retry" || manual) { 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; @@ -1714,12 +1734,16 @@ describe("automatic retained workspace recovery", () => { if (entry === "automatic cleanup") await heartbeat.sweepOrphanedActiveLeases({ backoffMs: 0 }); } await expect.poll(() => f.runtime.retryPendingSandboxTeardown.mock.calls.length, { timeout: 20_000 }).toBe(1); - if (requestId) await heartbeat.resumeTaskRetry(f.companyId, requestId); - else await heartbeat.resumeQueuedRuns(); + if (!manual) { + 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) .not.toHaveProperty("remoteExecutionTermination"); + if (manual) expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]) + .toMatchObject({ executionRunId: f.runId, checkoutRunId: f.runId }); await expect(readFile(path.join(f.host, "saved.txt"), "utf8")).rejects.toMatchObject({ code: "ENOENT" }); releaseStop(); await expect.poll(async () => (await successors()).length, { timeout: 20_000 }).toBe(1);