mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix(recovery): collapse the two lost-process reapers onto one stamp and the shared owner prover (#20004)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - When the control plane restarts, it must decide whether each running agent's process is gone before it releases the run > - Two reapers made that decision for legacy runs, with different rules and different outcomes. One also waited 5 minutes even when it could already prove the process was dead (RC-27) > - The durable ownership plan moves every such decision onto one prover, `proveOwnerGone`, and never releases work by age alone > - This pull request collapses the two reapers onto one owner stamp and the shared prover, removes the age wait for a proven-dead process, and never releases a process it cannot prove gone > - The benefit is that a lost process ends one way, at once, with its normal retry, and a process on another host is never released by mistake ## Linked Issues or Issue Description Refs #19831 (plan, Phase 2 sections A and C), #19838 (inventory: "One fact, many proofs" and the age-only list), #19871, #19883, #19953, #19997. **What happened?** - `recoverActiveExecutions` (heartbeat) and `terminalizeOrphanedRunningRun` (recovery service) both judged a lost legacy child. They used different liveness rules (EPERM counted as alive in one, dead in the other) and gave different outcomes: `failed/process_lost` with a retry in one, `interrupted/orphaned_running_run` with no retry in the other. - The active-execution reaper kept a 5-minute age guard even when the child was provably dead. - Hot-restart adoption judged the child with raw PID checks. **Expected behavior** One prover decides. A child that is proven gone is settled at once, through one path with one outcome. An unprovable child (another PID space, a sandbox, or a pre-#19871 row on an ephemeral server) is never released, and Stop is its operator action. ## What Changed - New `server/src/services/run-process-owner.ts`: `lostProcessOwnerStamp` and `proveRunProcessGone`. They wrap the shared `runProcessOwnerStamp` (#19997) with the one-release legacy fallback (#19883) and the group-leader case. - `heartbeat/recovery.ts`: - The active-execution reaper judges a legacy child with the shared prover. A child proven gone skips the age guard; master's exemption for a proven-gone controller (#19953) is kept. - An unprovable child gets the detached marker, whose message says to stop the run. - The legacy host fallback is recorded as `LEGACY_HOST_FALLBACK_EVENT`. - `heartbeat/recovery.ts` (hot-restart adoption): a run is marked lost only on a `gone` proof. An unprovable child stays adopted and held. - `recovery/service.ts`: the stale-lock sweep no longer judges a legacy child. It hands the run to `settleLostRunProcess`, which is the active-execution reaper wired in `heartbeat.ts`, and then reads the outcome back. Native runs keep the sweep's own backstop, now judged with the shared prover. Salvaged from the stopped Phase 2 recovery agent's uncommitted work, which had no tests. I completed and corrected it: the salvaged adoption marked unprovable children lost, which risked duplicate execution. ## Verification - `recovery-stale-issue-lock-sweep.test.ts` (51 pass). Three new tests fail on `master` and pass here: - "hands a legacy run whose child is proven gone to the one lost-process reaper"; - "settles a legacy run whose child is proven gone without waiting out the age guard"; - "never releases a legacy run whose child is unprovable, and names Stop as the way out". Updated tests: - The PID-space test now expects `failed` for the one legacy outcome. - The explicit-retry race now injects `settleLostRunProcess`. - The two tests of the sweep's own terminal write use native runs, which keep that path. - `heartbeat-process-recovery.test.ts`: 418 of 419 pass. The one failure, "redacts opaque environment-bound credentials from Sentry diagnostics", also fails on unchanged `master`. - `heartbeat/recovery.test.ts` (97), `ownership-reconciliation-report.test.ts` (24), `native-remote-retry-admission.test.ts` (92) and `heartbeat-provider-limit-wait.test.ts` (4) pass. - The server package typecheck passes. ## Risks - Behavior change: a legacy run whose child died, when the orphan sweep finds it, now ends `failed/process_lost` with the reaper's retry policy, instead of `interrupted/orphaned_running_run`. - A pre-#19871 row on an ephemeral server, or a child in another PID space, now stays `running` with a detached marker until Stop. It is no longer released by age. - Not changed: a dispatched legacy run that recorded no PID and has no controller stamp keeps its lease-expiry plus age fence. Rows written since #19953 are judged by the controller proof. Older rows, and remote handoff, keep the fence. A different host is never proof, so removing the fence would strand runs after an ephemeral replacement. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
946c098bf3
commit
7c508d3e42
7 files changed
+207
-61
No files matched your search
@@ -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 },
|
||||
});
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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", {
|
||||
|
||||
@@ -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<unknown>;
|
||||
beforeOrphanedRunTerminalWrite?: (runId: string) => Promise<void>;
|
||||
beforeStaleLockClear?: (issueId: string) => Promise<void>;
|
||||
},
|
||||
@@ -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.
|
||||
|
||||
@@ -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<typeof heartbeatRuns.$inferSelect,
|
||||
"processPid" | "processGroupId" | "processStartedAt" | "processHostHash">;
|
||||
|
||||
/**
|
||||
* 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<OwnerProof | null> {
|
||||
const stamp = lostProcessOwnerStamp(run, scope);
|
||||
return stamp ? proveOwnerGone(stamp, probe) : null;
|
||||
}
|
||||
Reference in new issue
Block a user