diff --git a/server/src/__tests__/chat-run-milestone-recovery-work.test.ts b/server/src/__tests__/chat-run-milestone-recovery-work.test.ts index f5fb96b4e1d..f6b43e3b611 100644 --- a/server/src/__tests__/chat-run-milestone-recovery-work.test.ts +++ b/server/src/__tests__/chat-run-milestone-recovery-work.test.ts @@ -50,7 +50,9 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu it.each(["process_gone", "issue_cancelled"] as const)("publishes the recovered terminal state after %s and becomes quiet", async reason => { const runId = randomUUID(); await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, - status: "running", runtimeMode: "legacy", responsibleUserId: f.userId, + // The sweep's own process-gone backstop is native; a legacy run goes to + // the one lost-process reaper. + status: "running", runtimeMode: reason === "process_gone" ? "native" : "legacy", responsibleUserId: f.userId, processPid: reason === "process_gone" ? 2_000_000_000 : process.pid, contextSnapshot: { issueId: f.issueId, source: "chat:slack", endpointId: f.endpointId, wakeCommentId: f.commentId }, }); diff --git a/server/src/__tests__/live-recovery-work.test.ts b/server/src/__tests__/live-recovery-work.test.ts index eaade7ebc9c..4bdacaf6c7f 100644 --- a/server/src/__tests__/live-recovery-work.test.ts +++ b/server/src/__tests__/live-recovery-work.test.ts @@ -1,3 +1,4 @@ +import { currentProcessSpaceHash } from "../services/controller-process-identity.js"; import { randomUUID } from "node:crypto"; import os from "node:os"; import { spawn } from "node:child_process"; @@ -346,14 +347,17 @@ describe("residual recovery production matrix", () => { expect(await getRun(run.id)).toMatchObject({ status: "running", executionStage: "dispatching", errorCode: null }); expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run.id))).toEqual([]); }); - it("keeps the stale-age guard for this boot, a recorded process, or a dispatched adapter", async () => { + // A recorded process is judged by the shared prover, not by age: a live one + // is held (a proven-dead one settles at once, recovery-stale-issue-lock-sweep). + it("keeps the stale-age guard for this boot or a dispatched adapter, and holds a live recorded process", async () => { const f = await fixture(), now = Date.now(), expired = new Date(now - 1_000); const [sameBoot] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, status: "running", runtimeMode: "legacy", responsibleUserId: f.actorId, controllerBootId: legacyControllerBootId, controllerLeaseExpiresAt: expired, executionStage: "preparing", updatedAt: new Date(now), startedAt: new Date(now) }).returning(); const [withProcess] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, status: "running", runtimeMode: "legacy", responsibleUserId: f.actorId, controllerBootId: randomUUID(), controllerLeaseExpiresAt: expired, - executionStage: "dispatching", processPid: 2 ** 22 + 17, updatedAt: new Date(now), startedAt: new Date(now) }).returning(); + executionStage: "dispatching", processPid: process.pid, processHostHash: currentProcessSpaceHash(), + updatedAt: new Date(now), startedAt: new Date(now) }).returning(); // A dispatched adapter may hold remote work without a local process. const [dispatched] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, status: "running", runtimeMode: "legacy", responsibleUserId: f.actorId, controllerBootId: randomUUID(), controllerLeaseExpiresAt: expired, diff --git a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts index 9459d0572f0..8efca7b2f77 100644 --- a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts +++ b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts @@ -127,16 +127,19 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { await db.update(heartbeatRuns).set({ processPid: 2_000_000_000, contextSnapshot: { issueId, explicitUserContinuation: { previousRunId: failedRunId, commentId: randomUUID() } }, }).where(eq(heartbeatRuns.id, runningRunId)); - const beforeOrphanedRunTerminalWrite = vi.fn(async () => { + // The sweep hands the legacy run to the one lost-process settle path; + // finalization wins that race and records the run timed out. + const settleLostRunProcess = vi.fn(async () => { await db.update(heartbeatRuns).set({ status: "timed_out", finishedAt: new Date() }) .where(eq(heartbeatRuns.id, runningRunId)); }); const settleExplicitContinuationRetry = vi.fn(async () => {}); - const result = await recoveryService(db, { enqueueWakeup: vi.fn(), beforeOrphanedRunTerminalWrite, + const result = await recoveryService(db, { enqueueWakeup: vi.fn(), settleLostRunProcess, ...(withSettler ? { settleExplicitContinuationRetry } : {}), }).sweepStaleIssueLocks(); - expect(beforeOrphanedRunTerminalWrite).toHaveBeenCalledOnce(); - expect(result).toEqual({ cleared: 0, issueIds: [], terminalizedRunIds: [] }); + expect(settleLostRunProcess).toHaveBeenCalledOnce(); + // The sweep reads the outcome back and never clears the retry's lock. + expect(result).toEqual({ cleared: 0, issueIds: [], terminalizedRunIds: [runningRunId] }); expect(settleExplicitContinuationRetry).toHaveBeenCalledTimes(withSettler ? 1 : 0); if (withSettler) expect(settleExplicitContinuationRetry).toHaveBeenCalledWith(expect.objectContaining({ id: runningRunId, status: "timed_out", @@ -439,17 +442,19 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { const result = await heartbeatService(db).sweepStaleIssueLocks(); expect(result.terminalizedRunIds.includes(runningRunId)).toBe(terminalized); const [run] = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runningRunId)); - expect(run?.status).toBe(terminalized ? "interrupted" : "running"); + // A legacy run with a proven-gone child gets the one lost-process + // reaper's outcome; an unprovable one stays running (Stop releases it). + expect(run?.status).toBe(terminalized ? "failed" : "running"); }); - it("terminalizes an orphaned running run whose process is gone, then clears the lock", async () => { + it("terminalizes an orphaned running native run whose process is gone, then clears the lock", async () => { const { companyId, agentId, runningRunId } = await seed(); // The run recorded a pid, but the process and its sandbox are gone. A pid - // this large never maps to a live process, so isPidAlive returns false. + // this large never maps to a live process, so the prover finds it gone. // The issue is not terminal, so only the process-death authority applies. await db .update(heartbeatRuns) - .set({ processPid: 2_000_000_000 }) + .set({ processPid: 2_000_000_000, runtimeMode: "native" }) .where(eq(heartbeatRuns.id, runningRunId)); const issueId = randomUUID(); await db.insert(issues).values({ @@ -1084,6 +1089,67 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { expect(activeLock).toEqual({ checkoutRunId: runningRunId, executionRunId: runningRunId }); }); + // One lost-process reaper (Phase 3, non-wake recovery): the orphan sweep no + // longer judges a legacy child itself. It hands the run to the + // active-execution reaper, so the run gets that reaper's failed/process_lost + // outcome and retry policy, with no age guard once the child is proven gone, + // and the dead run no longer holds the task. + it("hands a legacy run whose child is proven gone to the one lost-process reaper", async () => { + const { companyId, agentId, runningRunId } = await seed(); + await db.update(heartbeatRuns).set({ processPid: 2_000_000_000, processHostHash: currentProcessSpaceHash(), + updatedAt: new Date() }).where(eq(heartbeatRuns.id, runningRunId)); + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Lost legacy child", status: "in_progress", + priority: "high", assigneeAgentId: agentId, checkoutRunId: runningRunId, executionRunId: runningRunId, + executionLockedAt: new Date() }); + const result = await heartbeatService(db).sweepStaleIssueLocks(); + expect(result.terminalizedRunIds).toEqual([runningRunId]); + const [run] = await db.select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, runningRunId)); + expect(run).toEqual({ status: "failed", errorCode: "process_lost" }); + const [lock] = await db.select({ checkoutRunId: issues.checkoutRunId, executionRunId: issues.executionRunId }) + .from(issues).where(eq(issues.id, issueId)); + expect(lock.checkoutRunId).not.toBe(runningRunId); + expect(lock.executionRunId).not.toBe(runningRunId); + }); + + // RC-27: the reaper waited out a 5-minute age guard even when the shared + // prover could already show the legacy child was gone. Age is never proof. + it("settles a legacy run whose child is proven gone without waiting out the age guard", async () => { + const { runningRunId } = await seed(); + await db.update(heartbeatRuns).set({ processPid: 2_000_000_000, processHostHash: currentProcessSpaceHash(), + updatedAt: new Date() }).where(eq(heartbeatRuns.id, runningRunId)); + await heartbeatService(db).reapOrphanedRuns({ staleThresholdMs: 5 * 60_000 }); + const [run] = await db.select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, runningRunId)); + expect(run).toEqual({ status: "failed", errorCode: "process_lost" }); + }); + + // Greptile on #20004: the reaper proved children only for session adapters, + // so a handed-over `process` run with a live child fell through to + // process_lost once the age guard passed. + it("never releases a process-adapter run whose recorded child is still alive", async () => { + const { agentId, runningRunId } = await seed(); + await db.update(agents).set({ adapterType: "process" }).where(eq(agents.id, agentId)); + await db.update(heartbeatRuns).set({ processPid: process.pid, processHostHash: currentProcessSpaceHash(), + updatedAt: new Date(Date.now() - 10 * 60_000) }).where(eq(heartbeatRuns.id, runningRunId)); + await heartbeatService(db).reapOrphanedRuns({ staleThresholdMs: 5 * 60_000 }); + const [run] = await db.select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, runningRunId)); + expect(run).toEqual({ status: "running", errorCode: "process_detached" }); + }); + + it("never releases a legacy run whose child is unprovable, and names Stop as the way out", async () => { + const { runningRunId } = await seed(); + await db.update(heartbeatRuns).set({ processPid: 2_000_000_000, processHostHash: "another-host-pid-space", + updatedAt: new Date(Date.now() - 10 * 60_000) }).where(eq(heartbeatRuns.id, runningRunId)); + await heartbeatService(db).reapOrphanedRuns({ staleThresholdMs: 5 * 60_000 }); + const [run] = await db.select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode, error: heartbeatRuns.error }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, runningRunId)); + expect(run).toMatchObject({ status: "running", errorCode: "process_detached" }); + expect(run.error).toMatch(/cannot prove child pid .* stopped .*stop the run/); + }); + it("still clears the lock when the audit write fails after terminalization", async () => { const { companyId, agentId, runningRunId } = await seed(); // The run recorded a pid that never maps to a live process, so the sweep @@ -1091,7 +1157,7 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { // process-death authority drives the terminalization here. await db .update(heartbeatRuns) - .set({ processPid: 2_000_000_000 }) + .set({ processPid: 2_000_000_000, runtimeMode: "native" }) .where(eq(heartbeatRuns.id, runningRunId)); const issueId = randomUUID(); await db.insert(issues).values({ diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 785f697d1fe..19901756c0e 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1002,6 +1002,8 @@ export function heartbeatService( const recovery = recoveryService(db, { enqueueWakeup, liveRunExecutions, + // Bound at call time: the heartbeat recovery module is created below. + settleLostRunProcess: scope => recoverActiveExecutions(scope), settleExplicitContinuationRetry: releaseIssueExecutionAndPromote, scheduleRecoveryRetry: async (runId) => { const [run] = await db @@ -1058,6 +1060,7 @@ export function heartbeatService( prepareHotRestartShutdown, reconcileHotRestartAdoption, recoverNativeRunsAfterRestart, + recoverActiveExecutions, reapOrphanedRuns, sweepOrphanedActiveLeases, drainRunningRunsForShutdown, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index ea4a088ea48..cc442890675 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -98,7 +98,9 @@ import { } from "../native-runtime/native-runner-ownership.js"; import { performance } from "node:perf_hooks"; import { buildProcessLossDiagnostic } from "../process-loss-diagnostics.js"; -import { logActivity } from "../activity-log.js"; +import { logActivity, persistActivity } from "../activity-log.js"; +import { proveRunProcessGone } from "../run-process-owner.js"; +import { LEGACY_HOST_FALLBACK_EVENT } from "../native-runtime/native-restart-recovery.js"; import { releaseRuntimeServicesForRun } from "../workspace-runtime.js"; import { isProcessGroupAlive } from "../local-service-supervisor.js"; import { @@ -789,13 +791,16 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD const processPid = run.processPid ?? candidate.processPid; const processGroupId = run.processGroupId ?? candidate.processGroupId; - const processPidAlive = isProcessAlive(processPid); - const processGroupAlive = isProcessGroupAlive(processGroupId); if (!processPid && !processGroupId) { classify(candidate, "lost", "missing_process_metadata", patch); continue; } - if (!processPidAlive && !processGroupAlive) { + // The shared prover decides. Only a proven-gone child is lost; an + // unprovable one is never released, so it stays adopted and the + // active-execution reaper keeps holding it (Stop is its operator action). + const processProof = await proveRunProcessGone({ processPid, processGroupId, + processStartedAt: run.processStartedAt, processHostHash: run.processHostHash }, "hot_restart_adoption_marker"); + if (!processProof || processProof.verdict === "gone") { classify(candidate, "lost", "process_not_alive", patch); continue; } @@ -869,7 +874,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD classify( candidate, "adopted", - processPidAlive ? "process_pid_alive" : "process_group_alive", + processProof.reason === "process_group_alive" ? "process_group_alive" : "process_pid_alive", patch, ); } @@ -2535,15 +2540,6 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD run.controllerBootId !== legacyControllerBootId && !run.processPid && !run.processGroupId; - // Apply staleness threshold to avoid false positives - if (staleThresholdMs > 0 && !priorBootSetupExpired && !legacyControllerGone) { - const refTime = run.updatedAt ? new Date(run.updatedAt).getTime() : 0; - if (now.getTime() - refTime < staleThresholdMs) { - scope.onDeferredDeadline?.(refTime + staleThresholdMs); - continue; - } - } - const currentAdapterTracksLocalChild = isTrackedLocalChildProcessAdapter(adapterType); const tracksLegacyLocalChild = @@ -2554,16 +2550,35 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // overlapping provider/tool execution while that child is still alive. const checksPersistedChildLiveness = nativeRun ? nativeLocalProcess : currentAdapterTracksLocalChild; - const processPidAlive = - checksPersistedChildLiveness && - run.processPid && - isProcessAlive(run.processPid); + // A legacy child is judged by the shared owner prover, with the stamp the + // orphan sweep and the ownership report read. A proven-gone child needs + // no age guard. An unprovable one (another PID space, a sandbox, or an + // ephemeral server's pre-#19871 row) is never released here; Stop is its + // operator action. + // Any recorded child is proven, whatever the adapter: a run handed over + // by the orphan sweep (for example the `process` adapter) must never be + // released while its child may still work. + const legacyProcessProof = !nativeRun && (checksPersistedChildLiveness || run.processPid != null || run.processGroupId != null) + ? await proveRunProcessGone(run, "legacy_run_process") : null; + // Apply staleness threshold to avoid false positives + if (staleThresholdMs > 0 && !priorBootSetupExpired && !legacyControllerGone && legacyProcessProof?.verdict !== "gone") { + const refTime = run.updatedAt ? new Date(run.updatedAt).getTime() : 0; + if (now.getTime() - refTime < staleThresholdMs) { + scope.onDeferredDeadline?.(refTime + staleThresholdMs); + continue; + } + } + + const processPidAlive = nativeRun + ? checksPersistedChildLiveness && !!run.processPid && isProcessAlive(run.processPid) + : !!legacyProcessProof && legacyProcessProof.verdict !== "gone"; const processGroupAlive = + nativeRun && checksPersistedChildLiveness && - run.processGroupId && + !!run.processGroupId && isProcessGroupAlive(run.processGroupId); if ( - (processPidAlive || processGroupAlive) && + (nativeRun ? processPidAlive || processGroupAlive : legacyProcessProof?.verdict === "alive") && readHotRestartAdoptionMetadata(parseObject(run.resultJson)) ) { scope.onDeferredDeadline?.(Date.now() + 30_000); @@ -2571,13 +2586,18 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD } if (processPidAlive || processGroupAlive) { if (run.errorCode !== DETACHED_PROCESS_ERROR_CODE) { - const detachedMessage = processPidAlive - ? `Lost in-memory process handle, but child pid ${run.processPid} is still alive` - : `Lost in-memory process handle, but persisted process group ${run.processGroupId} is still alive`; - const detachedRun = await setRunStatus(run.id, "running", { + const detachedMessage = legacyProcessProof?.verdict === "unprovable" + ? `Lost in-memory process handle, and cannot prove child pid ${run.processPid ?? run.processGroupId} stopped (${legacyProcessProof.reason}); stop the run to release it` + : processPidAlive && legacyProcessProof?.reason !== "process_group_alive" + ? `Lost in-memory process handle, but child pid ${run.processPid} is still alive` + : `Lost in-memory process handle, but persisted process group ${run.processGroupId} is still alive`; + // Only a run that is still running: a Stop or the stale-lock sweep may + // have finished it since this pass read it, and that outcome stands. + const detachedWrite = await setRunStatusFromLive(run.id, "running", ["running"], { error: detachedMessage, errorCode: DETACHED_PROCESS_ERROR_CODE, }); + const detachedRun = detachedWrite.updated ? detachedWrite.run : null; if (detachedRun) { await appendRunEvent(detachedRun, { eventType: "lifecycle", @@ -2692,6 +2712,14 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD }, ); if (!failureWrite.updated || !failureWrite.run) continue; + // Counted by the release that removes the pre-#19871 fallback. It never + // holds the settle back. + if (legacyProcessProof?.legacyHostFallback) { + await persistActivity(db, { companyId: run.companyId, actorType: "system", actorId: "orphan-reaper", + action: LEGACY_HOST_FALLBACK_EVENT, entityType: "heartbeat_run", entityId: run.id, runId: run.id, + agentId: run.agentId, details: { reason: "process_lost", source: "active_execution_recovery" } }) + .catch(error => logger.error({ err: error, runId: run.id }, "failed to record the legacy host fallback event")); + } let finalizedRun: typeof heartbeatRuns.$inferSelect | null = failureWrite.run; await setWakeupStatus(run.wakeupRequestId, "failed", { diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 63d9798ce47..7f27b08a211 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -104,12 +104,7 @@ import { import { visibleIssueCondition } from "../issue-visibility.js"; import { forbidden, notFound } from "../../errors.js"; import { logger } from "../../middleware/logger.js"; -import { - isPidAlive, - isProcessGroupAlive, -} from "../local-service-supervisor.js"; -import { sameProcessSpace } from "../controller-process-identity.js"; -import { ephemeralServerEnabled } from "../ephemeral-server-mode.js"; +import { proveRunProcessGone } from "../run-process-owner.js"; import { LEGACY_HOST_FALLBACK_EVENT } from "../native-runtime/native-restart-recovery.js"; import { redactSensitiveText } from "../../redaction.js"; import { isUniqueViolation } from "../../db-errors.js"; @@ -926,6 +921,13 @@ export function recoveryService( run: typeof heartbeatRuns.$inferSelect, ) => boolean; liveRunExecutions?: Readonly<{ has(id: string): boolean }>; + /** + * The one settle path for a `running` legacy run whose adapter child is + * lost: heartbeat's active-execution reaper (failed/process_lost, with its + * retry policy). The stale-lock sweep hands such runs to it instead of + * judging process death itself. + */ + settleLostRunProcess?: (scope: { companyId: string; runId: string; onDeferredDeadline?: (at: number) => void }) => Promise; beforeOrphanedRunTerminalWrite?: (runId: string) => Promise; beforeStaleLockClear?: (issueId: string) => Promise; }, @@ -6094,9 +6096,12 @@ export function recoveryService( // authority that catches the reuse-lease path: the release stops the sandbox // but keeps the server process alive, so the in-memory handle and the // recorded pid can both persist. - // - Process-death authority: the run has no in-memory handle and its recorded - // process and process group are both gone. This catches a hard server crash - // that skipped the graceful teardown, even when the issue is not terminal. + // - Process death: the run has no in-memory handle and the shared owner + // prover finds its recorded child gone. A legacy run is handed to the one + // lost-process settle path (heartbeat's active-execution reaper), so it + // gets the same failed/process_lost outcome and retry as any other lost + // child. A native run keeps this backstop's interrupted outcome; native + // recovery owns its retries. async function terminalizeOrphanedRunningRun( run: typeof heartbeatRuns.$inferSelect, options?: { @@ -6182,24 +6187,28 @@ export function recoveryService( // group. Require recorded process metadata, so this authority never fires // on a run that has not yet stored its pid. let processGone = false; + let legacyHostFallback = false; const hasLiveExecution = deps.liveRunExecutions?.has(run.id) ?? runningProcesses.has(run.id); - // A recorded PID is judged only in the PID space that recorded it: a dead - // local PID never proves that a process on another host, in another - // namespace, or in a sandbox ("remote") is gone. Rows saved before the - // space was recorded keep the previous local proof for one release, and - // only on a single-host server: an ephemeral server cannot assume the - // row was written here. Its operator path is the run's Stop control. - const legacyHostFallback = run.processHostHash == null && !ephemeralServerEnabled(); - const processSpaceJudgeable = legacyHostFallback || sameProcessSpace(run.processHostHash); - if (!hasLiveExecution && processSpaceJudgeable) { - if (typeof pid === "number" || typeof processGroupId === "number") { - const processAlive = - (typeof pid === "number" && isPidAlive(pid)) || - (typeof processGroupId === "number" && - isProcessGroupAlive(processGroupId)); - processGone = !processAlive; - } + if (!issueTerminalStatus && !hasLiveExecution && run.runtimeMode !== "native") { + // The lost-process settle path proves the child itself and owns the + // outcome. Without it (a bare recovery service) nothing is judged here. + if (!deps.settleLostRunProcess) return { terminalized: false, status: run.status }; + await deps.settleLostRunProcess({ companyId: run.companyId, runId: run.id, + onDeferredDeadline: options?.onDeferredDeadline }); + const [current] = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id))); + const status = current?.status ?? run.status; + return { terminalized: status !== run.status && TERMINAL_HEARTBEAT_RUN_STATUSES.has(status), status }; + } + // The shared owner prover judges a recorded PID only in the PID space that + // recorded it (#19871). Rows saved before the space was recorded keep the + // previous local proof for one release, on single-host servers only. An + // unprovable child is never released here; its operator path is Stop. + if (!hasLiveExecution) { + const proof = await proveRunProcessGone(run, "orphaned_running_run"); + processGone = proof?.verdict === "gone"; + legacyHostFallback = proof?.legacyHostFallback === true; } // Neither authority applies. The run is still live, so leave it alone. diff --git a/server/src/services/run-process-owner.ts b/server/src/services/run-process-owner.ts new file mode 100644 index 00000000000..6f7468fefc7 --- /dev/null +++ b/server/src/services/run-process-owner.ts @@ -0,0 +1,34 @@ +import type { heartbeatRuns } from "@paperclipai/db"; +import { proveOwnerGone, type LocalOwnerStamp, type OwnerProof, type ProcessProbe } from "./owner-proof.js"; +import { runProcessOwnerStamp } from "./native-runtime/native-owner-stamps.js"; + +/** The process identity a heartbeat run records for its adapter child + * (`persistHeartbeatRunProcessMetadata`). */ +export type RunProcessIdentity = Pick; + +/** + * The owner stamp the lost-process reapers judge a run's adapter child by. + * + * It is the shared `runProcessOwnerStamp` (native-owner-stamps.ts) with the + * one-release local proof for rows saved before `process_host_hash` (#19871, + * #19883; single-host servers only). A row that recorded only a process group + * is judged by that group: its leader's PID is the group id. The active + * execution reaper, the orphan sweep and hot-restart adoption all read it, so + * they reach one verdict for one row. + * + * Returns null when the run recorded no process. + */ +export function lostProcessOwnerStamp(run: RunProcessIdentity, scope: string): LocalOwnerStamp | null { + const pid = run.processPid ?? run.processGroupId ?? null; + if (pid === null) return null; + return runProcessOwnerStamp({ ...run, processPid: pid }, scope, { legacyFallback: true }); +} + +/** Proves whether a run's adapter child is gone. Null when no process was + * recorded. `unprovable` never releases the run: its operator action is Stop. */ +export async function proveRunProcessGone(run: RunProcessIdentity, scope: string, + probe?: ProcessProbe): Promise { + const stamp = lostProcessOwnerStamp(run, scope); + return stamp ? proveOwnerGone(stamp, probe) : null; +}