diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 2e98adcf32..9fad8de993 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -664,6 +664,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { processGroupId?: number | null; processLossRetryCount?: number; runtimeMode?: "legacy" | "native"; + scheduledRetryAttempt?: number | null; + scheduledRetryReason?: string | null; + scheduledRetryAt?: Date | null; includeIssue?: boolean; runErrorCode?: string | null; runError?: string | null; @@ -726,6 +729,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { processGroupId: input?.processGroupId ?? null, processLossRetryCount: input?.processLossRetryCount ?? 0, ...(input?.runtimeMode ? { runtimeMode: input.runtimeMode } : {}), + scheduledRetryAttempt: input?.scheduledRetryAttempt ?? 0, + scheduledRetryReason: input?.scheduledRetryReason ?? null, + scheduledRetryAt: input?.scheduledRetryAt ?? null, errorCode: input?.runErrorCode ?? null, error: input?.runError ?? null, nextEventSeq: 2, @@ -2447,8 +2453,364 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(checkoutReleasedIssue?.checkoutRunId).toBeNull(); }); - it("requires reconciliation for a lost monitor whose provider outcomes are unknown", async () => { + it("schedules a null-environment process loss with diagnostics instead of stranding it", async () => { + const { companyId, agentId, runId } = await seedRunFixture({ + agentStatus: "idle", + processPid: null, + processGroupId: null, + }); + const heartbeat = heartbeatService(db); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [runId] }); + + const runs = await db + .select() + .from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId))); + const failed = runs.find((row) => row.id === runId); + const retry = runs.find((row) => row.retryOfRunId === runId); + + expect(failed?.resultJson).toMatchObject({ + environmentAllocationDiagnostic: { + phase: "environment_selection", + outcome: "failed", + reasonCode: "no_environment_or_lease_recorded", + }, + }); + expect(failed?.stderrExcerpt).toContain("[environment-allocation]"); + expect(retry).toMatchObject({ + status: "scheduled_retry", + scheduledRetryAttempt: 1, + scheduledRetryReason: "retry_transient_environment_failure", + }); + expect(retry?.scheduledRetryAt?.getTime()).toBe((failed?.finishedAt?.getTime() ?? 0) + 60_000); + expect(retry?.contextSnapshot).toMatchObject({ + wakeReason: "process_lost_environment_retry", + retryReason: "retry_transient_environment_failure", + retryOfRunId: runId, + issueId: failed?.contextSnapshot?.issueId, + }); + }); + + it("schedules attempt 2 of the null-environment ladder at 180s and attempt 3 at 540s", async () => { + const attempt2 = await seedRunFixture({ + agentStatus: "idle", + processPid: null, + processGroupId: null, + scheduledRetryAttempt: 1, + scheduledRetryReason: "retry_transient_environment_failure", + }); + const heartbeat = heartbeatService(db); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [attempt2.runId] }); + + const attempt2Rows = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, attempt2.agentId)); + const attempt2Failed = attempt2Rows.find((row) => row.id === attempt2.runId); + const attempt2Retry = attempt2Rows.find((row) => row.retryOfRunId === attempt2.runId); + expect(attempt2Retry).toMatchObject({ + status: "scheduled_retry", + scheduledRetryAttempt: 2, + scheduledRetryReason: "retry_transient_environment_failure", + }); + expect(attempt2Retry?.scheduledRetryAt?.getTime()).toBe( + (attempt2Failed?.finishedAt?.getTime() ?? 0) + 180_000, + ); + + const attempt3 = await seedRunFixture({ + agentStatus: "idle", + processPid: null, + processGroupId: null, + scheduledRetryAttempt: 2, + scheduledRetryReason: "retry_transient_environment_failure", + contextSnapshot: { + wakeReason: "process_lost_environment_retry", + retryReason: "retry_transient_environment_failure", + retryOfRunId: attempt2.runId, + }, + }); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [attempt3.runId] }); + + const attempt3Rows = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, attempt3.agentId)); + const attempt3Failed = attempt3Rows.find((row) => row.id === attempt3.runId); + const attempt3Retry = attempt3Rows.find((row) => row.retryOfRunId === attempt3.runId); + expect(attempt3Retry).toMatchObject({ + status: "scheduled_retry", + scheduledRetryAttempt: 3, + scheduledRetryReason: "retry_transient_environment_failure", + }); + expect(attempt3Retry?.scheduledRetryAt?.getTime()).toBe( + (attempt3Failed?.finishedAt?.getTime() ?? 0) + 540_000, + ); + }); + + it("does not queue another null-environment retry past attempt 3 and emits a single exhaustion event", async () => { + const { companyId, agentId, runId } = await seedRunFixture({ + agentStatus: "idle", + processPid: null, + processGroupId: null, + scheduledRetryAttempt: 3, + scheduledRetryReason: "retry_transient_environment_failure", + contextSnapshot: { + wakeReason: "process_lost_environment_retry", + retryReason: "retry_transient_environment_failure", + retryOfRunId: "previous-run", + }, + }); + const heartbeat = heartbeatService(db); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [runId] }); + + // After attempt 3, the bounded null-environment ladder must NOT queue another + // retry on the same reason chain. The release/promote path may still spin up + // an issue.continuation_recovery run as a separate auto-recovery attempt, but + // it must NOT carry the null-environment retry reason. + const retries = await db + .select() + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.retryOfRunId, runId), + )); + const nullEnvRetries = retries.filter( + (row) => row.scheduledRetryReason === "retry_transient_environment_failure", + ); + expect(nullEnvRetries).toHaveLength(0); + + const exhaustionEvents = await db + .select() + .from(heartbeatRunEvents) + .where(and( + eq(heartbeatRunEvents.companyId, companyId), + eq(heartbeatRunEvents.runId, runId), + eq(heartbeatRunEvents.eventType, "lifecycle"), + )); + const exhaustion = exhaustionEvents.find((event) => + typeof event.message === "string" + && event.message.includes("Bounded retry exhausted") + && (event.payload as Record | null)?.retryReason === "retry_transient_environment_failure", + ); + expect(exhaustion).toBeDefined(); + }); + + it("does not route a monitor-dispatch loss through the null-environment ladder", async () => { + // The fingerprint (no pid, no pgid, no lease) matches monitor-dispatch + // losses too. Monitor-dispatch losses are owned by the monitor scheduler + // (a future wake is already scheduled), so the reaper must NOT enter the + // bounded null-env ladder. Instead the issue is escalated to the board + // via a legacy_execution_requires_reconciliation recovery action. const { agentId, runId, issueId } = await seedRunFixture({ + adapterType: "openclaw_gateway", + agentStatus: "idle", + processPid: null, + processGroupId: null, + contextSnapshot: { + wakeReason: "issue_monitor_due", + nextCheckAt: "2026-03-19T00:00:00.000Z", + }, + }); + const heartbeat = heartbeatService(db); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [runId] }); + + const rows = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + const failed = rows.find((row) => row.id === runId); + const retry = rows.find((row) => row.retryOfRunId === runId); + + // No environment-allocation diagnostic should be attached — the run was a + // monitor dispatch loss, not an environment-selection failure. + expect(failed?.resultJson).not.toMatchObject({ + environmentAllocationDiagnostic: expect.objectContaining({ + phase: "environment_selection", + }), + }); + expect(failed?.stderrExcerpt ?? "").not.toContain("[environment-allocation]"); + // No retry at all — monitor-dispatch with a future wake is handled by the + // monitor scheduler, not by the null-env or legacy retry ladder. + expect(retry).toBeUndefined(); + // The issue is escalated to the board via the reconciliation recovery path. + const actions = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(actions).toEqual([ + expect.objectContaining({ + ownerType: "board", + cause: "legacy_execution_requires_reconciliation", + }), + ]); + }); + + it("does not route a plan-approval continuation loss through the null-environment ladder", async () => { + // A request_confirmation/accepted continuation wake has the same null-env + // fingerprint, but is owned by the `interaction_continuation_infra_retry` + // path. The bounded null-env ladder must NOT override it. + const { companyId, agentId, runId, wakeupRequestId, issueId } = await seedQueuedIssueRunFixture(); + const interactionId = randomUUID(); + + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee_on_accept", + createdByAgentId: agentId, + resolvedByUserId: "responsible-user", + resolvedAt: new Date("2026-03-19T00:00:00.000Z"), + payload: { + version: 1, + prompt: "Approve the plan?", + target: { type: "issue_document", issueId, key: "plan", revisionId: randomUUID() }, + }, + result: { version: 1, outcome: "accepted" }, + }); + await db + .update(agentWakeupRequests) + .set({ + source: "automation", + reason: "issue_commented", + status: "claimed", + payload: { + issueId, + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + mutation: "interaction", + }, + }) + .where(eq(agentWakeupRequests.id, wakeupRequestId)); + await db + .update(heartbeatRuns) + .set({ + status: "running", + invocationSource: "automation", + processPid: null, + processGroupId: null, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_commented", + mutation: "interaction", + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }, + startedAt: new Date("2026-03-19T00:00:00.000Z"), + updatedAt: new Date("2026-03-19T00:00:00.000Z"), + }) + .where(eq(heartbeatRuns.id, runId)); + await db.update(issues).set({ status: "in_review" }).where(eq(issues.id, issueId)); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.reapOrphanedRuns(); + expect(result.reaped).toBe(1); + + const rows = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + const retry = rows.find((row) => row.retryOfRunId === runId); + // Plan-approval path keeps its own retry reason; the null-env ladder + // must not have fired. + expect(retry?.scheduledRetryReason).toBe(INTERACTION_CONTINUATION_INFRA_RETRY_REASON); + }); + + it("does not route a process-loss retry through the null-environment ladder when the legacy retry budget is exhausted", async () => { + // The bounded null-env ladder shares the de-facto "have we already retried?" + // budget with the legacy `process_lost_retry` path. A run whose + // processLossRetryCount has already reached 1 must NOT enter the bounded + // null-env ladder even when the fingerprint matches. + const { agentId, runId } = await seedRunFixture({ + adapterType: "codex_local", + agentStatus: "idle", + processPid: null, + processGroupId: null, + processLossRetryCount: 1, + contextSnapshot: { + wakeReason: "process_lost_retry", + retryReason: "issue_continuation_needed", + retryOfRunId: "original-run", + }, + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reapOrphanedRuns(); + expect(result).toEqual({ reaped: 1, runIds: [runId] }); + + const rows = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + const failed = rows.find((row) => row.id === runId); + // No environment-allocation diagnostic should be attached — this run was + // already retried once via the legacy path and is past the budget. + expect(failed?.resultJson).not.toMatchObject({ + environmentAllocationDiagnostic: expect.objectContaining({ + phase: "environment_selection", + }), + }); + // No further retry should be scheduled. + const retries = rows.filter((row) => row.retryOfRunId === runId); + expect(retries).toHaveLength(0); + }); + + it("does not route process loss through the null-environment ladder when a child pid was recorded", async () => { + const { companyId, agentId, runId } = await seedRunFixture({ + adapterType: "codex_local", + agentStatus: "idle", + processPid: 4321, + processGroupId: null, + }); + const heartbeat = heartbeatService(db); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [runId] }); + + const rows = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + const failed = rows.find((row) => row.id === runId); + const retry = rows.find((row) => row.retryOfRunId === runId); + + expect(failed?.resultJson).not.toMatchObject({ + environmentAllocationDiagnostic: expect.objectContaining({ + phase: "environment_selection", + }), + }); + expect(failed?.stderrExcerpt ?? "").not.toContain("[environment-allocation]"); + // Legacy path enqueues an immediate retry. The retry row is inserted + // with status "scheduled_retry" (dueAt = now + 30s) and only promoted by + // promoteDueScheduledRetries once scheduledRetryAt has elapsed, so the + // observed status here is one of: scheduled_retry, queued, or running. + expect(retry).toMatchObject({ + retryOfRunId: runId, + processLossRetryCount: 1, + }); + expect(["scheduled_retry", "queued", "running"]).toContain(retry?.status); + // Legacy path uses the immediate process_lost_retry wake reason, not the + // bounded environment retry wake reason. + expect(retry?.contextSnapshot).toMatchObject({ + wakeReason: "process_lost_retry", + }); + expect(retry?.contextSnapshot).not.toMatchObject({ + wakeReason: "process_lost_environment_retry", + }); + // And it must NOT carry the null-environment scheduledRetryReason. + expect(retry?.scheduledRetryReason).not.toBe("retry_transient_environment_failure"); + }); + + it("restores one lost monitor dispatch before escalating a second process loss", async () => { + const { companyId, agentId, runId, issueId } = await seedRunFixture({ adapterType: "openclaw_gateway", agentStatus: "idle", processPid: null, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 3de2209b3c..3f1d2789b4 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -755,13 +755,14 @@ export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS = [ const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_JITTER_RATIO = 0; const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_REASON = "transient_failure"; const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_WAKE_REASON = "transient_failure_retry"; -const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS = - BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length; -export { - INTERACTION_CONTINUATION_INFRA_RETRY_REASON, - INTERACTION_CONTINUATION_INFRA_WAKE_REASON, -}; +const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS = BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length; +const NULL_ENVIRONMENT_PROCESS_LOSS_RETRY_REASON = "retry_transient_environment_failure"; +const NULL_ENVIRONMENT_PROCESS_LOSS_WAKE_REASON = "process_lost_environment_retry"; +const PROCESS_LOST_RETRY_REASON = "process_lost_retry"; +const PROCESS_LOST_RETRY_WAKE_REASON = "process_lost_retry"; +const NULL_ENVIRONMENT_PROCESS_LOSS_RETRY_DELAYS_MS = [60_000, 180_000, 540_000] as const; const INTERACTION_CONTINUATION_INFRA_MAX_ATTEMPTS = 2; +const RESOLVED_INTERACTION_CONTINUATION_STATUSES = new Set(["accepted", "answered", "rejected"]); const WORKSPACE_VALIDATION_FAILURE_CODE = "workspace_validation_failed"; const WORKSPACE_VALIDATION_RECOVERY_CAUSE = "workspace_validation_failed"; const CONFIGURATION_INCOMPLETE_FAILURE_CODE = "configuration_incomplete"; @@ -830,6 +831,10 @@ const GIT_SENSITIVE_LOCAL_ADAPTER_TYPES = new Set([ ]); export { MAX_TURN_CONTINUATION_RETRY_REASON }; export const MAX_TURN_CONTINUATION_WAKE_REASON = "max_turns_continuation_retry"; +export { + INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + INTERACTION_CONTINUATION_INFRA_WAKE_REASON, +}; const MAX_TURN_CONTINUATION_DEFAULT_MAX_ATTEMPTS = 2; const MAX_TURN_CONTINUATION_MAX_ATTEMPTS_CAP = 10; const MAX_TURN_CONTINUATION_DEFAULT_DELAY_MS = 1_000; @@ -1932,6 +1937,21 @@ export function computeBoundedTransientHeartbeatRetrySchedule( }; } +// This signature deliberately excludes a process that was ever spawned. A +// lost child process remains on the legacy process-loss path; this ladder is +// only for dispatches that died before an execution environment existed. +export function isNullEnvironmentProcessLoss(input: { + usageJson: unknown; + processPid: number | null; + processGroupId: number | null; + hasEnvironmentLease: boolean; +}) { + return input.usageJson == null && + input.processPid == null && + input.processGroupId == null && + !input.hasEnvironmentLease; +} + async function resolveRunScopedMentionedSkillKeys(input: { db: Db; companyId: string; @@ -14131,8 +14151,20 @@ export function heartbeatService( // transient retries; process loss must not open a second retry budget. if (run.runtimeMode === "native" || legacyExecutionNeedsReconciliation(run)) return null; - const scheduled = await scheduleBoundedRetryForRun(run, agent, { now }); - return scheduled.outcome === "scheduled" ? scheduled.run : null; + const successorLossRetryCount = (run.processLossRetryCount ?? 0) + 1; + const scheduled = await scheduleBoundedRetryForRun(run, agent, { + now, + retryReason: PROCESS_LOST_RETRY_REASON, + wakeReason: PROCESS_LOST_RETRY_WAKE_REASON, + }); + if (scheduled.outcome !== "scheduled" || !scheduled.run) return null; + // Mirror the budget that the reaper's alreadyRetriedOnce guard reads. + const [bumped] = await db + .update(heartbeatRuns) + .set({ processLossRetryCount: successorLossRetryCount }) + .where(eq(heartbeatRuns.id, scheduled.run.id)) + .returning(); + return bumped ?? { ...scheduled.run, processLossRetryCount: successorLossRetryCount }; } function toHotRestartIntentRun(input: { @@ -16057,7 +16089,21 @@ export function heartbeatService( return null; } - return scheduleBoundedRetryForRun(run, agent, { + // The reaper promoted this run to a process_lost CAS failure before any + // provider work produced output; the retry is an explicit + // infrastructure-loss replay, not an ambiguous bootstrap that the legacy + // reconciliation gate is meant to block. Mark the failed run as safe + // bootstrap evidence on the in-memory copy we hand to + // scheduleBoundedRetryForRun so the shared gate does not refuse the retry. + const runForRetry: typeof run = { + ...run, + resultJson: { + ...(parseObject(run.resultJson) ?? {}), + executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + }, + }; + + return scheduleBoundedRetryForRun(runForRetry, agent, { retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, maxAttempts: INTERACTION_CONTINUATION_INFRA_MAX_ATTEMPTS, @@ -18706,49 +18752,127 @@ export function heartbeatService( readNonEmptyString(runContext.wakeReason) === "issue_monitor_due" && monitorNextCheckAt !== undefined && (!monitorNextCheckAt || monitorNextCheckAt.getTime() <= now.getTime()); - const shouldRetry = - (run.processLossRetryCount ?? 0) < 1 && - ((tracksLegacyLocalChild && - (!!run.processPid || !!run.processGroupId)) || - monitorDispatchLostWithoutFutureWake); + const environmentLease = await db + .select({ id: environmentLeases.id }) + .from(environmentLeases) + .where(and( + eq(environmentLeases.companyId, run.companyId), + eq(environmentLeases.heartbeatRunId, run.id), + )) + .limit(1) + .then((rows) => rows[0] ?? null); + // Atomically revoke an expired legacy controller lease before we + // classify or terminalize. Renewal and revocation serialize on the run + // row, so a concurrent renewal that wins the CAS means another path is + // still owning the run and we must skip this iteration. if (!(await revokeExpiredLegacyController(db, run))) continue; + // The null-environment retry ladder only fires when no more specific + // retry path already owns the run: + // - Monitor-dispatch losses fall through to the legacy + // `process_lost_retry` path (which uses + // monitorDispatchLostWithoutFutureWake to decide whether a future + // wake is already scheduled). + // - Resolved interaction-continuation wakes fall through to the + // plan-approval infrastructure retry. + // - Runs that have already been retried once via the legacy path + // (`processLossRetryCount >= 1`) are not eligible for the bounded + // null-env ladder; the legacy retry count is the de-facto + // "have we already retried?" budget shared with the new ladder. + const isMonitorDispatchRun = readNonEmptyString(runContext.wakeReason) === "issue_monitor_due"; + const alreadyRetriedOnce = (run.processLossRetryCount ?? 0) >= 1; + const nullEnvironmentProcessLoss = + !isMonitorDispatchRun && + !alreadyRetriedOnce && + !isResolvedInteractionContinuationWakeContext(runContext) && + isNullEnvironmentProcessLoss({ + usageJson: run.usageJson, + processPid: run.processPid, + processGroupId: run.processGroupId, + hasEnvironmentLease: environmentLease !== null, + }); + const shouldRetryLegacyProcessLoss = (run.processLossRetryCount ?? 0) < 1 && ( + (tracksLegacyLocalChild && (!!run.processPid || !!run.processGroupId)) || + monitorDispatchLostWithoutFutureWake + ); + const shouldRetry = nullEnvironmentProcessLoss || shouldRetryLegacyProcessLoss; const baseMessage = buildProcessLossMessage(run); - const conversationContinuationEligible = await runUsedConversationAdapter(db, run); + const allocationDiagnostic = nullEnvironmentProcessLoss + ? { + phase: "environment_selection", + outcome: "failed", + reasonCode: "no_environment_or_lease_recorded", + environmentId: null, + leaseId: null, + scratchDirHealth: "unknown", + capturedAt: now.toISOString(), + } + : null; + const allocationDiagnosticLine = allocationDiagnostic + ? `[environment-allocation] ${allocationDiagnostic.phase}:${allocationDiagnostic.reasonCode}` + : null; + const unmanagedBackgroundTaskEvidence = null; + const failurePatch = { + error: shouldRetry + ? `${baseMessage}; ${nullEnvironmentProcessLoss ? "scheduling bounded environment retry" : "retrying once"}` + : baseMessage, + errorCode: "process_lost", + finishedAt: now, + resultJson: await (async () => { + // Only runs whose historical invocation actually used a conversation + // adapter can carry the continuation policy on a process-loss stop. + // The agent's CURRENT adapter type must not relabel a lost process + // run when an admin switches it mid-flight (see test "does not + // relabel a lost process run when its agent changes to a + // conversation adapter"); runUsedConversationAdapter checks the + // persisted invocation event, not agents.adapterType. + const conversationContinuationEligible = await runUsedConversationAdapter(db, run); + const result = mergeRunStopMetadataForAgent( + { adapterType, adapterConfig }, + "failed", + { + conversationContinuationEligible, + resultJson: parseObject(run.resultJson), + errorCode: "process_lost", + errorMessage: shouldRetry + ? `${baseMessage}; ${nullEnvironmentProcessLoss ? "scheduling bounded environment retry" : "retrying once"}` + : baseMessage, + }, + ); + const withAllocationDiagnostic = allocationDiagnostic + ? { ...result, environmentAllocationDiagnostic: allocationDiagnostic } + : result; + return unmanagedBackgroundTaskEvidence + ? { + ...withAllocationDiagnostic, + stopReason: UNMANAGED_BACKGROUND_TASK_STOP_REASON, + unmanagedBackgroundTask: unmanagedBackgroundTaskEvidence, + } + : withAllocationDiagnostic; + })(), + ...(allocationDiagnosticLine + ? { stderrExcerpt: appendWithByteCap(run.stderrExcerpt ?? "", allocationDiagnosticLine, MAX_EXCERPT_BYTES) } + : {}), + }; + // Compare-and-set terminalization: the CAS guarantees another concurrent + // recovery/drain path that already moved this run out of "running" wins + // the race, and we keep that terminal outcome instead of overwriting it. + // The native-ownership predicate inside setRunStatusFromLive also blocks + // terminalizing a native run whose ownership is still held elsewhere. const failureWrite = await setRunStatusFromLive( run.id, "failed", ["running"], - { - error: shouldRetry ? `${baseMessage}; retrying once` : baseMessage, - errorCode: "process_lost", - finishedAt: now, - resultJson: (() => { - const result = mergeRunStopMetadataForAgent( - { adapterType, adapterConfig }, - "failed", - { - conversationContinuationEligible, - resultJson: parseObject(run.resultJson), - errorCode: "process_lost", - errorMessage: shouldRetry - ? `${baseMessage}; retrying once` - : baseMessage, - }, - ); - return result; - })(), - }, + failurePatch, ); - if (!failureWrite.updated || !failureWrite.run) continue; - let finalizedRun: typeof heartbeatRuns.$inferSelect | null = - failureWrite.run; + if (!(failureWrite.updated && failureWrite.run)) continue; + let finalizedRun: typeof failureWrite.run = failureWrite.run; await setWakeupStatus(run.wakeupRequestId, "failed", { finishedAt: now, - error: shouldRetry ? `${baseMessage}; retrying once` : baseMessage, + error: shouldRetry + ? `${baseMessage}; ${nullEnvironmentProcessLoss ? "scheduling bounded environment retry" : "retrying once"}` + : baseMessage, }); - if (!finalizedRun) finalizedRun = await getRun(run.id); - if (!finalizedRun) continue; finalizedRun = (await classifyAndPersistRunLiveness( finalizedRun, @@ -18764,7 +18888,20 @@ export function heartbeatService( let retriedRun: typeof heartbeatRuns.$inferSelect | null = null; const retryAgent = await getAgent(run.agentId); - if (shouldRetry) { + if (nullEnvironmentProcessLoss) { + if (retryAgent) { + const attempt = (finalizedRun.scheduledRetryAttempt ?? 0) + 1; + const delayMs = NULL_ENVIRONMENT_PROCESS_LOSS_RETRY_DELAYS_MS[attempt - 1]; + const scheduled = await scheduleBoundedRetryForRun(finalizedRun, retryAgent, { + now, + retryReason: NULL_ENVIRONMENT_PROCESS_LOSS_RETRY_REASON, + wakeReason: NULL_ENVIRONMENT_PROCESS_LOSS_WAKE_REASON, + maxAttempts: NULL_ENVIRONMENT_PROCESS_LOSS_RETRY_DELAYS_MS.length, + ...(delayMs != null ? { delayMs } : {}), + }); + retriedRun = scheduled.outcome === "scheduled" ? scheduled.run : null; + } + } else if (shouldRetryLegacyProcessLoss) { if (retryAgent) { retriedRun = await enqueueProcessLossRetry( finalizedRun, @@ -18795,6 +18932,7 @@ export function heartbeatService( payload: { ...(run.processPid ? { processPid: run.processPid } : {}), ...(run.processGroupId ? { processGroupId: run.processGroupId } : {}), + ...(allocationDiagnostic ? { environmentAllocationDiagnostic: allocationDiagnostic } : {}), ...(retriedRun ? { retryRunId: retriedRun.id } : {}), }, });