From 27812e76abe9887a4d7a60995de66eaa15099ade Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 26 Aug 2026 20:43:14 +0000 Subject: [PATCH] fix(runtime): retain inconclusive provider ownership Classify provider identity checks as owned, foreign, empty, or inconclusive. Keep durable metadata when validation cannot complete, and retry terminal-run cleanup on later orphan-reaper passes until the group is proven owned or stale. Co-authored-by: Codex --- .../heartbeat-process-recovery.test.ts | 58 +++++++++++++++++ server/src/services/heartbeat.ts | 62 ++++++++++++++----- 2 files changed, 106 insertions(+), 14 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 9f30430cc74..3ed328bdf95 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1401,6 +1401,64 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("retains inconclusive provider ownership for terminal cleanup retry", async () => { + const ownerToken = randomUUID(); + const guardian = spawn( + "/bin/sh", + ["-c", "sleep 3600 & echo $!; sleep 0.2"], + { + detached: true, + env: { ...process.env, PAPERCLIP_PROVIDER_OWNER_TOKEN: ownerToken }, + stdio: ["ignore", "pipe", "ignore"], + }, + ); + childProcesses.add(guardian); + expect(guardian.pid).toBeTypeOf("number"); + const descendantPid = await new Promise((resolvePid, rejectPid) => { + guardian.stdout?.once("data", (chunk: Buffer) => { + const parsedPid = Number.parseInt(chunk.toString("utf8").trim(), 10); + if (Number.isInteger(parsedPid) && parsedPid > 0) resolvePid(parsedPid); + else rejectPid(new Error("Retry descendant pid was not emitted")); + }); + guardian.once("error", rejectPid); + }); + cleanupPids.add(descendantPid); + await new Promise((resolveExit) => guardian.once("exit", () => resolveExit())); + + const { runId } = await seedRunFixture({ + processPid: 999_999_999, + providerProcessPid: guardian.pid ?? null, + providerProcessGroupId: guardian.pid ?? null, + providerProcessStartedAt: new Date("2020-01-01T00:00:00.000Z"), + providerProcessOwnerToken: null, + includeIssue: false, + }); + const heartbeat = heartbeatService(db); + + expect((await heartbeat.reapOrphanedRuns()).reaped).toBe(1); + expect(isPidAlive(descendantPid)).toBe(true); + expect(await heartbeat.getRun(runId)).toMatchObject({ + status: "failed", + providerProcessPid: guardian.pid, + providerProcessGroupId: guardian.pid, + providerProcessOwnerToken: null, + }); + + await db + .update(heartbeatRuns) + .set({ providerProcessOwnerToken: ownerToken }) + .where(eq(heartbeatRuns.id, runId)); + await heartbeat.reapOrphanedRuns(); + + expect(await waitForPidExit(descendantPid)).toBe(true); + expect(await heartbeat.getRun(runId)).toMatchObject({ + providerProcessPid: null, + providerProcessGroupId: null, + providerProcessStartedAt: null, + providerProcessOwnerToken: null, + }); + }); + it("skips generic timer wakes without invoking an adapter when no assigned work is actionable", async () => { const { companyId, agentId } = await seedIdleTimerAgentFixture(); const heartbeat = heartbeatService(db); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 2d3abec98c6..069945630c5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -3,7 +3,7 @@ import path from "node:path"; import { execFile as execFileCallback } from "node:child_process"; import { promisify } from "node:util"; import { createHash, randomUUID } from "node:crypto"; -import { and, asc, desc, eq, getTableColumns, gt, gte, inArray, isNull, lt, lte, ne, notInArray, or, sql } from "drizzle-orm"; +import { and, asc, desc, eq, getTableColumns, gt, gte, inArray, isNotNull, isNull, lt, lte, ne, notInArray, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { AGENT_DEFAULT_MAX_CONCURRENT_RUNS, @@ -6397,9 +6397,10 @@ function isProcessAlive(pid: number | null | undefined) { async function isOwnedProviderProcessGroup( processGroupId: number, ownerToken: string, -): Promise { - if (process.platform !== "linux" || !ownerToken) return false; - const entries = await fs.readdir("/proc", { withFileTypes: true }).catch(() => []); +): Promise<"owned" | "foreign" | "inconclusive" | "empty"> { + if (process.platform !== "linux" || !ownerToken) return "inconclusive"; + const entries = await fs.readdir("/proc", { withFileTypes: true }).catch(() => null); + if (!entries) return "inconclusive"; const memberPids: number[] = []; for (const entry of entries) { if (!entry.isDirectory() || !/^\d+$/.test(entry.name)) continue; @@ -6412,13 +6413,15 @@ async function isOwnedProviderProcessGroup( const observedGroupId = Number.parseInt(fields[2] ?? "", 10); if (observedGroupId === processGroupId) memberPids.push(pid); } - if (memberPids.length === 0) return false; + if (memberPids.length === 0) return "empty"; const expectedEntry = `PAPERCLIP_PROVIDER_OWNER_TOKEN=${ownerToken}`; const ownership = await Promise.all(memberPids.map(async (pid) => { const environ = await fs.readFile(`/proc/${pid}/environ`).catch(() => null); - return environ?.toString("utf8").split("\0").includes(expectedEntry) ?? false; + if (!environ) return null; + return environ.toString("utf8").split("\0").includes(expectedEntry); })); - return ownership.every(Boolean); + if (ownership.some((owned) => owned === null)) return "inconclusive"; + return ownership.every(Boolean) ? "owned" : "foreign"; } async function terminateHeartbeatRunProcess(input: { @@ -9987,28 +9990,58 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const processGroupId = run.providerProcessGroupId; const pidAlive = isProcessAlive(pid); const processGroupAlive = isProcessGroupAlive(processGroupId); - let ownsLiveProcess = false; + let identity: "owned" | "foreign" | "inconclusive" | "empty" = "empty"; if (pidAlive && pid && run.providerProcessStartedAt) { const observedStartedAt = await readProcessStartedAt(pid).catch(() => null); - ownsLiveProcess = observedStartedAt !== null - && Date.parse(observedStartedAt) === run.providerProcessStartedAt.getTime(); + identity = observedStartedAt === null + ? "inconclusive" + : Date.parse(observedStartedAt) === run.providerProcessStartedAt.getTime() + ? "owned" + : "foreign"; } else if ( processGroupAlive && processGroupId && run.providerProcessOwnerToken ) { - ownsLiveProcess = await isOwnedProviderProcessGroup( + identity = await isOwnedProviderProcessGroup( processGroupId, run.providerProcessOwnerToken, ); + } else if (pidAlive || processGroupAlive) { + identity = "inconclusive"; } - if (ownsLiveProcess) { + if (identity === "owned") { await terminateHeartbeatRunProcess({ pid, processGroupId }); } - if (pid || processGroupId) { + if (identity !== "inconclusive" && (pid || processGroupId)) { await clearRunProviderProcessMetadata(run.id); } - return ownsLiveProcess; + return identity === "owned"; + } + + async function reapTerminalProviderProcesses() { + const terminalProviderRuns = await db + .select() + .from(heartbeatRuns) + .where(and( + notInArray(heartbeatRuns.status, ["queued", "running"]), + or( + isNotNull(heartbeatRuns.providerProcessPid), + isNotNull(heartbeatRuns.providerProcessGroupId), + ), + )); + let reaped = 0; + for (const run of terminalProviderRuns) { + try { + if (await terminateRunProviderProcess(run)) reaped += 1; + } catch (error) { + logger.warn( + { runId: run.id, errorKind: "provider_process_reap_failed" }, + "provider process recovery will retry after an inconclusive cleanup", + ); + } + } + return reaped; } async function clearDetachedRunWarning(runId: string) { @@ -13757,6 +13790,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } async function reapOrphanedRuns(opts?: { staleThresholdMs?: number }) { + await reapTerminalProviderProcesses(); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date();