From 026dc6d135bbfcfc845fd48b230526c0a3bf2b9b Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Fri, 6 Nov 2026 07:19:09 -0800 Subject: [PATCH] fix(native): release a session-resume owner when the resumed run fails before its provider claim (#19371) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runs keep a finalization row. Its lease owner tells the control plane that an execution may still be live. > - Pausing an agent waits for `agentExecutionsHaveStopped`. That check refuses to finish while any run of the agent has a lease owner, unless the run was cancelled and has stop proof. > - A same-run session resume claims that lease before it dispatches the run. If the run then fails before the native executor takes its own provider claim, nothing releases the resume owner. > - The leaked owner blocks pause and resume for good, and the task keeps looking like it has a live recovery path. > - This pull request settles those rows at the source, and it adds a stop-proof-gated release for rows that already leaked. > - The benefit is that agents can pause and resume after an ordinary resume failure, without database edits, and the cancelled-run stop-proof rule stays as it is. ## Linked Issues or Issue Description **What happened?** An agent stayed in `pausing` with `hostComplete=false` and climbing attempts. Its runs had `native_run_finalizations` rows with `phase: observed`, `recovery_state: resuming_session`, a `lease_owner` like `default:resume:`, and no `result_id`. Each heartbeat run had ended `failed`: either `setup_failed` (`Failed to acquire lease for environment … Sandbox creation has failed`, `native_remote_recovery_lease_mismatch`) or `adapter_failed` (`runner_state_identity_mismatch: …`). The lease expired about 20 minutes after the run finished. Nothing settled the row. The only fix was to set `lease_owner` to null by hand. **Expected behavior** When a resumed run fails before any provider turn starts, its finalization row goes to `terminal_failure` and its lease owner is released, as the native executor does for failures it sees. A row that leaked before this fix is released in the background once its lease expires and the run has stop proof. Pause then completes. **Steps to reproduce** 1. Leave a native run in `retryable_failure` with a persisted checkpoint, on a task that already has an active recovery incident. 2. Let session resume claim it (`claimNativeSessionResumptions` sets `phase: observed` and a `…:resume:…` owner). 3. Make the resumed run fail before the executor claims it, for example when the environment driver cannot create the sandbox. 4. The heartbeat failure path marks the run `failed`. The issue release path does not settle the row, because it settles only when it opens a new incident. The owner stays set, and `agentExecutionsHaveStopped` returns false indefinitely. **Paperclip version or commit** `master` at 5ac05871b5. **Additional context** Root cause: two writers can end a resumed run without the executor. `failRun` and the setup-failure path in `heartbeat/run-completion.ts` never touch the coordinator. `recordNativeTerminalRecoveryIfNeeded` in the wake-queue adapter settles the coordinator only when no recovery incident exists yet. Agents that need their computer connection paired again usually already have an incident, so the owner leaks. Related, but different: #18905, #18866, #18072, and #17692 change how the stop check judges runs. This pull request does not change the stop check. ## What Changed - `server/src/services/native-runtime/native-failed-run-owner-release.ts` (new): one settlement for a result-less `observed` owner on a run that ended `failed`, `timed_out`, or `interrupted`. It writes `terminal_failure`, null lease fields, `recoveryState: "blocked"`, the run's failure code, and a `releasedOwner` record. It sets `nativePhase`, sends the same finalization and execution work notifications as the executor's terminal path, and logs `native.failed_run_owner_released` activity. Locks follow the issue, coordinator, run order. - `settleUnstartedNativeSessionResume`: runs after a dispatched session resume returns. It matches only the exact claim owner and the attempt the claim saw. The executor increments the attempt when it takes provider ownership, so an unchanged attempt proves that no provider turn started under this owner. The executor's own failures no longer match. - `releaseAbandonedNativeOwners`: the background release for rows that already leaked. It requires an expired lease and all of the stop proof that `agentExecutionsHaveStopped` checks: settled environment leases, no live run process or process group, no live controller lease, and no live controller process. This process answers from its in-memory execution registries. It skips cancelled runs, pending native Stops, and pinned workspace restores, because those have their own stop-proof owners. - `native-finalization-reconciler.ts`: `NativeSessionResumeClaim` now carries `attempt`. - `heartbeat.ts` and `heartbeat/recovery.ts`: both session-resume dispatch sites settle the claim after `executeRun` returns. This is one `.finally` per site. - `execution-control-reconciliation.ts` and `execution-control-work.ts`: the execution-finalization queue runs the background release. Its deadline includes the earliest lease expiry among eligible rows, so the release does not poll an idle instance. Rows with unreleased sandboxes are excluded from the deadline. - `agent-execution-stop.ts` is unchanged. The cancelled-only release keeps its stop-proof rule. ## Verification - `npx vitest run src/__tests__/heartbeat-native-resume-lease-failure.test.ts` drives the real heartbeat resume dispatch (`reapOrphanedRuns`). A sandbox driver's `acquireRunLease` throws `Sandbox creation has failed`, and the task has a recovery incident. The run fails with `Failed to acquire lease for environment "Managed computer" (sandbox): Sandbox creation has failed`, and the coordinator ends `terminal_failure` with no lease owner. I removed the `.finally` wiring and ran the test again. It then failed, and the row stayed `observed` with a `default:resume:` owner and a 20-minute expiry, which is the production state. - `npx vitest run src/services/native-runtime/native-failed-run-owner-release.test.ts` (10 tests): - A claimed resume that fails at lease acquisition is settled. `agentExecutionsHaveStopped` changes from false to true. A repeated settlement does nothing. - An owner the executor claimed (incremented attempt) and a run that is still running are not changed. - A leaked `runner_state_identity_mismatch` row is released by `reconcileAbandonedExecutionControl`, and then `agentExecutionsHaveStopped` returns true. - An unexpired owner still blocks the stop check. Its expiry becomes the finalization queue deadline. - An active environment lease and a live controller process block release. After both stop, the row is released. - Cancelled runs and pending native Stops are not changed. - Added during review: - Releasing an owner wakes the cost-accounting worker. - Releasing a run's environment lease wakes the finalization worker, and the owner is then released. - A recorded PID and group that now belong to an unrelated process do not hold the owner. - Bounded passes rotate past a held row (`[0, 1, 1]` with page size 1), then wrap back to it. - The immediate settle keeps the owner while the run's recorded process is alive. - Suites that cover the changed files pass together: the two new files, `native-session-resumption.test.ts` (with the claim shape updated), `execution-control-reconciliation.test.ts`, `execution-control-work.test.ts`, and `agent-lifecycle/__tests__/lifecycle.test.ts`. With `environment-cleanup-work.test.ts` added, that is 7 files and 156 tests. I did not run the full `heartbeat-process-recovery.test.ts` suite locally. CI runs it. - `pnpm --filter @paperclipai/server typecheck` passes. ## Risks - Behaviour change: a resumed run that fails during setup now ends with a terminal coordinator in every case. Before, this happened only when the issue release path opened a new incident. After a restart, restart recovery no longer sees such a row as `observed`. This matches what the release path already does when no incident exists. - The background release trusts an expired lease on a run that is already terminal, together with the same process and lease evidence that the stop check uses. A controller on another host cannot be probed by PID. The stop check has the same limit today. A live owner still renews its lease, and lease-gated writes from an expired owner are already refused. - If an expired row's process or controller still looks live, the release retries at the shared 15-second finalization cadence until that process exits. Each pass handles one bounded page, and a rotating cursor moves to the next page on each pass. - A run-scoped environment cleanup notification now also wakes the execution-finalization queue. Passes are coalesced, and a pass is a cheap deadline query when nothing is eligible. - No schema change and no migration. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`) through Claude Code, with tool use (shell, file edits, local test runs). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/idle-execution-control.md | 2 +- ...rtbeat-native-resume-lease-failure.test.ts | 88 ++++++ .../native-session-resumption.test.ts | 7 +- .../environment-cleanup-notifications.ts | 2 + .../execution-control-reconciliation.ts | 9 +- server/src/services/execution-control-work.ts | 6 +- server/src/services/heartbeat.ts | 3 +- server/src/services/heartbeat/recovery.ts | 3 +- .../native-failed-run-owner-release.test.ts | 294 ++++++++++++++++++ .../native-failed-run-owner-release.ts | 271 ++++++++++++++++ .../native-finalization-reconciler.ts | 7 +- 11 files changed, 681 insertions(+), 11 deletions(-) create mode 100644 server/src/__tests__/heartbeat-native-resume-lease-failure.test.ts create mode 100644 server/src/services/native-runtime/native-failed-run-owner-release.test.ts create mode 100644 server/src/services/native-runtime/native-failed-run-owner-release.ts diff --git a/doc/idle-execution-control.md b/doc/idle-execution-control.md index a18600124ec..7cdc5458c99 100644 --- a/doc/idle-execution-control.md +++ b/doc/idle-execution-control.md @@ -9,7 +9,7 @@ inside the six owning services. No table or migration is added. | Duty | Trigger and durable source | Remaining scheduled work | Quiet condition | | --- | --- | --- | --- | -| Abandoned finalization | Committed control-deadline writes on `heartbeat_runs` and `native_run_finalizations` | Earliest non-null control deadline; a due row still present after recovery retries after 15 seconds | No control deadlines | +| Abandoned finalization | Committed control-deadline writes on `heartbeat_runs` and `native_run_finalizations`; committed run transitions that leave a result-less `observed` owner on a failed, timed-out, or interrupted native run | Earliest non-null control deadline, or earliest lease expiry of such an owner whose environment leases are settled; a due row still present after recovery retries after 15 seconds | No control deadlines and no such owner | | Safe native replacement | Committed run transitions and reconciliation decisions; existing failed run/coordinator rows | Retry while the existing replacement predicate selects a candidate, including one awaiting provider termination or sandbox release | No replacement candidate, or a recorded successor/denial | | Reconciled continuation | Committed recovery-action marking/resolution; `evidence.continuationDelivery` | Retry pending delivery after 15 seconds | No resolved action awaiting continuation delivery | | Execution status | Committed run status writes; `execution_status_delivery_id` | Retry while a delivery marker remains | All markers acknowledged | diff --git a/server/src/__tests__/heartbeat-native-resume-lease-failure.test.ts b/server/src/__tests__/heartbeat-native-resume-lease-failure.test.ts new file mode 100644 index 00000000000..a0f5c1bd6ba --- /dev/null +++ b/server/src/__tests__/heartbeat-native-resume-lease-failure.test.ts @@ -0,0 +1,88 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { agents, authUsers, companyMemberships, environments, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { beforeAll, expect, it, vi } from "vitest"; +import { heartbeatService } from "../services/heartbeat.js"; +import { classifyEnvironmentCapabilities, environmentRuntimeService } from "../services/environment-runtime.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; +import { buildNativeExecutionInput } from "../services/native-runtime/native-execution-input.js"; +import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js"; +import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js"; + +describeEmbeddedPostgres("heartbeat native session resume environment failure", () => { + const fixture = nativeRestartFixture(); + let companyId: string, agentId: string; + beforeAll(() => { ({ companyId, agentId } = fixture.current); }); + + it("releases the resume owner when the resumed run cannot acquire its environment lease", async () => { + const f = await fixture.seedRun("RESUME-LEASE-FAILURE"); + const db = f.db, environmentId = randomUUID(); + const settings = instanceSettingsService(db), previous = await settings.getExperimental(); + let heartbeat: ReturnType | undefined; + try { + const userId = randomUUID(); + await db.insert(authUsers).values({ id: userId, name: "Operator", email: `${userId}@example.invalid`, + createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" }); + await db.insert(environments).values({ id: environmentId, name: "Managed computer", driver: "sandbox", + config: { provider: "fake", image: "fixture:test" } }); + await db.update(agents).set({ defaultEnvironmentId: environmentId }).where(eq(agents.id, agentId)); + const execution = buildNativeExecutionInput({ companyId, agentId, runId: f.runId, + issue: { id: f.issueId, identifier: "NRR-RESUME-LEASE-FAILURE", title: "Resume after a sandbox failure", + description: null, workMode: "standard" }, + taskPrompt: "Continue the original task.", provider: "codex", + workspace: { id: randomUUID(), cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null }, + normalizedSessionId: f.native.normalizedSessionId, runtimeContext: nativeRuntimeContextFixture(), + completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1", + contract: { revision: "1", objective: "Resume", criteria: [{ id: "objective", requirement: "Resume the session" }] } } }); + const identity = { companyId, issueId: f.issueId, runId: f.runId, agentId, sessionId: f.native.normalizedSessionId }; + // An earlier provider failure left a retryable, result-less run for same-run resume. + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), errorCode: "native_session_interrupted", + runnerProfileJson: { ...f.preparedProfile, nativeExecutionInput: execution, + sessionCheckpoint: { identity, providerSessionId: "original-provider" } }, responsibleUserId: userId }) + .where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0), + recoveryState: "resuming_session", failureCode: "native_session_interrupted" }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + // The task already has an open incident, so releasing the issue lock does + // not open another one and therefore never settles the coordinator. + await issueRecoveryActionService(db).upsertSourceScoped({ companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: agentId, cause: "native_session_interrupted", + fingerprint: `native-interruption:${f.runId}`, evidence: { runId: f.runId }, + nextAction: "Re-pair the computer connection.", maxAttempts: 3, wakePolicy: null, supersedeOnIdentityChange: true }); + + const acquireRunLease = vi.fn(async () => { throw new Error("Sandbox creation has failed"); }); + const unexpected = async () => { throw new Error("A failed lease acquisition must not reach the sandbox"); }; + const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", acquireRunLease, + resumeRunLease: unexpected, releaseRunLease: async () => null, retryPendingSandboxTeardown: async () => undefined, + realizeWorkspace: unexpected, execute: unexpected, syncIn: unexpected, syncOut: unexpected, supportsSync: () => true, + resolveCapabilities: async () => classifyEnvironmentCapabilities({ + verifiedMethods: ["environmentSyncIn", "environmentSyncOut", "environmentExecute"], declared: null, narrowing: null }), + }] }); + const backendFactory = vi.fn(() => { throw new Error("No provider backend may be created"); }); + await settings.updateExperimental({ enableNativeRunner: true }); + heartbeat = heartbeatService(db, { environmentRuntime: runtime, nativeSessionBackendFactory: backendFactory, + runtimeEnv: { PAPERCLIP_INSTANCE_ID: "default" } }); + + await heartbeat.reapOrphanedRuns({ staleThresholdMs: 0 }); + await heartbeat.drainActiveRunExecutions(); + + expect(acquireRunLease).toHaveBeenCalledOnce(); + expect(backendFactory).not.toHaveBeenCalled(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(run).toMatchObject({ status: "failed", errorCode: "setup_failed", + error: expect.stringContaining('Failed to acquire lease for environment "Managed computer"') }); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(coordinator).toMatchObject({ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null, + recoveryState: "blocked", resultId: null, failureCode: "setup_failed", + failureDetail: expect.objectContaining({ releasedOwner: expect.objectContaining({ reason: "unstarted_resume", + leaseOwner: expect.stringContaining("default:resume:"), attempt: 1 }) }) }); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(task.executionRunId).toBeNull(); + } finally { + try { await heartbeat?.drainActiveRunExecutions(); } + finally { await settings.updateExperimental(previous); } + } + }, 60_000); +}); diff --git a/server/src/__tests__/native-session-resumption.test.ts b/server/src/__tests__/native-session-resumption.test.ts index 0040f019223..82babcd16b7 100644 --- a/server/src/__tests__/native-session-resumption.test.ts +++ b/server/src/__tests__/native-session-resumption.test.ts @@ -342,6 +342,7 @@ describe("P6-25 pre-result native session recovery", () => { { runId: eligibleRunId, leaseOwner: expect.stringContaining("bounded-reaper:resume:"), + attempt: expect.any(Number), }, ]); const held = await db @@ -388,14 +389,14 @@ describe("P6-25 pre-result native session recovery", () => { leaseExpiresAt: null, nextAttemptAt: new Date(0), }).where(eq(nativeRunFinalizations.runId, runId)); - const dispatched: Array<{ runId: string; leaseOwner: string }> = []; + const dispatched: Array<{ runId: string; leaseOwner: string; attempt: number }> = []; await expect(dispatchNativeSessionResumptions({ db, runnerInstanceId: "heartbeat-reaper", runIds: [runId], dispatch: (claim) => dispatched.push(claim), })).resolves.toHaveLength(1); - expect(dispatched).toEqual([{ runId, leaseOwner: expect.stringContaining("heartbeat-reaper:resume:") }]); + expect(dispatched).toEqual([{ runId, leaseOwner: expect.stringContaining("heartbeat-reaper:resume:"), attempt: 1 }]); await expect(db.select().from(heartbeatRuns)).resolves.toHaveLength(8); await expect(db.select().from(nativeRunFinalizations)).resolves.toHaveLength(8); }); @@ -406,7 +407,7 @@ describe("P6-25 pre-result native session recovery", () => { runnerInstanceId: "reaper", runIds: [bootstrapRetryRunId], })).resolves.toEqual([ - { runId: bootstrapRetryRunId, leaseOwner: expect.stringContaining("reaper:resume:") }, + { runId: bootstrapRetryRunId, leaseOwner: expect.stringContaining("reaper:resume:"), attempt: 1 }, ]); }); diff --git a/server/src/services/environment-cleanup-notifications.ts b/server/src/services/environment-cleanup-notifications.ts index 958ec68666b..fbbf3352359 100644 --- a/server/src/services/environment-cleanup-notifications.ts +++ b/server/src/services/environment-cleanup-notifications.ts @@ -30,6 +30,8 @@ export async function notifyEnvironmentCleanup(transaction: object, scope?: Envi } await notifyNativeEnvironmentSettlement(transaction, scope); await notifyQueuedRunDispatch(transaction, scope); + // A settled run lease can make that run's abandoned native owner releasable. + if (scope?.runId) await notifyDeliveryWork(transaction, DELIVERY_QUEUES.executionFinalization); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.environmentCleanup); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup); } diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts index 3e4c5a080cd..dffbf539735 100644 --- a/server/src/services/execution-control-reconciliation.ts +++ b/server/src/services/execution-control-reconciliation.ts @@ -25,6 +25,7 @@ import { preserveNativeFailedWorkspace } from "./native-runtime/native-failed-wo import { EXECUTION_RECONCILIATION_INTERVAL_MS } from "./execution-control-deadline.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { NATIVE_STARTUP_CLEANUP_PENDING, readNativeStartupContext, promoteNativeStartupCleanup } from "./native-runtime/native-startup-cleanup.js"; +import { releaseAbandonedNativeOwners } from "./native-runtime/native-failed-run-owner-release.js"; /** Only newly recorded control deadlines are eligible. Upgrades never replay ambiguous historical runs. */ export async function reconcileAbandonedExecutionControl( @@ -346,5 +347,11 @@ export async function reconcileAbandonedExecutionControl( } } })); - return { scanned: due.length, surfaced }; + // Owners left on runs that already ended outside the native executor have no + // control deadline; their expired lease is the due time. + const releasedOwners = await releaseAbandonedNativeOwners(db, now).catch((error: unknown) => { + logger.warn({ err: error }, "Abandoned native owner release remains pending"); + return 0; + }); + return { scanned: due.length, surfaced, releasedOwners }; } diff --git a/server/src/services/execution-control-work.ts b/server/src/services/execution-control-work.ts index f282b9d5820..7cedc0c2ea3 100644 --- a/server/src/services/execution-control-work.ts +++ b/server/src/services/execution-control-work.ts @@ -6,6 +6,7 @@ import { type DeliveryQueue } from "./delivery-work-notifications.js"; import { EXECUTION_RECONCILIATION_INTERVAL_MS } from "./execution-control-deadline.js"; import { obsoleteExecutionConversationHold, unresolvedExecutionDispositions } from "./execution-recovery-resolution.js"; import { nativeReplacementPending } from "./native-runtime/native-safe-replacement.js"; +import { nextAbandonedNativeOwnerReleaseAt } from "./native-runtime/native-failed-run-owner-release.js"; import { hasPendingPauseResume } from "./issue-tree-resume.js"; export function registerExecutionControlWork( @@ -59,13 +60,14 @@ export function executionControlSchedule(db: Db) { }; return { async finalization() { - const [native, legacy] = await Promise.all([ + const [native, legacy, abandonedOwner] = await Promise.all([ db.select({ at: nativeRunFinalizations.controlDeadlineAt }).from(nativeRunFinalizations) .where(isNotNull(nativeRunFinalizations.controlDeadlineAt)).orderBy(asc(nativeRunFinalizations.controlDeadlineAt)).limit(1), db.select({ at: heartbeatRuns.executionControlDeadlineAt }).from(heartbeatRuns) .where(isNotNull(heartbeatRuns.executionControlDeadlineAt)).orderBy(asc(heartbeatRuns.executionControlDeadlineAt)).limit(1), + nextAbandonedNativeOwnerReleaseAt(db), ]); - return deadline([native[0]?.at, legacy[0]?.at]); + return deadline([native[0]?.at, legacy[0]?.at, abandonedOwner]); }, async workspaceRecovery() { const at = await nextWorkspaceRecoveryAt(db); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 4b6ce6b629d..4f54e6d2915 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -406,6 +406,7 @@ import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE, } from "./native-runtime/native-runner-ownership.js"; import { NativeStartupCleanupPendingError, requestNativeStartupCleanup, verifyNativeStartupCleanup, readNativeStartupCleanup } from "./native-runtime/native-startup-cleanup.js"; +import { settleUnstartedNativeSessionResume } from "./native-runtime/native-failed-run-owner-release.js"; import { emitAgentTaskRun, emitAgentTaskRunById, @@ -1783,7 +1784,7 @@ export function heartbeatService( { err: error, runId: claim.runId }, "scheduled native session resume failed", ); - }); + }).finally(() => settleUnstartedNativeSessionResume(db, claim)); activeRunExecutionPromises.add(execution); void execution.finally(() => activeRunExecutionPromises.delete(execution), diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index e485af9e49f..c3501633098 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -74,6 +74,7 @@ import { reconcileNativeFinalizations, reconcileRetainedNativeSessionCleanup, } from "../native-runtime/index.js"; +import { settleUnstartedNativeSessionResume } from "../native-runtime/native-failed-run-owner-release.js"; import { runningProcesses } from "../../adapters/index.js"; import { parseObject } from "../../adapters/utils.js"; import { @@ -2183,7 +2184,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD { err: error, runId: claim.runId }, "persisted native session resume failed", ); - }); + }).finally(() => settleUnstartedNativeSessionResume(db, claim)); activeRunExecutionPromises.add(execution); void execution.finally(() => activeRunExecutionPromises.delete(execution), diff --git a/server/src/services/native-runtime/native-failed-run-owner-release.test.ts b/server/src/services/native-runtime/native-failed-run-owner-release.test.ts new file mode 100644 index 00000000000..d65ecf3850d --- /dev/null +++ b/server/src/services/native-runtime/native-failed-run-owner-release.test.ts @@ -0,0 +1,294 @@ +import { spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { once } from "node:events"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { and, eq } from "drizzle-orm"; +import { + activityLog, + agents, + companies, + createDb, + EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, + environmentLeases, + heartbeatRuns, + issues, + nativeRunFinalizations, + subscribeDatabaseWork, +} from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { agentExecutionsHaveStopped } from "../agent-execution-stop.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { environmentService } from "../environments.js"; +import { executionControlSchedule } from "../execution-control-work.js"; +import { reconcileAbandonedExecutionControl } from "../execution-control-reconciliation.js"; +import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; +import { + nextAbandonedNativeOwnerReleaseAt, + releaseAbandonedNativeOwners, + resetAbandonedNativeOwnerReleaseForTests, + settleUnstartedNativeSessionResume, +} from "./native-failed-run-owner-release.js"; + +const LEASE_FAILURE = 'Failed to acquire lease for environment "Managed computer" (sandbox): Sandbox creation has failed'; + +describe("failed native run owner release", () => { + let temporary: Awaited> | null = null; + let db: ReturnType; + let companyId: string; + let agentId: string; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-owner-release-"); + db = createDb(temporary.connectionString); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + afterAll(async () => temporary?.cleanup()); + // The release deadline is instance-wide; keep each case's rows to itself. + afterEach(async () => { + resetAbandonedNativeOwnerReleaseForTests(); + await db.delete(environmentLeases); + await db.delete(nativeRunFinalizations); + }); + + beforeEach(async () => { + companyId = randomUUID(); + agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Owner release", issuePrefix: `OR${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Resume agent", adapterType: "codex_local", status: "running" }); + }); + + async function nativeRun(input: { + status: string; + errorCode?: string; + error?: string; + coordinator: Partial; + }) { + const issueId = randomUUID(); + const runId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Resume the native run", status: "in_progress", + assigneeAgentId: agentId, workMode: "standard" }); + await db.insert(heartbeatRuns).values({ + id: runId, companyId, agentId, nativeIssueId: issueId, status: input.status, runtimeMode: "native", + runtimeModeResolvedAt: new Date(), contextSnapshot: { issueId }, + startedAt: new Date(), finishedAt: input.status === "running" ? null : new Date(), + errorCode: input.errorCode ?? null, error: input.error ?? null, + runnerProfileJson: { + mode: "native", + nativeExecutionInput: { schema: "paperclip.native-execution-input.v1", binding: { runId } }, + sessionCheckpoint: { backendKind: "codex_app_server", sessionId: "persisted-session", identity: { runId }, + providerSessionId: "provider-session", activeTurnId: "active-turn" }, + }, + }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed", attempt: 1, + ...input.coordinator } as typeof nativeRunFinalizations.$inferInsert); + return { runId, issueId }; + } + + const coordinator = async (runId: string) => + (await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0]!; + + /** What the heartbeat setup-failure path writes; it never touches the coordinator. */ + async function failRunDuringSetup(runId: string, errorCode = "setup_failed", error = LEASE_FAILURE) { + await db.update(heartbeatRuns).set({ status: "failed", errorCode, error, finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, runId)); + } + + it("releases a session resume that failed at environment lease acquisition", async () => { + const { runId } = await nativeRun({ status: "failed", errorCode: "native_session_interrupted", + coordinator: { phase: "retryable_failure", nextAttemptAt: new Date(0), recoveryState: "resuming_session" } }); + const [claim] = await claimNativeSessionResumptions({ db, runnerInstanceId: "default", runIds: [runId] }); + expect(claim).toEqual({ runId, leaseOwner: expect.stringContaining("default:resume:"), attempt: 1 }); + expect(await coordinator(runId)).toMatchObject({ phase: "observed", leaseOwner: claim!.leaseOwner }); + await failRunDuringSetup(runId); + // This is the leaked state that kept the agent in `pausing`. + expect(await agentExecutionsHaveStopped(db, [agentId])).toBe(false); + + await expect(settleUnstartedNativeSessionResume(db, claim!)).resolves.toBe(true); + + expect(await coordinator(runId)).toMatchObject({ + phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null, recoveryState: "blocked", + failureCode: "setup_failed", resultId: null, + failureDetail: expect.objectContaining({ message: LEASE_FAILURE, originalFailureCode: "setup_failed", + releasedOwner: { reason: "unstarted_resume", leaseOwner: claim!.leaseOwner, attempt: 1 } }), + }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]) + .toMatchObject({ status: "failed", nativePhase: "terminal_failure" }); + expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, companyId), + eq(activityLog.action, "native.failed_run_owner_released")))).toHaveLength(1); + expect(await agentExecutionsHaveStopped(db, [agentId])).toBe(true); + // Settlement is idempotent. + await expect(settleUnstartedNativeSessionResume(db, claim!)).resolves.toBe(false); + }); + + it("leaves an owner the native executor claimed, and a run that is still running", async () => { + const { runId } = await nativeRun({ status: "failed", + coordinator: { phase: "retryable_failure", nextAttemptAt: new Date(0) } }); + const [claim] = await claimNativeSessionResumptions({ db, runnerInstanceId: "default", runIds: [runId] }); + // Still preparing: the run has not ended. + await expect(settleUnstartedNativeSessionResume(db, claim!)).resolves.toBe(false); + // A failed run whose recorded process is still alive keeps its owner. + await failRunDuringSetup(runId); + const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1_000)"], { stdio: "ignore" }); + await once(child, "spawn"); + try { + await db.update(heartbeatRuns).set({ processPid: child.pid! }).where(eq(heartbeatRuns.id, runId)); + await expect(settleUnstartedNativeSessionResume(db, claim!)).resolves.toBe(false); + } finally { + const exited = once(child, "exit"); + child.kill("SIGKILL"); + await exited; + } + await db.update(heartbeatRuns).set({ status: "running", processPid: null }).where(eq(heartbeatRuns.id, runId)); + // The executor's claim bumps the attempt; its own failure path owns settlement. + await db.update(nativeRunFinalizations).set({ attempt: 2 }).where(eq(nativeRunFinalizations.runId, runId)); + await failRunDuringSetup(runId, "adapter_failed", "provider turn failed"); + await expect(settleUnstartedNativeSessionResume(db, claim!)).resolves.toBe(false); + expect(await coordinator(runId)).toMatchObject({ phase: "observed", leaseOwner: claim!.leaseOwner, attempt: 2 }); + }); + + it("reaps an already leaked owner on a failed run so the agent can stop", async () => { + const leaseOwner = `default:resume:${randomUUID()}`; + const { runId } = await nativeRun({ status: "failed", errorCode: "adapter_failed", + error: "runner_state_identity_mismatch: session_scope_mismatch", + coordinator: { leaseOwner, leaseExpiresAt: new Date(Date.now() - 60_000), recoveryState: "resuming_session" } }); + expect(await agentExecutionsHaveStopped(db, [agentId])).toBe(false); + expect(await nextAbandonedNativeOwnerReleaseAt(db)).not.toBeNull(); + + const reconciled = await reconcileAbandonedExecutionControl(db); + expect(reconciled.releasedOwners).toBe(1); + + expect(await coordinator(runId)).toMatchObject({ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null, + recoveryState: "blocked", failureCode: "adapter_failed", + failureDetail: expect.objectContaining({ releasedOwner: { reason: "expired_owner", leaseOwner, attempt: 1 } }) }); + expect(await agentExecutionsHaveStopped(db, [agentId])).toBe(true); + // Nothing is left to schedule once the owner is settled. + expect(await nextAbandonedNativeOwnerReleaseAt(db)).toBeNull(); + }); + + it("keeps an unexpired owner blocking the stop check and schedules its expiry", async () => { + const leaseExpiresAt = new Date(Date.now() + 10 * 60_000); + const { runId } = await nativeRun({ status: "failed", errorCode: "setup_failed", error: LEASE_FAILURE, + coordinator: { leaseOwner: `default:resume:${randomUUID()}`, leaseExpiresAt } }); + + await expect(releaseAbandonedNativeOwners(db)).resolves.toBe(0); + expect(await coordinator(runId)).toMatchObject({ phase: "observed", leaseOwner: expect.any(String) }); + expect(await agentExecutionsHaveStopped(db, [agentId])).toBe(false); + expect((await nextAbandonedNativeOwnerReleaseAt(db))?.getTime()).toBe(leaseExpiresAt.getTime()); + expect(await executionControlSchedule(db).finalization()).toBe(leaseExpiresAt.getTime()); + }); + + it("does not reap while the run's environment lease or controller may still be live", async () => { + const leaseOwner = `default:resume:${randomUUID()}`; + const { runId, issueId } = await nativeRun({ status: "failed", errorCode: "adapter_failed", + coordinator: { leaseOwner, leaseExpiresAt: new Date(0) } }); + const [lease] = await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: runId, + provider: "test", providerLeaseId: `lease-${runId}`, status: "active" }).returning(); + await expect(releaseAbandonedNativeOwners(db)).resolves.toBe(0); + // An unreleased sandbox is outside this pass, so it is not polled for. + expect(await nextAbandonedNativeOwnerReleaseAt(db)).toBeNull(); + + await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success" }) + .where(eq(environmentLeases.id, lease!.id)); + const controller = spawn(process.execPath, ["-e", "setInterval(() => {}, 1_000)"], { stdio: "ignore" }); + await once(controller, "spawn"); + try { + await db.update(nativeRunFinalizations).set({ controllerPid: controller.pid!, controllerBootId: randomUUID() }) + .where(eq(nativeRunFinalizations.runId, runId)); + await expect(releaseAbandonedNativeOwners(db)).resolves.toBe(0); + expect(await coordinator(runId)).toMatchObject({ phase: "observed", leaseOwner }); + } finally { + const exited = once(controller, "exit"); + controller.kill("SIGKILL"); + await exited; + } + await expect(releaseAbandonedNativeOwners(db)).resolves.toBe(1); + expect(await coordinator(runId)).toMatchObject({ phase: "terminal_failure", leaseOwner: null }); + }); + + /** Records committed wake signals for one delivery queue while `work` runs. */ + async function committedWakes(topic: string, work: () => Promise) { + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, topic, event => { events.push(event); }); + try { await work(); } finally { unsubscribe(); } + return events.filter(event => event === "settled").length; + } + + it("wakes cost accounting when it settles an owner", async () => { + const { runId } = await nativeRun({ status: "failed", errorCode: "adapter_failed", + coordinator: { leaseOwner: `default:resume:${randomUUID()}`, leaseExpiresAt: new Date(0) } }); + await db.update(heartbeatRuns).set({ costAccountingPending: true }).where(eq(heartbeatRuns.id, runId)); + expect(await committedWakes(DELIVERY_QUEUES.accounting, () => releaseAbandonedNativeOwners(db))).toBeGreaterThan(0); + expect(await coordinator(runId)).toMatchObject({ phase: "terminal_failure", leaseOwner: null }); + }); + + it("wakes the finalization queue when the run's environment lease settles", async () => { + const leaseOwner = `default:resume:${randomUUID()}`; + const { runId, issueId } = await nativeRun({ status: "failed", errorCode: "adapter_failed", + coordinator: { leaseOwner, leaseExpiresAt: new Date(0) } }); + const [lease] = await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: runId, + provider: "test", providerLeaseId: `lease-${runId}`, status: "active" }).returning(); + // Blocked and unscheduled while the sandbox lease is unsettled. + expect(await nextAbandonedNativeOwnerReleaseAt(db)).toBeNull(); + expect(await committedWakes(DELIVERY_QUEUES.executionFinalization, + () => environmentService(db).releaseLease(lease!.id, "released", { cleanupStatus: "success" }))).toBeGreaterThan(0); + expect(await nextAbandonedNativeOwnerReleaseAt(db)).not.toBeNull(); + await expect(releaseAbandonedNativeOwners(db)).resolves.toBe(1); + expect(await agentExecutionsHaveStopped(db, [agentId])).toBe(true); + }); + + it("ignores a recorded process and group whose PID now belongs to another process", async () => { + const { runId } = await nativeRun({ status: "failed", errorCode: "adapter_failed", + coordinator: { leaseOwner: `default:resume:${randomUUID()}`, leaseExpiresAt: new Date(0) } }); + // A detached child leads its own group, as a local native runner does. + const unrelated = spawn(process.execPath, ["-e", "setInterval(() => {}, 1_000)"], { stdio: "ignore", detached: true }); + await once(unrelated, "spawn"); + try { + await db.update(heartbeatRuns).set({ processPid: unrelated.pid!, processGroupId: unrelated.pid!, + processStartedAt: new Date("2020-01-01T00:00:00.000Z") }).where(eq(heartbeatRuns.id, runId)); + await expect(releaseAbandonedNativeOwners(db)).resolves.toBe(1); + expect(await coordinator(runId)).toMatchObject({ phase: "terminal_failure", leaseOwner: null }); + } finally { + const exited = once(unrelated, "exit"); + unrelated.kill("SIGKILL"); + await exited; + } + }); + + it("rotates bounded passes past owners that a live controller still holds", async () => { + const runs = await Promise.all([0, 1, 2].map(() => nativeRun({ status: "failed", errorCode: "adapter_failed", + coordinator: { leaseOwner: `default:resume:${randomUUID()}`, leaseExpiresAt: new Date(0) } }))); + const [held, ...free] = [...runs].sort((left, right) => left.runId.localeCompare(right.runId)); + const controller = spawn(process.execPath, ["-e", "setInterval(() => {}, 1_000)"], { stdio: "ignore" }); + await once(controller, "spawn"); + try { + // The first page holds only the row a live controller still owns. + await db.update(nativeRunFinalizations).set({ controllerPid: controller.pid!, controllerBootId: randomUUID() }) + .where(eq(nativeRunFinalizations.runId, held!.runId)); + // Each pass handles one bounded page; the cursor moves past the held row. + const passes = []; + for (let pass = 0; pass < 3; pass += 1) passes.push(await releaseAbandonedNativeOwners(db, new Date(), 1)); + expect(passes).toEqual([0, 1, 1]); + // After a full rotation the held row is inspected again, not skipped forever. + await expect(releaseAbandonedNativeOwners(db, new Date(), 1)).resolves.toBe(0); + expect(await coordinator(held!.runId)).toMatchObject({ phase: "observed", leaseOwner: expect.any(String) }); + for (const { runId } of free) expect(await coordinator(runId)).toMatchObject({ phase: "terminal_failure", leaseOwner: null }); + } finally { + const exited = once(controller, "exit"); + controller.kill("SIGKILL"); + await exited; + } + }); + + it("leaves cancelled runs and pending Stops to their own stop-proof paths", async () => { + const { runId: cancelledRunId } = await nativeRun({ status: "cancelled", + coordinator: { leaseOwner: `default:resume:${randomUUID()}`, leaseExpiresAt: new Date(0) } }); + const { runId: stoppingRunId } = await nativeRun({ status: "interrupted", + coordinator: { leaseOwner: `default:resume:${randomUUID()}`, leaseExpiresAt: new Date(0) } }); + await db.update(heartbeatRuns).set({ resultJson: { nativeCancellation: { dispatchState: "pending" } } }) + .where(eq(heartbeatRuns.id, stoppingRunId)); + await expect(releaseAbandonedNativeOwners(db)).resolves.toBe(0); + expect(await nextAbandonedNativeOwnerReleaseAt(db)).toBeNull(); + for (const runId of [cancelledRunId, stoppingRunId]) { + expect(await coordinator(runId)).toMatchObject({ phase: "observed", leaseOwner: expect.any(String) }); + } + }); +}); diff --git a/server/src/services/native-runtime/native-failed-run-owner-release.ts b/server/src/services/native-runtime/native-failed-run-owner-release.ts new file mode 100644 index 00000000000..1db97b0a861 --- /dev/null +++ b/server/src/services/native-runtime/native-failed-run-owner-release.ts @@ -0,0 +1,271 @@ +import { and, asc, eq, gt, inArray, isNotNull, isNull, lte, or, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { logger } from "../../middleware/logger.js"; +import { adapterExecutionControls } from "../adapter-execution-control.js"; +import { logActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; +import { notifyExecutionRunWork } from "../execution-control-notifications.js"; +import { readProcessStartedAt } from "../hot-restart.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; +import { LEGACY_WORKSPACE_RECOVERY_KEY } from "../workspace-restore-recovery-state.js"; +import { activeNativeExecutionRunIds } from "./native-active-runs.js"; +import { isCurrentNativeControllerProcess } from "./native-restart-recovery.js"; + +/** + * A native run can fail outside the native session executor: environment lease + * acquisition, runner state verification, or any other step before the + * executor takes its own provider claim. The heartbeat failure path then marks + * the run failed but never sees the finalization row, so the claim that + * dispatched the run keeps `observed` ownership. That owner blocks agent stop + * checks and keeps the task looking like it still has a live recovery path. + * + * This module settles those rows to `terminal_failure`, the state the executor + * itself writes for a failure it observes. Run status alone is never enough: + * each path below also proves that no provider work can still be running under + * the owner it releases. + */ + +const RELEASABLE_RUN_STATUSES = ["failed", "timed_out", "interrupted"]; +const FALLBACK_FAILURE_CODE = "native_session_resume_failed"; +const NEXT_ACTION = "The run ended before it could resume its native provider session. Inspect the run failure, then explicitly retry the task."; +const ACTOR_ID = "native-finalization-owner-release"; + +type Run = typeof heartbeatRuns.$inferSelect; +type Coordinator = typeof nativeRunFinalizations.$inferSelect; +type Release = + /** The caller dispatched this claim and its execution has returned. */ + | { kind: "unstarted_resume"; runId: string; leaseOwner: string; attempt: number } + /** A background pass found an expired owner and must prove it stopped. */ + | { kind: "expired_owner"; runId: string; leaseOwner: string; attempt: number; controllerGeneration: number }; + +/** Environment leases that `agentExecutionsHaveStopped` would still treat as live. */ +const unsettledRunEnvironmentLease = sql`exists ( + select 1 from ${environmentLeases} lease + where lease.company_id = ${heartbeatRuns.companyId} + and lease.heartbeat_run_id = ${heartbeatRuns.id} + and ((lease.released_at is null and not (lease.status = 'retained' and lease.cleanup_status = 'success')) + or lease.status = 'pending_cleanup' or lease.cleanup_status = 'failed') +)`; + +/** A pending Stop belongs to native restart recovery's stop cleanup. */ +const pendingNativeStop = sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) #>> '{nativeCancellation,dispatchState}' = 'pending'`; + +/** + * Rows the background pass may release once their lease expires. A pinned + * workspace restore belongs to observed-workspace preservation, which must stop + * and retain the sandbox before it settles the same row. + */ +function abandonedOwnerCondition() { + return and( + eq(heartbeatRuns.runtimeMode, "native"), + inArray(heartbeatRuns.status, RELEASABLE_RUN_STATUSES), + eq(nativeRunFinalizations.phase, "observed"), + isNull(nativeRunFinalizations.resultId), + isNotNull(nativeRunFinalizations.leaseOwner), + isNotNull(nativeRunFinalizations.leaseExpiresAt), + sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) -> ${LEGACY_WORKSPACE_RECOVERY_KEY}::text is null`, + sql`not coalesce(${pendingNativeStop}, false)`, + sql`not ${unsettledRunEnvironmentLease}`, + )!; +} + +/** + * The earliest lease expiry the background pass will need to inspect. Sandbox + * cleanup is excluded, so an unreleased lease is never polled for here. An + * expired row whose process or controller may still be live keeps the shared + * finalization retry cadence until that process exits. + */ +export async function nextAbandonedNativeOwnerReleaseAt(db: Db): Promise { + const [row] = await db.select({ at: nativeRunFinalizations.leaseExpiresAt }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId))) + .where(abandonedOwnerCondition()) + .orderBy(asc(nativeRunFinalizations.leaseExpiresAt)).limit(1); + return row?.at ?? null; +} + +function processAbsent(pid: number): boolean { + try { process.kill(pid, 0); return false; } + catch (error) { return (error as NodeJS.ErrnoException).code === "ESRCH"; } +} + +/** True unless the recorded process is gone or provably a different process. */ +async function recordedProcessMayBeLive(pid: number | null, startedAt: Date | null): Promise { + if (!pid || processAbsent(pid)) return false; + if (!startedAt) return true; + const observed = await readProcessStartedAt(pid).catch(() => null); + return !observed || new Date(observed).getTime() === startedAt.getTime(); +} + +/** + * The run-level process evidence `agentExecutionsHaveStopped` requires. This + * process answers for itself from its in-memory execution registries. + */ +async function runProcessMayBeLive(run: Run, now: Date): Promise { + if (adapterExecutionControls.has(run.id) || activeNativeExecutionRunIds.has(run.id)) return true; + if (run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > now) return true; + // Same process identity rule as `agentExecutionsHaveStopped`: a recorded PID + // that now has another start time was recycled, and so was its own group. + let processAlive = Boolean(run.processPid && !processAbsent(run.processPid)); + let groupAlive = Boolean(run.processGroupId && !processAbsent(-run.processGroupId)); + if (processAlive && run.processStartedAt) { + const observed = await readProcessStartedAt(run.processPid!).catch(() => null); + if (observed && new Date(observed).getTime() !== run.processStartedAt.getTime()) { + processAlive = false; + if (run.processGroupId === run.processPid) groupAlive = false; + } + } + return processAlive || groupAlive; +} + +/** An expired owner's recorded controller, unless this process is that controller. */ +async function controllerMayBeLive(coordinator: Coordinator): Promise { + if (isCurrentNativeControllerProcess(coordinator)) return false; + // A previous boot of this process can share its PID, but cannot be its controller. + if (coordinator.controllerPid === process.pid) return false; + return recordedProcessMayBeLive(coordinator.controllerPid, coordinator.controllerProcessStartedAt); +} + +async function settleFailedRunOwner(db: Db, release: Release, now: Date): Promise { + const publications: ActivityPublication[] = []; + const settled = await db.transaction(async transaction => { + const tx = transaction as unknown as Db; + await tx.execute(sql`select set_config('statement_timeout', '15000', true), set_config('lock_timeout', '1000', true)`); + const [binding] = await tx.select({ companyId: nativeRunFinalizations.companyId, issueId: nativeRunFinalizations.issueId }) + .from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, release.runId)); + if (!binding) return false; + // Same issue-before-coordinator-before-run order as admission and failure settlement. + await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, binding.companyId), + eq(issues.id, binding.issueId))).for("update"); + const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, binding.companyId), eq(nativeRunFinalizations.runId, release.runId), + eq(nativeRunFinalizations.issueId, binding.issueId))).for("update"); + const [run] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, binding.companyId), eq(heartbeatRuns.id, release.runId))).for("update"); + if (!coordinator || !run || run.runtimeMode !== "native" || !RELEASABLE_RUN_STATUSES.includes(run.status) + || coordinator.phase !== "observed" || coordinator.resultId !== null + || coordinator.leaseOwner !== release.leaseOwner || coordinator.attempt !== release.attempt) return false; + if (release.kind === "expired_owner") { + if (!coordinator.leaseExpiresAt || coordinator.leaseExpiresAt > now + || coordinator.controllerGeneration !== release.controllerGeneration + || run.resultJson?.[LEGACY_WORKSPACE_RECOVERY_KEY] != null) return false; + const [heldElsewhere] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, run.id), or(unsettledRunEnvironmentLease, pendingNativeStop))).limit(1); + if (heldElsewhere || await runProcessMayBeLive(run, now) || await controllerMayBeLive(coordinator)) return false; + } else if (await runProcessMayBeLive(run, now)) { + // The claim never reached the executor, but the run may still own a local process. + return false; + } + const [accepted] = await tx.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( + eq(nativeRunResults.companyId, run.companyId), eq(nativeRunResults.runId, run.id), + eq(nativeRunResults.schemaStatus, "accepted"))).limit(1); + if (accepted) return false; + + const failureCode = coordinator.failureCode ?? run.errorCode ?? FALLBACK_FAILURE_CODE; + const scope = { companyId: run.companyId, agentId: run.agentId, runId: run.id, issueId: coordinator.issueId }; + await notifyExecutionRunWork(tx, scope); + await notifyNativeFinalizationWork(tx, scope); + // A result-less `observed` owner holds pending cost accounting; settlement releases it. + await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); + const [updated] = await tx.update(nativeRunFinalizations).set({ + phase: "terminal_failure", + leaseOwner: null, + leaseExpiresAt: null, + controlDeadlineAt: null, + nextAttemptAt: null, + recoveryState: "blocked", + failureCode, + failureDetail: { + ...coordinator.failureDetail, + ...(run.error ? { message: run.error.slice(0, 2_000) } : {}), + originalFailureCode: run.errorCode ?? failureCode, + releasedOwner: { reason: release.kind, leaseOwner: release.leaseOwner, attempt: release.attempt }, + recoveryOwner: { kind: "board" }, + nextAction: NEXT_ACTION, + }, + updatedAt: now, + }).where(and( + eq(nativeRunFinalizations.runId, run.id), + eq(nativeRunFinalizations.phase, "observed"), + isNull(nativeRunFinalizations.resultId), + eq(nativeRunFinalizations.leaseOwner, release.leaseOwner), + eq(nativeRunFinalizations.attempt, release.attempt), + )).returning({ runId: nativeRunFinalizations.runId }); + if (!updated) return false; + await tx.update(heartbeatRuns).set({ nativePhase: "terminal_failure", nativePhaseUpdatedAt: now, updatedAt: now }) + .where(eq(heartbeatRuns.id, run.id)); + await logActivity(tx, { + companyId: run.companyId, actorType: "system", actorId: ACTOR_ID, + action: "native.failed_run_owner_released", entityType: "heartbeat_run", entityId: run.id, + agentId: run.agentId, runId: run.id, issueId: coordinator.issueId, + details: { reason: release.kind, runStatus: run.status, failureCode, attempt: release.attempt, + controllerGeneration: coordinator.controllerGeneration }, + }, publications); + return true; + }); + for (const publication of publications) publishActivity(publication); + return settled; +} + +/** + * Settles a session-resume claim whose dispatched execution returned without + * the native executor ever taking its own provider claim. The executor bumps + * the coordinator attempt when it claims, so an unchanged attempt proves no + * provider turn started under this owner. Failures that reach the executor's + * claim are settled by the executor and no longer match. + */ +export async function settleUnstartedNativeSessionResume(db: Db, claim: { + runId: string; leaseOwner: string; attempt: number; +}): Promise { + return settleFailedRunOwner(db, { kind: "unstarted_resume", ...claim }, new Date()).catch((error: unknown) => { + logger.warn({ err: error, runId: claim.runId }, + "Native session resume owner release failed; the abandoned owner pass will release it after lease expiry"); + return false; + }); +} + +let abandonedOwnerReleaseCursor: string | null = null; + +export function resetAbandonedNativeOwnerReleaseForTests() { + abandonedOwnerReleaseCursor = null; +} + +/** + * Releases expired, result-less owners left on native runs that already ended + * failed, timed out, or interrupted. Each release re-proves under row locks that + * the run's environment leases are settled and that neither its process nor its + * controller may still be running. Cancelled runs keep their own stop-proof path + * in `agentExecutionsHaveStopped`. + */ +export async function releaseAbandonedNativeOwners(db: Db, now = new Date(), pageSize = 50): Promise { + // One bounded page per pass. The cursor rotates across passes, so rows that a + // live process still holds cannot starve stopped owners behind them; any + // remaining expired row keeps the finalization deadline due for the next pass. + const page = (after: string | null) => db.select({ + runId: nativeRunFinalizations.runId, + leaseOwner: nativeRunFinalizations.leaseOwner, + attempt: nativeRunFinalizations.attempt, + controllerGeneration: nativeRunFinalizations.controllerGeneration, + }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId))) + .where(and(abandonedOwnerCondition(), lte(nativeRunFinalizations.leaseExpiresAt, now), + after ? gt(nativeRunFinalizations.runId, after) : undefined)) + .orderBy(asc(nativeRunFinalizations.runId)) + .limit(pageSize); + let candidates = await page(abandonedOwnerReleaseCursor); + if (candidates.length === 0 && abandonedOwnerReleaseCursor !== null) candidates = await page(null); + abandonedOwnerReleaseCursor = candidates.length === pageSize ? candidates[candidates.length - 1]!.runId : null; + let released = 0; + for (const candidate of candidates) { + try { + if (await settleFailedRunOwner(db, { kind: "expired_owner", runId: candidate.runId, + leaseOwner: candidate.leaseOwner!, attempt: candidate.attempt, + controllerGeneration: candidate.controllerGeneration }, now)) released += 1; + } catch (error) { + logger.warn({ err: error, runId: candidate.runId }, + "Abandoned native owner release remains pending; continuing with other runs"); + } + } + return released; +} diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 493551935cc..ba43627916a 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -65,7 +65,8 @@ function record(value: unknown): Record { : {}; } -export type NativeSessionResumeClaim = { runId: string; leaseOwner: string }; +/** `attempt` is the coordinator attempt the claim saw; the executor bumps it when it takes provider ownership. */ +export type NativeSessionResumeClaim = { runId: string; leaseOwner: string; attempt: number }; export { reconcileRetainedNativeSessionCleanups } from "./retained-native-session-cleanup.js"; @@ -138,6 +139,7 @@ export async function claimNativeSessionResumptions(input: { const leaseOwner = `${input.runnerInstanceId}:resume:${randomUUID()}`; let terminalRunToEmit: typeof heartbeatRuns.$inferSelect | null = null; let claimFailed = false; + let claimedAttempt = 0; const claimed = await input.db.transaction(async (tx) => { await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`); const [issue] = await tx.select().from(issues).where(and( @@ -306,6 +308,7 @@ export async function claimNativeSessionResumptions(input: { eq(nativeRunFinalizations.phase, row.coordinator.phase), )).returning({ runId: nativeRunFinalizations.runId }).then((rows) => rows[0] ?? null); if (!updated) return false; + claimedAttempt = row.coordinator.attempt; await tx.update(heartbeatRuns).set({ status: "running", finishedAt: null, @@ -331,7 +334,7 @@ export async function claimNativeSessionResumptions(input: { void emitAgentTaskRun(input.db, terminalRunToEmit); void reportRunFailure(input.db, terminalRunToEmit); } - if (claimed) claims.push({ runId: candidate.runId, leaseOwner }); + if (claimed) claims.push({ runId: candidate.runId, leaseOwner, attempt: claimedAttempt }); else if (claimFailed || input.deferIneligible !== false) await input.db.transaction(async tx => { await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`); await notifyLiveRunRecovery(tx, { companyId: candidate.companyId, runId: candidate.runId });