mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 09:41:04 +01:00
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 <codex@openai.com>
This commit is contained in:
1 parent
91bc0ec2fe
commit
27812e76ab
2 files changed
+106
-14
No files matched your search
@@ -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<number>((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<void>((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);
|
||||
|
||||
@@ -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<boolean> {
|
||||
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();
|
||||
|
||||
|
||||
Reference in new issue
Block a user