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 });