diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index fcf339efa6..f77b10f3b0 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1478,6 +1478,30 @@ differed, for example `lease_status`, `adapter_type`, `provider_lease_id`, `remo `identity_run_id`. The detail contains no identifiers. Consumers match the code exactly or as the `code: ` prefix. +A continuation run starts with its predecessor's saved provider session. For a +warm session it also carries the predecessor's runner process identity. When the +run fails before it asks for a runner of its own (for example while it stages +its provider pack), a due same-run retry does not reattach: there is no runner +of this run to reattach to. The server retires the inherited process identity +and uses the ordinary same-run resume, which is the path the failed attempt +took. The resume prepares the run again in its own retained sandbox when that +sandbox lease is still active, and acquires one as an ordinary resume does when +it is gone. The server does this only when all of these are true: + +- the failed attempt is a result-less exact-checkpoint retry that recorded + `providerEventsExist: false`, and no controller owns it; +- the run has no runner start, process identity, adapter invocation, tool start, + runner event, controller checkpoint, provider event or provider session of + its own; +- the saved session has no active turn and no pending runtime request; and +- a settled predecessor of the same native session recorded the exact same + process identity. + +It records an `inherited_runner_never_launched` lifecycle event. Any other run, +and any run saved by an older release without these facts, keeps verified +reattachment. A reattachment that loses its runner still requests the user's +Retry as described above. + When a native attempt fails before launch because its predecessor's session is quarantined, explicit task Retry verifies the retained journal's actual owner. That owner must belong to the same company, task, agent, workspace, native session, diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index ddc034f725..b00ed1c709 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -369,6 +369,88 @@ describePostgres("heartbeat recovery module ownership", () => { } finally { kill.mockRestore(); try { await f.cleanup(); } finally { await child.stop(); } } }); + // Staging shape: a chat continuation inherited its predecessor's saved + // provider session and warm runner identity, then failed while staging its + // provider pack, before it ever asked for a runner of its own. Its workspace + // reference is saved and its own sandbox lease is still active. + type InheritedCase = "staging" | "own_turn" | "own_launch" | "own_checkpoint" | "no_predecessor" + | "pending_side_effect" | "legacy_detail"; + async function inheritedRunnerRetry(pid: number, condition: InheritedCase) { + const f = await remoteRetryFixture(pid, "remote_workspace"); + const run = await f.current(); + // The PID came from the predecessor, so this run recorded no process identity. + await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id)); + if (condition !== "own_checkpoint") await db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.run.id)); + const profile = run.runnerProfileJson ?? {}; + const checkpoint = profile.sessionCheckpoint as Record; + await db.update(agents).set({ status: "active" }).where(eq(agents.id, f.run.agentId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeExecutionInput: { fixture: true }, + remoteWorkspace: { ...(profile.remoteWorkspace as Record), state: "saved" }, + sessionCheckpoint: { ...checkpoint, activeTurnId: null, + pendingRuntimeRequests: condition === "pending_side_effect" ? [{ requestId: "tool-1", kind: "tool_call" }] : [] }, + } }).where(eq(heartbeatRuns.id, f.run.id)); + await db.update(environmentLeases).set({ executionWorkspaceId: null }).where(eq(environmentLeases.id, f.lease.id)); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: null, failureDetail: { + message: "runner_remote_provider_archive_invalid", recoveryMode: "exact_checkpoint_resume", checkpointExists: true, + providerSessionEstablished: true, ...(condition === "legacy_detail" ? {} : { providerEventsExist: false }), + } }).where(eq(nativeRunFinalizations.runId, f.run.id)); + const predecessorId = randomUUID(); + if (condition !== "no_predecessor") await db.insert(heartbeatRuns).values({ id: predecessorId, + companyId: f.run.companyId, agentId: f.run.agentId, issueId: f.run.issueId, invocationSource: "automation", + status: "succeeded", runtimeMode: "native", nativeSessionId: run.nativeSessionId, createdAt: new Date(Date.now() - 600_000), + finishedAt: new Date(Date.now() - 300_000), processPid: run.processPid, processGroupId: run.processGroupId, + processStartedAt: run.processStartedAt }); + if (condition === "own_turn") await appendHeartbeatRunEvent(db, { companyId: f.run.companyId, runId: f.run.id, + agentId: f.run.agentId, eventType: "turn.started", stream: "system", level: "info", message: "turn.started" }); + if (condition === "own_launch") await appendHeartbeatRunEvent(db, { companyId: f.run.companyId, runId: f.run.id, + agentId: f.run.agentId, eventType: PROCESS_START_REQUESTED, stream: "system", level: "info" }); + return { ...f, predecessorId, async cleanupAll() { + try { await f.cleanup(); } finally { await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, predecessorId)); } + } }; + } + + it("resumes a continuation that never launched its own runner in its retained sandbox, without reattaching", async () => { + const child = await ownedProcess(); + const f = await inheritedRunnerRetry(child.pid, "staging"); + try { + native.claims.mockClear(); + await createHeartbeatRecovery(db, f.deps).resumeRetryableNativeProcessSessions({ runId: f.run.id }); + await Promise.all([...f.deps.activeRunExecutionPromises]); + expect(native.claims.mock.calls.some(([input]) => input?.runIds?.includes(f.run.id))).toBe(false); + expect(f.deps.executeRun).toHaveBeenCalledExactlyOnceWith(f.run.id, { nativeLeaseOwner: expect.any(String) }); + const current = await f.current(); + expect(current).toMatchObject({ processPid: null, processGroupId: null, processStartedAt: null }); + expect(current.runnerProfileJson?.nativeInheritedRunnerRetirement).toEqual({ + schema: "paperclip.native-inherited-runner-retirement.v1", predecessorRunId: f.predecessorId, + retainedSandbox: { leaseId: f.lease.id, providerLeaseId: f.lease.providerLeaseId, remoteCwd: "/work" } }); + const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id)); + expect(events.find(event => event.payload?.reason === "inherited_runner_never_launched")?.payload) + .toMatchObject({ predecessorRunId: f.predecessorId, retainedSandbox: true }); + // The retained sandbox lease stays the run's only lease; nothing new was acquired. + expect(await db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, f.run.id))) + .toMatchObject([{ id: f.lease.id, status: "active" }]); + expect(f.deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled(); + expect(child.alive()).toBe(true); + } finally { try { await f.cleanupAll(); } finally { await child.stop(); } } + }); + + it.each(["own_turn", "own_launch", "own_checkpoint", "no_predecessor", "pending_side_effect", "legacy_detail"] as const)( + "keeps verified reattachment when the run's own runner may hold work or the proof is incomplete: %s", async condition => { + const child = await ownedProcess(); + const f = await inheritedRunnerRetry(child.pid, condition); + try { + native.claims.mockClear(); + await createHeartbeatRecovery(db, f.deps).resumeRetryableNativeProcessSessions({ runId: f.run.id }); + await Promise.all([...f.deps.activeRunExecutionPromises]); + expect(native.claims.mock.calls.some(([input]) => input?.runIds?.includes(f.run.id))).toBe(true); + expect(f.deps.executeRun).not.toHaveBeenCalledWith(f.run.id, { nativeLeaseOwner: expect.any(String) }); + const current = await f.current(); + expect(current.processPid).toBe(child.pid); + expect(current.runnerProfileJson).not.toHaveProperty("nativeInheritedRunnerRetirement"); + } finally { try { await f.cleanupAll(); } finally { await child.stop(); } } + }, + ); + it("routes a due plugin-task retry through its existing remote lease", async () => { const child = await ownedProcess(); const f = await remoteRetryFixture(child.pid, "plugin_task"); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index f52151df61..14d5562045 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -28,6 +28,7 @@ import { } from "../sandbox-stop-and-retain.js"; import { discoverSettledNativeRunProcesses, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, recordNativeLocalProcessStopInTransaction, releaseSettledNativeRunProcess, retryPendingSettledNativeRunProcesses } from "../native-local-process-stop.js"; import { pendingCaseHandoffForRun } from "./case-handoff-lifecycle.js"; +import { retireInheritedRemoteRunnerIdentity } from "../native-runtime/native-inherited-runner-retry.js"; import { legacyControllerBootId, liveLegacyControllerExpiresAt, @@ -2202,7 +2203,10 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD if (activeRunExecutions.has(run.id) || runningProcesses.has(run.id)) continue; const profile = parseObject(run.runnerProfileJson); if (profile.nativeWorkspaceSync != null || profile.remoteWorkspace != null) { - remoteRetryRunIds.push(run.id); + // A continuation that never started its own runner holds only its + // predecessor's identity: resume it the way its failed attempt ran. + if (await retireInheritedRemoteRunnerIdentity(db, run.id)) claimableNativeRunIds.add(run.id); + else remoteRetryRunIds.push(run.id); } else { const [taskLease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), diff --git a/server/src/services/heartbeat/workspace-preparation.test.ts b/server/src/services/heartbeat/workspace-preparation.test.ts index 1edc7bfe73..b01509a584 100644 --- a/server/src/services/heartbeat/workspace-preparation.test.ts +++ b/server/src/services/heartbeat/workspace-preparation.test.ts @@ -2065,6 +2065,51 @@ describe.skipIf(!support.supported && !externalDatabaseUrl)("heartbeat workspace expect(input.nativeRunnerPreparationSpans).toMatchObject([{ name: "environment.acquire", outcome: "failed" }]); }); + // A same-run resume whose inherited runner identity was retired keeps the + // run's own retained sandbox. When that sandbox is gone, it acquires as an + // ordinary resume does. + function retiredInheritedRunner(input: Awaited>["input"]) { + const location = { leaseId: randomUUID(), providerLeaseId: "retained-sandbox", remoteCwd: "/retained" }; + input.run.runnerProfileJson = { ...(input.run.runnerProfileJson ?? {}), + remoteWorkspace: { schema: "paperclip.remote-workspace-run.v1", workspaceId: randomUUID(), epoch: 2, + state: "saved", snapshotId: randomUUID(), ...location }, + nativeInheritedRunnerRetirement: { schema: "paperclip.native-inherited-runner-retirement.v1", + predecessorRunId: randomUUID(), retainedSandbox: location } }; + input.reuse.runOptions = { nativeLeaseOwner: "resume-owner" }; + return location; + } + + it("resumes a retired inherited runner in its own retained sandbox", async () => { + const { input, envOrchestrator } = await fixture(); + const location = retiredInheritedRunner(input); + const reached = new Error("Reached the retained sandbox"); + const acquire = vi.spyOn(envOrchestrator, "acquireForRun").mockRejectedValue(reached); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(reached); + expect(acquire).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ reattachRemoteLease: location })); + }); + + it("acquires as an ordinary resume when the retained sandbox is gone", async () => { + const { input, envOrchestrator } = await fixture(); + retiredInheritedRunner(input); + const reached = new Error("Reached ordinary acquisition"); + const acquire = vi.spyOn(envOrchestrator, "acquireForRun") + .mockRejectedValueOnce(new Error("native_remote_recovery_lease_mismatch: stage=environment_acquire field=lease_status")) + .mockRejectedValueOnce(reached); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(reached); + expect(acquire).toHaveBeenCalledTimes(2); + expect(acquire.mock.calls[1]![0]).not.toHaveProperty("reattachRemoteLease"); + }); + + it("does not reuse a retained sandbox without a retired inherited runner", async () => { + const { input, envOrchestrator } = await fixture(); + retiredInheritedRunner(input); + delete (input.run.runnerProfileJson as Record).nativeInheritedRunnerRetirement; + const reached = new Error("Reached ordinary acquisition"); + const acquire = vi.spyOn(envOrchestrator, "acquireForRun").mockRejectedValue(reached); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(reached); + expect(acquire.mock.calls[0]![0]).not.toHaveProperty("reattachRemoteLease"); + }); + it("binds a replacement workspace from realization and snapshots the restore lease", async () => { const { input, envOrchestrator, issue } = await fixture(); const realize = envOrchestrator.realizeForRun; diff --git a/server/src/services/heartbeat/workspace-preparation.ts b/server/src/services/heartbeat/workspace-preparation.ts index 7ff85efacf..248740c28a 100644 --- a/server/src/services/heartbeat/workspace-preparation.ts +++ b/server/src/services/heartbeat/workspace-preparation.ts @@ -1,4 +1,5 @@ -import { remoteRecoveryLeaseMismatch } from "../native-runtime/remote-recovery-lease-mismatch.js"; +import { NATIVE_REMOTE_RECOVERY_LEASE_MISMATCH, remoteRecoveryLeaseMismatch } from "../native-runtime/remote-recovery-lease-mismatch.js"; +import { readInheritedRunnerRetainedSandbox } from "../native-runtime/native-inherited-runner-retry.js"; import { resolveProviderApiPlacement } from "../native-runtime/provider-api-placement.js"; import { assertEphemeralWorkspaceExecution, readRemoteWorkspaceAuthority, RemoteWorkspaceWriterActiveError, reserveRemoteWorkspace, shouldUseRemoteWorkspaces } from "../remote-workspaces.js"; import { WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, WorkspaceBusyDeferral } from "./retries.js"; @@ -1306,7 +1307,22 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace : saved.remoteCwd !== sandboxRecovery.remoteCwd ? "remote_cwd" as const : null; if (mismatch) throw remoteRecoveryLeaseMismatch("workspace_preparation", mismatch); } - acquiredEnvironment = await timeWorkspace("workspace_environment_acquire", () => envOrchestrator.acquireForRun({ + // A same-run resume whose inherited runner identity was retired keeps the + // run's own retained sandbox, so its workspace and allocation carry over. + const retainedSandbox = runOptions.nativeLeaseOwner && !runOptions.nativeRestartRecovery + ? readInheritedRunnerRetainedSandbox(parseObject(run.runnerProfileJson)) : null; + const acquireRetainedOrNew = async (input: Parameters[0]) => { + if (!retainedSandbox) return envOrchestrator.acquireForRun(input); + try { + return await envOrchestrator.acquireForRun({ ...input, reattachRemoteLease: retainedSandbox }); + } catch (error) { + // The retained sandbox is gone. Acquire as an ordinary resume does. + if (!(error instanceof Error) || !error.message.startsWith(`${NATIVE_REMOTE_RECOVERY_LEASE_MISMATCH}: `)) throw error; + logger.warn({ runId: run.id, detail: error.message }, "retained sandbox unavailable for same-run resume; acquiring"); + return envOrchestrator.acquireForRun(input); + } + }; + acquiredEnvironment = await timeWorkspace("workspace_environment_acquire", () => acquireRetainedOrNew({ companyId: agent.companyId, selectedEnvironmentId: selectedEnvironmentId ?? localEnvironment.id, localEnvironmentId: localEnvironment.id, diff --git a/server/src/services/native-runtime/native-inherited-runner-retry.ts b/server/src/services/native-runtime/native-inherited-runner-retry.ts new file mode 100644 index 0000000000..3dd5384a7f --- /dev/null +++ b/server/src/services/native-runtime/native-inherited-runner-retry.ts @@ -0,0 +1,150 @@ +import { and, desc, eq, inArray, isNull, lt, ne, or, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; +import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; +import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; +import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; +import { hasNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; +import { nativeSessionProviderEvidence } from "./native-session-resume.js"; +import { hasNativeExecutionStartEvidence } from "./native-restart-recovery.js"; +import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; +import { logger } from "../../middleware/logger.js"; + +/** + * A continuation run starts with its predecessor's saved provider session and, + * for a warm session, the predecessor's runner process identity. When such a + * run fails before it ever asks for its own runner (for example while staging + * its provider pack), the recorded process identity is the predecessor's, not + * its own. A due same-run retry used to reattach to "this run's" runner, which + * never existed, and always failed with `native_remote_recovery_lease_mismatch`. + * + * This module proves that shape from durable state and retires the inherited + * identity, so the retry takes the ordinary same-run resume: the same path the + * failed attempt took. It also records the run's own retained sandbox lease, + * so preparation resumes that exact sandbox instead of acquiring another one. + */ +export const NATIVE_INHERITED_RUNNER_RETIREMENT_KEY = "nativeInheritedRunnerRetirement"; +const RETIREMENT_SCHEMA = "paperclip.native-inherited-runner-retirement.v1"; + +const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]; + +export type InheritedRunnerRetainedSandbox = { leaseId: string; providerLeaseId: string; remoteCwd: string }; + +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + +function sameTime(left: Date | null, right: Date | null) { + return (left?.getTime() ?? null) === (right?.getTime() ?? null); +} + +/** The run's saved remote location: a native workspace sync or a remote workspace reference. */ +function savedRemoteLocation(profile: Record): InheritedRunnerRetainedSandbox | null { + const sync = readNativeWorkspaceSyncReference(profile.nativeWorkspaceSync); + if (sync) return { leaseId: sync.leaseId, providerLeaseId: sync.providerLeaseId, remoteCwd: sync.remoteCwd }; + const remote = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace); + return remote.success + ? { leaseId: remote.data.leaseId, providerLeaseId: remote.data.providerLeaseId, remoteCwd: remote.data.remoteCwd } + : null; +} + +/** The retained sandbox recorded when the inherited identity was retired, if it still matches the saved location. */ +export function readInheritedRunnerRetainedSandbox(profile: Record | null | undefined): InheritedRunnerRetainedSandbox | null { + const marker = record(profile?.[NATIVE_INHERITED_RUNNER_RETIREMENT_KEY]); + const sandbox = record(marker.retainedSandbox); + const location = savedRemoteLocation(record(profile)); + if (marker.schema !== RETIREMENT_SCHEMA || !location || sandbox.leaseId !== location.leaseId + || sandbox.providerLeaseId !== location.providerLeaseId || sandbox.remoteCwd !== location.remoteCwd) return null; + return location; +} + +/** + * Retires the inherited remote runner identity of a due same-run retry when + * durable state proves all of these: + * + * - the failed attempt is a result-less exact-checkpoint retry that recorded + * `providerEventsExist: false`, and no controller owns it; + * - this run never requested or identified a runner, invoked the adapter, + * started a tool, or committed a runner event, has no controller checkpoint + * of its own, and has no provider event or provider session of its own; + * - its saved session has no active turn and no pending runtime request; and + * - its exact process identity is the one a settled predecessor of the same + * native session recorded. + * + * Anything else, including rows saved by a release that did not record these + * facts, keeps the existing verified reattachment path. + */ +export async function retireInheritedRemoteRunnerIdentity(db: Db, runId: string): Promise { + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + await tx.execute(sql`select set_config('lock_timeout', '2000', true)`); + const [candidate] = await tx.select({ companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.nativeIssueId }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).limit(1); + if (!candidate?.issueId) return false; + // Same lock order as same-run resumption: issue, coordinator, run. + const [issue] = await tx.select({ id: issues.id }).from(issues).where(and( + eq(issues.companyId, candidate.companyId), eq(issues.id, candidate.issueId))).for("update").limit(1); + if (!issue) return false; + const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, candidate.companyId), eq(nativeRunFinalizations.runId, runId), + eq(nativeRunFinalizations.issueId, issue.id))).for("update").limit(1); + const [run] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, runId))).for("update").limit(1); + if (!coordinator || !run || run.runtimeMode !== "native" || isNativeRunnerOwnershipHeld(run) + || !["running", "failed"].includes(run.status) || run.sessionIdAfter + || (run.processPid === null && run.processGroupId === null) || !run.nativeSessionId) return false; + const detail = record(coordinator.failureDetail); + if (coordinator.phase !== "retryable_failure" || coordinator.resultId !== null || coordinator.leaseOwner !== null + || detail.recoveryMode !== "exact_checkpoint_resume" || detail.providerEventsExist !== false + || typeof detail.successorRunId === "string") return false; + + const profile = record(run.runnerProfileJson); + const location = savedRemoteLocation(profile); + const checkpoint = record(profile.sessionCheckpoint); + const pending = checkpoint.pendingRuntimeRequests; + if (!location || checkpoint.activeTurnId != null + || (pending !== undefined && pending !== null && (!Array.isArray(pending) || pending.length > 0))) return false; + if (await hasNativeExecutionStartEvidence(tx, run)) return false; + if ((await nativeSessionProviderEvidence(tx, [run.id])).has(run.id)) return false; + if (await hasNativeRunnerCheckpoint(tx, { companyId: run.companyId, issueId: issue.id, runId: run.id })) return false; + + // The identity must be exactly the one a settled predecessor recorded. + const [predecessor] = await tx.select({ id: heartbeatRuns.id, processStartedAt: heartbeatRuns.processStartedAt }) + .from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId), + eq(heartbeatRuns.nativeSessionId, run.nativeSessionId), ne(heartbeatRuns.id, run.id), + lt(heartbeatRuns.createdAt, run.createdAt), inArray(heartbeatRuns.status, TERMINAL_RUN_STATUSES), + run.processPid === null ? isNull(heartbeatRuns.processPid) : eq(heartbeatRuns.processPid, run.processPid), + run.processGroupId === null ? isNull(heartbeatRuns.processGroupId) : eq(heartbeatRuns.processGroupId, run.processGroupId), + )).orderBy(desc(heartbeatRuns.createdAt)).limit(1); + if (!predecessor || !sameTime(predecessor.processStartedAt, run.processStartedAt)) return false; + + // Keep the run's own sandbox when its lease is still live. + const [lease] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.id, location.leaseId), + eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.providerLeaseId, location.providerLeaseId), + eq(environmentLeases.status, "active"), isNull(environmentLeases.releasedAt), isNull(environmentLeases.cleanupStatus), + or(isNull(environmentLeases.expiresAt), sql`${environmentLeases.expiresAt} > now()`))).limit(1); + + await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id, agentId: run.agentId }); + await tx.update(heartbeatRuns).set({ + processPid: null, processGroupId: null, processStartedAt: null, + runnerProfileJson: { ...profile, [NATIVE_INHERITED_RUNNER_RETIREMENT_KEY]: { + schema: RETIREMENT_SCHEMA, predecessorRunId: predecessor.id, + ...(lease ? { retainedSandbox: location } : {}), + } }, + updatedAt: new Date(), + }).where(eq(heartbeatRuns.id, run.id)); + await appendHeartbeatRunEvent(tx, { + companyId: run.companyId, runId: run.id, agentId: run.agentId, + eventType: "lifecycle", stream: "system", level: "info", + message: "Retrying this run in its own sandbox: it never started a runner, so there was nothing to reattach to.", + payload: { reason: "inherited_runner_never_launched", predecessorRunId: predecessor.id, + retainedSandbox: Boolean(lease) }, + }); + return true; + }).catch(error => { + logger.warn({ err: error, runId }, "failed to retire an inherited remote runner identity"); + return false; + }); +} diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 480ad9248a..3c862b9f81 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -714,7 +714,7 @@ export function classifyNativeRunnerRecoveryEvidence(input: { /** Any record that execution reached a runner: a requested or identified * process, an adapter invocation, a tool start, or a runner-sourced event. */ -async function hasNativeExecutionStartEvidence( +export async function hasNativeExecutionStartEvidence( tx: Pick, run: { companyId: string; id: string }, ): Promise { const [executionEvent] = await tx