diff --git a/packages/db/src/migrations/0377_clear_settled_native_run_process_identity.sql b/packages/db/src/migrations/0377_clear_settled_native_run_process_identity.sql new file mode 100644 index 00000000000..10627bf5fb1 --- /dev/null +++ b/packages/db/src/migrations/0377_clear_settled_native_run_process_identity.sql @@ -0,0 +1,43 @@ +UPDATE "heartbeat_runs" AS "run" +SET "process_pid" = NULL, "process_group_id" = NULL, "process_started_at" = NULL +WHERE "run"."runtime_mode" = 'native' + AND "run"."status" IN ('failed', 'cancelled', 'timed_out', 'interrupted') + AND "run"."finished_at" < now() - interval '1 day' + AND ("run"."process_pid" IS NOT NULL OR "run"."process_group_id" IS NOT NULL OR "run"."process_started_at" IS NOT NULL) + AND NOT EXISTS ( + SELECT 1 FROM "native_run_finalizations" AS "coordinator" + WHERE "coordinator"."run_id" = "run"."id" + AND "coordinator"."phase" NOT IN ('committed', 'terminal_failure') + ) + AND ( + coalesce(jsonb_typeof("run"."runner_profile_json" -> 'sessionCheckpoint'), 'null') <> 'null' + OR EXISTS ( + SELECT 1 FROM "heartbeat_run_events" AS "event" + WHERE "event"."run_id" = "run"."id" + AND "event"."event_type" IN ('harness.ready', 'session.started', 'session.resumed', 'session.updated', 'turn.started', 'provider.event', 'provider.rpc_result') + ) + ) + AND EXISTS ( + SELECT 1 FROM "environment_leases" AS "lease" + WHERE "lease"."heartbeat_run_id" = "run"."id" + ) + AND NOT EXISTS ( + SELECT 1 FROM "environment_leases" AS "lease" + WHERE "lease"."heartbeat_run_id" = "run"."id" + AND NOT coalesce(( + "lease"."company_id" = "run"."company_id" + AND "lease"."provider" IS NOT NULL + AND "lease"."provider" <> 'local' + AND "lease"."provider_lease_id" IS NOT NULL + AND "lease"."released_at" IS NOT NULL + AND "lease"."cleanup_status" = 'success' + AND "lease"."status" IN ('released', 'expired', 'failed') + AND "lease"."metadata" -> 'remoteExecutionTermination' ->> 'schema' = 'paperclip.remote-termination.v1' + AND "lease"."metadata" -> 'remoteExecutionTermination' ->> 'companyId' = "lease"."company_id"::text + AND "lease"."metadata" -> 'remoteExecutionTermination' ->> 'runId' = "lease"."heartbeat_run_id"::text + AND "lease"."metadata" -> 'remoteExecutionTermination' ->> 'leaseId' = "lease"."id"::text + AND "lease"."metadata" -> 'remoteExecutionTermination' ->> 'provider' = "lease"."provider" + AND "lease"."metadata" -> 'remoteExecutionTermination' ->> 'providerLeaseId' = "lease"."provider_lease_id" + AND "lease"."metadata" -> 'remoteExecutionTermination' ->> 'state' IN ('stopped', 'destroyed') + ), false) + ); diff --git a/packages/db/src/migrations/meta/0372_snapshot.json b/packages/db/src/migrations/meta/0377_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0372_snapshot.json rename to packages/db/src/migrations/meta/0377_snapshot.json index e1aee913afc..6e89edc3a9b 100644 --- a/packages/db/src/migrations/meta/0372_snapshot.json +++ b/packages/db/src/migrations/meta/0377_snapshot.json @@ -1,6 +1,6 @@ { - "id": "9c1b4cac-52f1-4c6e-89b3-c9a7816016c7", - "prevId": "7caf727e-0913-40b3-b01c-842e9447666f", + "id": "a0fcd9b8-f125-4985-8795-f567466b9aef", + "prevId": "7832df38-fa1b-4687-bd06-eccbc29bd28c", "version": "7", "dialect": "postgresql", "tables": { @@ -28508,6 +28508,13 @@ "primaryKey": false, "notNull": true }, + "checkout_name": { + "name": "checkout_name", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "''" + }, "object_key": { "name": "object_key", "type": "text", @@ -28601,8 +28608,8 @@ "method": "btree", "with": {} }, - "execution_workspace_restore_points_live_generation_uniq": { - "name": "execution_workspace_restore_points_live_generation_uniq", + "execution_workspace_restore_points_live_status_uniq": { + "name": "execution_workspace_restore_points_live_status_uniq", "columns": [ { "expression": "execution_workspace_id", @@ -28615,10 +28622,16 @@ "isExpression": false, "asc": true, "nulls": "last" + }, + { + "expression": "checkout_name", + "isExpression": false, + "asc": true, + "nulls": "last" } ], "isUnique": true, - "where": "\"execution_workspace_restore_points\".\"status\" <> 'deleted'", + "where": "\"execution_workspace_restore_points\".\"status\" in ('pending', 'active', 'expiring')", "concurrently": false, "method": "btree", "with": {} @@ -38858,6 +38871,18 @@ "primaryKey": false, "notNull": false }, + "dismissed_at": { + "name": "dismissed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "dismissed_by_user_id": { + "name": "dismissed_by_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "created_at": { "name": "created_at", "type": "timestamp with time zone", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 2d3b7e1d5b0..c69031562fa 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2626,6 +2626,13 @@ "when": 1792783857131, "tag": "0376_clear_committed_native_recovery_state", "breakpoints": true + }, + { + "idx": 377, + "version": "7", + "when": 1792790134086, + "tag": "0377_clear_settled_native_run_process_identity", + "breakpoints": true } ] } \ No newline at end of file diff --git a/server/src/__tests__/native-finalization-migration.test.ts b/server/src/__tests__/native-finalization-migration.test.ts index 050be89dfa5..66cd295a8b9 100644 --- a/server/src/__tests__/native-finalization-migration.test.ts +++ b/server/src/__tests__/native-finalization-migration.test.ts @@ -7,12 +7,14 @@ import { applyPendingMigrations, companies, createDb, + environmentLeases, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; describe("P6-18 / MIG-01..04 native finalization migration", () => { it("repairs only later duplicates and preserves legacy event bytes and cursors", async () => { @@ -173,4 +175,86 @@ describe("P6-18 / MIG-01..04 native finalization migration", () => { await temporary.cleanup(); } }, 60_000); + it("clears process identity only on old settled native runs whose sandbox confirmed termination", async () => { + const temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-process-backfill-"); + const migration = await readFile( + new URL("../../../packages/db/src/migrations/0377_clear_settled_native_run_process_identity.sql", import.meta.url), + "utf8", + ); + const db = createDb(temporary.connectionString); + try { + const [company] = await db.insert(companies).values({ name: "Process backfill", issuePrefix: "PBF" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company!.id, name: "Process backfill agent" }).returning(); + const [issue] = await db.insert(issues).values({ companyId: company!.id, title: "Process backfill", status: "in_progress" }).returning(); + const old = new Date(Date.now() - 2 * 24 * 60 * 60 * 1000); + const checkpoint = { sessionCheckpoint: { providerSessionId: "thread" } }; + type Fixture = { + key: string; status?: string; runtimeMode?: string; finishedAt?: Date; phase?: string | null; + profile?: Record; providerEvent?: boolean; + leases?: Array<"stopped" | "unconfirmed" | "local">; + }; + const fixtures: Fixture[] = [ + { key: "failed" }, + { key: "cancelled", status: "cancelled" }, + { key: "timed_out", status: "timed_out", phase: "terminal_failure" }, + { key: "interrupted", status: "interrupted", phase: null }, + { key: "provider_event", profile: {}, providerEvent: true }, + { key: "recent", finishedAt: new Date(Date.now() - 60 * 60 * 1000) }, + { key: "succeeded", status: "succeeded" }, + { key: "running", status: "running" }, + { key: "legacy", runtimeMode: "legacy" }, + { key: "retryable", phase: "retryable_failure" }, + { key: "finalizing", phase: "workspace_finalizing" }, + { key: "no_evidence", profile: {} }, + { key: "unconfirmed", leases: ["stopped", "unconfirmed"] }, + { key: "local", leases: ["local"] }, + { key: "no_lease", leases: [] }, + ]; + const runIds = new Map(); + for (const fixture of fixtures) { + const [run] = await db.insert(heartbeatRuns).values({ + companyId: company!.id, agentId: agent!.id, status: fixture.status ?? "failed", + runtimeMode: fixture.runtimeMode ?? "native", nativeIssueId: issue!.id, + finishedAt: fixture.finishedAt ?? old, runnerProfileJson: fixture.profile ?? checkpoint, + processPid: 4242, processGroupId: 4242, processStartedAt: old, + }).returning(); + runIds.set(fixture.key, run!.id); + if (fixture.phase !== null) { + await db.insert(nativeRunFinalizations).values({ + runId: run!.id, companyId: company!.id, issueId: issue!.id, phase: fixture.phase ?? "committed", + }); + } + if (fixture.providerEvent) { + await db.insert(heartbeatRunEvents).values({ + companyId: company!.id, runId: run!.id, agentId: agent!.id, seq: 1, eventType: "session.started", + }); + } + for (const kind of fixture.leases ?? ["stopped"]) { + const [lease] = await db.insert(environmentLeases).values({ + companyId: company!.id, issueId: issue!.id, heartbeatRunId: run!.id, + provider: kind === "local" ? "local" : "sandbox-fixture", + providerLeaseId: kind === "local" ? null : `${fixture.key}-${kind}`, + status: "released", releasedAt: old, cleanupStatus: "success", + }).returning(); + if (kind === "stopped") { + await db.update(environmentLeases).set({ metadata: { + remoteExecutionTermination: remoteTerminationReceipt(lease!, { providerLeaseId: lease!.providerLeaseId, state: "destroyed" }), + } }).where(eq(environmentLeases.id, lease!.id)); + } + } + } + for (const statement of migration.split("--> statement-breakpoint")) await db.execute(sql.raw(statement)); + const rows = new Map((await db.select().from(heartbeatRuns)).map((row) => [row.id, row])); + const cleared = new Set(["failed", "cancelled", "timed_out", "interrupted", "provider_event"]); + for (const fixture of fixtures) { + const row = rows.get(runIds.get(fixture.key)!)!; + expect({ key: fixture.key, processPid: row.processPid, processGroupId: row.processGroupId, startedAt: row.processStartedAt === null }) + .toEqual(cleared.has(fixture.key) + ? { key: fixture.key, processPid: null, processGroupId: null, startedAt: true } + : { key: fixture.key, processPid: 4242, processGroupId: 4242, startedAt: false }); + } + } finally { + await temporary.cleanup(); + } + }, 60_000); }); diff --git a/server/src/__tests__/native-settled-process-release.test.ts b/server/src/__tests__/native-settled-process-release.test.ts new file mode 100644 index 00000000000..be5c00a4516 --- /dev/null +++ b/server/src/__tests__/native-settled-process-release.test.ts @@ -0,0 +1,287 @@ +import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; +import { randomUUID } from "node:crypto"; +import { spawn } from "node:child_process"; +import os from "node:os"; +import { and, eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agents, + authUsers, + companies, + companyMemberships, + createDb, + environmentLeases, + heartbeatRunEvents, + heartbeatRuns, + issues, + nativeRunFinalizations, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; +import { persistHeartbeatRunProcessMetadata } from "../services/heartbeat/run-lifecycle.js"; +import { + hasNativeLocalProcessStop, + readNativeLocalProcessStop, + discoverSettledNativeRunProcesses, + pendingSettledNativeRunProcessIds, + releaseSettledNativeRunProcess, + retryPendingSettledNativeRunProcesses, + settleNativeRunProcess, + withReleasedNativeRunProcess, +} from "../services/native-local-process-stop.js"; +import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; +import { heartbeatService } from "../services/heartbeat.js"; + +const execute = vi.hoisted(() => vi.fn()); +vi.mock("../adapters/index.js", async () => { + const actual = await vi.importActual("../adapters/index.js"); + return { ...actual, getServerAdapter: vi.fn(() => ({ supportsLocalAgentJwt: false, execute })) }; +}); + +const support = await getEmbeddedPostgresTestSupport(); +const suite = support.supported ? describe : describe.skip; + +function startDetached() { + const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { detached: true, stdio: "ignore" }); + child.unref(); + return child; +} + +async function exitedPid() { + const child = startDetached(); + const pid = child.pid!; + await new Promise(resolve => { child.once("exit", () => resolve()); child.kill("SIGKILL"); }); + return pid; +} + +suite("settled native run process identity", () => { + let db: ReturnType; + let temporary: Awaited>; + let companyId: string; + let agentId: string; + const children: ReturnType[] = []; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("native-settled-process-"); + db = createDb(temporary.connectionString); + companyId = randomUUID(); + agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Settled processes", issuePrefix: "SPR", defaultResponsibleUserId: "owner" }); + await db.insert(authUsers).values({ id: "owner", name: "Fixture user", email: "owner@example.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "owner", status: "active", membershipRole: "member" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", role: "engineer", status: "idle", adapterType: "process", + adapterConfig: {}, permissions: {}, runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true, maxConcurrentRuns: 20 } } }); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + + afterEach(() => { + for (const child of children.splice(0)) { + try { process.kill(-child.pid!, "SIGKILL"); } catch { } + } + }); + + afterAll(async () => { + await db?.$client.end({ timeout: 0 }); + await temporary?.cleanup(); + }); + + type Fixture = { + status?: string; + phase?: string | null; + pid?: number; + hostname?: string; + checkpoint?: boolean; + providerEvidence?: boolean; + leases?: Array<"local" | "local_unreleased" | "remote_stopped" | "remote_unconfirmed">; + finishedAt?: Date; + }; + + async function seed(fixture: Fixture = {}) { + const issueId = randomUUID(); + const runId = randomUUID(); + const pid = fixture.pid ?? await exitedPid(); + await db.insert(issues).values({ id: issueId, companyId, title: "Settled process", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ + id: runId, companyId, agentId, status: "running", runtimeMode: "native", nativeIssueId: issueId, + invocationSource: "assignment", triggerDetail: "system", contextSnapshot: { issueId }, + runnerProfileJson: fixture.checkpoint === false ? {} : { sessionCheckpoint: { providerSessionId: "thread" } }, + }); + if (fixture.hostname) { + await db.update(heartbeatRuns).set({ processPid: pid, processGroupId: pid, processStartedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + await appendHeartbeatRunEvent(db, { companyId, runId, agentId, eventType: "native.process_identity_recorded", stream: "system", level: "info", + message: "Process identity recorded", payload: { targetKind: "local", hostname: fixture.hostname, processPid: pid, processGroupId: pid } }); + } else { + await persistHeartbeatRunProcessMetadata(db, runId, { pid, processGroupId: pid, startedAt: new Date().toISOString(), targetKind: "local" }); + } + if (fixture.providerEvidence) { + await appendHeartbeatRunEvent(db, { companyId, runId, agentId, eventType: "harness.ready", stream: "system", level: "info", message: "ready" }); + } + if (fixture.phase !== null) { + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: fixture.phase ?? "committed" }); + } + for (const kind of fixture.leases ?? ["local"]) { + const remote = kind.startsWith("remote"); + const [lease] = await db.insert(environmentLeases).values({ + companyId, issueId, heartbeatRunId: runId, provider: remote ? "sandbox-fixture" : "local", + providerLeaseId: remote ? randomUUID() : null, leasePolicy: "ephemeral", + status: kind === "local_unreleased" ? "active" : "released", releasedAt: kind === "local_unreleased" ? null : new Date(), + cleanupStatus: kind === "local_unreleased" ? null : "success", + }).returning(); + if (kind === "remote_stopped") { + await db.update(environmentLeases).set({ metadata: { + remoteExecutionTermination: remoteTerminationReceipt(lease!, { providerLeaseId: lease!.providerLeaseId, state: "stopped" }), + } }).where(eq(environmentLeases.id, lease!.id)); + } + } + await db.update(heartbeatRuns).set({ status: fixture.status ?? "failed", finishedAt: fixture.finishedAt ?? new Date() }) + .where(eq(heartbeatRuns.id, runId)); + return { runId, issueId, pid }; + } + + const readRun = (runId: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then(rows => rows[0]!); + + it("releases a failed native run's local identity with a stop receipt once its process is gone", async () => { + const { runId, pid } = await seed(); + expect(await readRun(runId)).toMatchObject({ processPid: pid, processGroupId: pid }); + const before = await readRun(runId); + expect(await releaseSettledNativeRunProcess(db, runId)).toBe("released"); + const after = await readRun(runId); + expect(after).toMatchObject({ status: "failed", processPid: null, processGroupId: null, processStartedAt: null }); + expect(await hasNativeLocalProcessStop(db, companyId, runId)).toBe(true); + expect(await readNativeLocalProcessStop(db, companyId, runId)).toEqual({ + processPid: pid, processGroupId: pid, processStartedAt: before.processStartedAt!.toISOString(), + }); + expect(await withReleasedNativeRunProcess(db, after)).toMatchObject({ + processPid: pid, processGroupId: pid, processStartedAt: before.processStartedAt, + }); + expect(await releaseSettledNativeRunProcess(db, runId)).toBe("skipped"); + }); + + it("releases the identity from the periodic recovery sweep", async () => { + const { runId } = await seed({ status: "timed_out" }); + const child = startDetached(); + children.push(child); + const alive = await seed({ pid: child.pid! }); + const heartbeat = heartbeatService(db); + await heartbeat.reapOrphanedRuns(); + expect((await readRun(runId)).processPid).toBeNull(); + expect(await hasNativeLocalProcessStop(db, companyId, runId)).toBe(true); + expect((await readRun(alive.runId)).processPid).toBe(child.pid); + await new Promise(resolve => { child.once("exit", () => resolve()); process.kill(-child.pid!, "SIGKILL"); }); + await heartbeat.reapOrphanedRuns(); + expect((await readRun(alive.runId)).processPid).toBeNull(); + }, 30_000); + + it.each([ + ["failed", true], + ["succeeded", false], + ] as const)("releases a %s native run's identity when its executor settles: %s", async (status, released) => { + const pid = await exitedPid(); + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Executor settlement", status: "todo", assigneeAgentId: agentId, responsibleUserId: "owner" }); + let runId = ""; + execute.mockReset().mockImplementation(async (context: { runId: string }) => { + if (runId) return { exitCode: 0, signal: null, timedOut: false, summary: "Follow-up" }; + runId = context.runId; + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, + runnerProfileJson: { sessionCheckpoint: { providerSessionId: "thread" } } }).where(eq(heartbeatRuns.id, runId)); + await persistHeartbeatRunProcessMetadata(db, runId, { pid, processGroupId: pid, startedAt: new Date().toISOString(), targetKind: "local" }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "committed" }); + await db.update(heartbeatRuns).set({ status, finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + return { exitCode: status === "failed" ? 1 : 0, signal: null, timedOut: false, summary: "Settled" }; + }); + const heartbeat = heartbeatService(db); + await heartbeat.invoke(agentId, "on_demand", { issueId }, "manual"); + await heartbeat.drainActiveRunExecutions(); + const run = await readRun(runId); + expect(run.status).toBe(status); + expect(run.processPid).toBe(released ? null : pid); + expect(await hasNativeLocalProcessStop(db, companyId, runId)).toBe(released); + const events = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, runId), + eq(heartbeatRunEvents.eventType, "native.local_process_stopped"))); + expect(events).toHaveLength(released ? 1 : 0); + }, 30_000); + + it.each(["cancelled", "timed_out", "interrupted"])("releases a settled %s native run", async status => { + const { runId } = await seed({ status }); + expect(await releaseSettledNativeRunProcess(db, runId)).toBe("released"); + expect((await readRun(runId)).processPid).toBeNull(); + }); + + it("keeps the identity while the process is still alive, then releases it after exit", async () => { + const child = startDetached(); + children.push(child); + const { runId } = await seed({ pid: child.pid! }); + expect(await releaseSettledNativeRunProcess(db, runId)).toBe("pending"); + expect((await readRun(runId)).processPid).toBe(child.pid); + expect(await hasNativeLocalProcessStop(db, companyId, runId)).toBe(false); + await new Promise(resolve => { child.once("exit", () => resolve()); process.kill(-child.pid!, "SIGKILL"); }); + expect(await releaseSettledNativeRunProcess(db, runId)).toBe("released"); + expect((await readRun(runId)).processPid).toBeNull(); + }); + + it.each([ + ["a succeeded run that can hand its runner to the next turn", { status: "succeeded" }, "skipped"], + ["a same-run retry that still owns the process identity", { phase: "retryable_failure" }, "skipped"], + ["unfinished workspace finalization", { phase: "workspace_finalizing" }, "skipped"], + ["an unreleased local lease", { leases: ["local_unreleased"] }, "pending"], + ["a process recorded by another host", { hostname: `${os.hostname()}-other` }, "skipped"], + ["a launch without provider evidence or checkpoint", { checkpoint: false }, "skipped"], + ["a sandbox without a termination receipt", { leases: ["remote_unconfirmed"] }, "pending"], + ["a mixed local and sandbox run", { leases: ["local", "remote_stopped"] }, "pending"], + ] as const)("keeps the identity for %s", async (_name, fixture, outcome) => { + const { runId, pid } = await seed(fixture as Fixture); + expect(await releaseSettledNativeRunProcess(db, runId)).toBe(outcome); + expect(await readRun(runId)).toMatchObject({ processPid: pid, processGroupId: pid }); + }); + + it("releases evidence-only launches and runs without a coordinator", async () => { + const evidence = await seed({ checkpoint: false, providerEvidence: true }); + const uncoordinated = await seed({ phase: null }); + expect(await releaseSettledNativeRunProcess(db, evidence.runId)).toBe("released"); + expect(await releaseSettledNativeRunProcess(db, uncoordinated.runId)).toBe("released"); + }); + + it("releases a sandbox identity only after every lease confirmed termination", async () => { + const { runId } = await seed({ leases: ["remote_stopped"] }); + expect(await releaseSettledNativeRunProcess(db, runId)).toBe("released"); + expect((await readRun(runId)).processPid).toBeNull(); + expect(await hasNativeLocalProcessStop(db, companyId, runId)).toBe(false); + }); + + it("discovers settled runs in pages, skips live executions, and retries only pending releases", async () => { + const first = await seed(); + const skipped = await seed(); + const child = startDetached(); + children.push(child); + const alive = await seed({ pid: child.pid! }); + await discoverSettledNativeRunProcesses(db, { limit: 1, skip: runId => runId === skipped.runId }); + expect((await readRun(first.runId)).processPid).toBeNull(); + expect((await readRun(skipped.runId)).processPid).toBe(skipped.pid); + expect((await readRun(alive.runId)).processPid).toBe(child.pid); + expect(pendingSettledNativeRunProcessIds()).toContain(alive.runId); + expect(pendingSettledNativeRunProcessIds()).not.toContain(skipped.runId); + await new Promise(resolve => { child.once("exit", () => resolve()); process.kill(-child.pid!, "SIGKILL"); }); + await retryPendingSettledNativeRunProcesses(db); + expect((await readRun(alive.runId)).processPid).toBeNull(); + expect(pendingSettledNativeRunProcessIds()).not.toContain(alive.runId); + await settleNativeRunProcess(db, skipped.runId); + expect(pendingSettledNativeRunProcessIds()).not.toContain(skipped.runId); + for (let attempt = 0; attempt < 25 && pendingSettledNativeRunProcessIds().length; attempt += 1) { + await retryPendingSettledNativeRunProcesses(db); + } + expect(pendingSettledNativeRunProcessIds()).toEqual([]); + const idle = new Proxy({}, { get: () => { throw new Error("idle retry touched the database"); } }) as typeof db; + await retryPendingSettledNativeRunProcesses(idle); + }); + + it("stops retrying a release that never becomes ready", async () => { + const { runId } = await seed({ leases: ["local_unreleased"] }); + expect(await settleNativeRunProcess(db, runId)).toBe("pending"); + expect(pendingSettledNativeRunProcessIds()).toContain(runId); + for (let attempt = 0; attempt < 18; attempt += 1) await retryPendingSettledNativeRunProcesses(db); + expect(pendingSettledNativeRunProcessIds()).toContain(runId); + await retryPendingSettledNativeRunProcesses(db); + expect(pendingSettledNativeRunProcessIds()).not.toContain(runId); + expect((await readRun(runId)).processPid).not.toBeNull(); + }); +}); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index c405ab8a1e5..99060a28540 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -252,6 +252,7 @@ import { import { withAdapterExecutionPhase } from "@paperclipai/adapter-utils/execution-phase"; import { recordNativeLocalProcessStop, + settleNativeRunProcess, } from "./native-local-process-stop.js"; import { @@ -3850,6 +3851,12 @@ export function heartbeatService( ${JSON.stringify({ startupPreparationSettledAt: new Date().toISOString() })}::jsonb`, }).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "cancelled"))); } + if (latestRun?.runtimeMode === "native" && isHeartbeatRunTerminalStatus(latestRun.status) && + !nativeSessionResumeScheduled && !nativeWorkspaceFinalizeScheduled && !nativeOwnershipHeld) { + await settleNativeRunProcess(db, run.id).catch(error => { + logger.warn({ err: error, runId: run.id }, "failed to release settled native process identity"); + }); + } await db.update(heartbeatRuns).set({ executionStage: sql`case when ${heartbeatRuns.status} = 'cancelled' then 'settled' else ${heartbeatRuns.executionStage} end`, controllerLeaseExpiresAt: null, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 8964c628b94..2d32030510e 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -8,7 +8,7 @@ import { hasStopOnlyCleanup, settleStopOnlyCleanup, } from "../sandbox-stop-and-retain.js"; -import { recordNativeLocalProcessStop } from "../native-local-process-stop.js"; +import { discoverSettledNativeRunProcesses, recordNativeLocalProcessStop, retryPendingSettledNativeRunProcesses } from "../native-local-process-stop.js"; import { pendingCaseHandoffForRun } from "./case-handoff-lifecycle.js"; import { legacyControllerBootId, @@ -1920,6 +1920,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD void cleanup.finally(() => activeRunExecutionPromises.delete(cleanup)); } + let settledProcessesDiscovered = false; + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number }) { await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; @@ -2051,6 +2053,18 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD if (cleared) claimableNativeRunIds.add(cleared.id); } + const executingRun = (runId: string) => activeRunExecutions.has(runId) || runningProcesses.has(runId); + if (!settledProcessesDiscovered) { + settledProcessesDiscovered = await discoverSettledNativeRunProcesses(db, { skip: executingRun }).then(() => true, (error) => { + logger.warn({ err: error }, "failed to discover settled native process identities"); + return false; + }); + } else { + await retryPendingSettledNativeRunProcesses(db, { skip: executingRun }).catch((error) => { + logger.warn({ err: error }, "failed to retry settled native process identity release"); + }); + } + // An explicit result-less retryable failure resumes on the original run. // The database lease is claimed before dispatch so concurrent service // instances cannot open competing recoveries; executeRun receives the exact diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 235aeb878cc..7e74b8eca1c 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -1,11 +1,15 @@ -import { and, desc, eq, inArray, isNull } from "drizzle-orm"; -import { environmentLeases, heartbeatRunEvents, heartbeatRuns, type Db } from "@paperclipai/db"; +import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, or } from "drizzle-orm"; +import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; +import { nativeSessionProviderEvidence } from "./native-runtime/native-session-resume.js"; +import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import os from "node:os"; export const PROCESS_START_REQUESTED = "native.process_start_requested"; export const PROCESS_IDENTITY_RECORDED = "native.process_identity_recorded"; const LOCAL_PROCESS_STOPPED = "native.local_process_stopped"; +const SETTLED_RUN_STATUSES = ["failed", "cancelled", "timed_out", "interrupted"]; +const SETTLED_COORDINATOR_PHASES = ["committed", "terminal_failure"]; function absent(pid: number): boolean { try { @@ -55,7 +59,8 @@ export async function recordNativeLocalProcessStop(db: Db, run: typeof heartbeat stream: "system", level: "info", message: "Recovery verified the local process stopped before clearing its identity.", - payload: { processPid: run.processPid, processGroupId: run.processGroupId }, + payload: { processPid: run.processPid, processGroupId: run.processGroupId, + processStartedAt: run.processStartedAt?.toISOString() ?? null }, }); return true; } @@ -106,6 +111,16 @@ export async function hasHistoricalSuspendedNativeSession(db: Db, run: typeof he }); } +export async function withReleasedNativeRunProcess(db: Db, run: T): Promise { + if (run.processPid !== null || run.processGroupId !== null || run.processStartedAt !== null) return run; + const stopped = await readNativeLocalProcessStop(db, run.companyId, run.id); + if (!stopped) return run; + return { ...run, processPid: stopped.processPid, processGroupId: stopped.processGroupId, + processStartedAt: stopped.processStartedAt ? new Date(stopped.processStartedAt) : null }; +} + /** Recover the exact stopped identity after the mutable run fields were cleared. */ export async function readNativeLocalProcessStop(db: Db, companyId: string, runId: string) { const [event] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload }) @@ -118,5 +133,90 @@ export async function readNativeLocalProcessStop(db: Db, companyId: string, runI const group = event?.payload?.processGroupId; if (event?.eventType !== LOCAL_PROCESS_STOPPED || typeof pid !== "number" || !Number.isSafeInteger(pid) || pid <= 1 || group !== pid || !absent(pid) || !absent(-pid)) return null; - return { processPid: pid, processGroupId: pid }; + const startedAt = event.payload?.processStartedAt; + return { processPid: pid, processGroupId: pid, + ...(typeof startedAt === "string" ? { processStartedAt: startedAt } : {}) }; +} + +type SettledProcessRelease = "released" | "pending" | "skipped"; + +export async function releaseSettledNativeRunProcess(db: Db, runId: string): Promise { + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update").limit(1); + if (!run || run.runtimeMode !== "native" || !SETTLED_RUN_STATUSES.includes(run.status) || !run.finishedAt || + (run.processPid === null && run.processGroupId === null && run.processStartedAt === null)) return "skipped"; + const [coordinator] = await tx.select({ phase: nativeRunFinalizations.phase }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, run.id)).limit(1); + if (coordinator && !SETTLED_COORDINATOR_PHASES.includes(coordinator.phase)) return "skipped"; + if (run.runnerProfileJson?.sessionCheckpoint == null && + !(await nativeSessionProviderEvidence(tx, [run.id])).has(run.id)) return "skipped"; + const leases = await tx.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + )); + if (leases.some(lease => lease.provider !== "local")) { + if (!leases.every(hasRemoteTerminationReceipt)) return "pending"; + } else { + if (leases.some(lease => !lease.releasedAt || lease.status === "pending_cleanup" || lease.cleanupStatus === "failed")) return "pending"; + const [identity] = await tx.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + eq(heartbeatRunEvents.eventType, PROCESS_IDENTITY_RECORDED), isNull(heartbeatRunEvents.sourceEventId), + )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); + if (identity?.payload?.targetKind !== "local" || identity.payload.hostname !== os.hostname() || + identity.payload.processPid !== run.processPid || identity.payload.processGroupId !== run.processGroupId) return "skipped"; + if (!(await recordNativeLocalProcessStop(tx, run))) return "pending"; + } + await tx.update(heartbeatRuns).set({ + processPid: null, processGroupId: null, processStartedAt: null, updatedAt: new Date(), + }).where(eq(heartbeatRuns.id, run.id)); + return "released"; + }); +} + +const MAX_SETTLED_PROCESS_RELEASE_ATTEMPTS = 20; +const pendingSettledProcessReleases = new Map(); + +export async function settleNativeRunProcess(db: Db, runId: string) { + const outcome = await releaseSettledNativeRunProcess(db, runId); + const attempts = (pendingSettledProcessReleases.get(runId) ?? 0) + 1; + if (outcome === "pending" && attempts < MAX_SETTLED_PROCESS_RELEASE_ATTEMPTS) { + pendingSettledProcessReleases.set(runId, attempts); + } else { + pendingSettledProcessReleases.delete(runId); + } + return outcome; +} + +export function pendingSettledNativeRunProcessIds() { + return [...pendingSettledProcessReleases.keys()]; +} + +export async function retryPendingSettledNativeRunProcesses(db: Db, input: { skip?: (runId: string) => boolean } = {}) { + for (const runId of pendingSettledNativeRunProcessIds()) { + if (input.skip?.(runId)) continue; + await settleNativeRunProcess(db, runId); + } +} + +export async function discoverSettledNativeRunProcesses(db: Db, input: { + limit?: number; skip?: (runId: string) => boolean; +} = {}) { + const limit = input.limit ?? 50; + let after: string | null = null; + let released = 0; + do { + const candidates: Array<{ id: string }> = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.runtimeMode, "native"), + inArray(heartbeatRuns.status, SETTLED_RUN_STATUSES), + isNotNull(heartbeatRuns.finishedAt), + or(isNotNull(heartbeatRuns.processPid), isNotNull(heartbeatRuns.processGroupId), isNotNull(heartbeatRuns.processStartedAt)), + ...(after ? [gt(heartbeatRuns.id, after)] : []), + )).orderBy(asc(heartbeatRuns.id)).limit(limit); + for (const candidate of candidates) { + if (input.skip?.(candidate.id)) continue; + if (await settleNativeRunProcess(db, candidate.id) === "released") released += 1; + } + after = candidates.length < limit ? null : candidates.at(-1)!.id; + } while (after); + return { released }; } diff --git a/server/src/services/native-runtime/native-provider-failure.integration.test.ts b/server/src/services/native-runtime/native-provider-failure.integration.test.ts index 60bf7a24ed7..36a288d8a08 100644 --- a/server/src/services/native-runtime/native-provider-failure.integration.test.ts +++ b/server/src/services/native-runtime/native-provider-failure.integration.test.ts @@ -28,6 +28,10 @@ import { NATIVE_STATUS_ARBITER_POLICY_VERSION } from "./status-arbiter.js"; import { issueThreadInteractionService } from "../issue-thread-interactions.js"; import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js"; import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js"; +import { persistHeartbeatRunProcessMetadata } from "../heartbeat/run-lifecycle.js"; +import { readNativeLocalProcessStop, releaseSettledNativeRunProcess } from "../native-local-process-stop.js"; + +const EXITED_PID = 4_194_304; describe("native completed failure diagnostics", () => { let temporary: Awaited>; @@ -273,6 +277,7 @@ describe("native completed failure diagnostics", () => { } else if (scenario === "unknown-error" || opencode) { // A lookalike diagnostic flag does not alter unknown failed-result policy. await db.update(heartbeatRuns).set({ resultJson: { nativeProviderFailure: NATIVE_MODEL_REJECTION_DIAGNOSTIC } }).where(eq(heartbeatRuns.id, runId)); + await persistHeartbeatRunProcessMetadata(db, runId, { pid: EXITED_PID, processGroupId: EXITED_PID, startedAt: new Date().toISOString(), targetKind: "local" }); await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded" }); const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, runId)); expect(decision.reasonCode).toBe("run_failed_partial_evidence_preserved"); @@ -282,6 +287,18 @@ describe("native completed failure diagnostics", () => { expect(finalRun.resultJson?.nativeTurnFailure).toEqual(turnFailure); const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); expect(wakes.filter(wake => wake.payload?.issueId === issueId)).toHaveLength(1); + if (scenario === "unknown-error") { + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + const [settled] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + expect(coordinator.phase).toBe("committed"); + expect(settled).toMatchObject({ status: "failed", processPid: EXITED_PID, processGroupId: EXITED_PID }); + await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 1_000, eventType: "harness.ready" }); + expect(await releaseSettledNativeRunProcess(db, runId)).toBe("released"); + const [released] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(released).toMatchObject({ status: "failed", processPid: null, processGroupId: null, processStartedAt: null }); + expect(await readNativeLocalProcessStop(db, companyId, runId)).toMatchObject({ processPid: EXITED_PID, processGroupId: EXITED_PID }); + } } }, ); diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index f33983f14c0..08174368731 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -3503,6 +3503,8 @@ describe("retained native cleanup activation", () => { "legacy_bad_proof", "legacy_extra_attempt", "legacy_activation", + "released_identity", + "released_identity_unproven", ])("preserves exact original evidence for %s", async (mode) => { const directory = await mkdtemp( join(tmpdir(), "paperclip-maintenance-activation-"), @@ -3745,8 +3747,17 @@ describe("retained native cleanup activation", () => { nativeSessionId: identity.normalizedSessionId, runnerInstanceId: identity.runnerInstanceId, finishedAt: new Date(), - processPid: mode.endsWith("live_owner") ? process.pid : 99_999_999, - processGroupId: mode.endsWith("live_owner") ? process.pid : 99_999_999, + processPid: mode.startsWith("released_identity") + ? null + : mode.endsWith("live_owner") + ? process.pid + : 99_999_999, + processGroupId: mode.startsWith("released_identity") + ? null + : mode.endsWith("live_owner") + ? process.pid + : 99_999_999, + processStartedAt: null, completionContractId: "contract", completionContractSha256: "sha", errorCode: mode === "provider_transport_failed" ? "provider_transport_failed" : "adapter_failed", @@ -3770,6 +3781,7 @@ describe("retained native cleanup activation", () => { recoveryHistory: [], }; let transactionOpen = false; + let stopReceiptRead = false; let releaseCommit!: () => void; const commitGate = new Promise((release) => { releaseCommit = release; @@ -3788,11 +3800,25 @@ describe("retained native cleanup activation", () => { : table === nativeRunResults ? [accepted] : table === heartbeatRunEvents - ? [identityRow] + ? mode.startsWith("released_identity") && !stopReceiptRead + ? ((stopReceiptRead = true), + mode === "released_identity" + ? [ + { + eventType: "native.local_process_stopped", + payload: { + processPid: 99_999_999, + processGroupId: 99_999_999, + }, + }, + ] + : []) + : [identityRow] : []; const query = { where: () => query, for: () => query, + orderBy: () => query, limit: async () => rows, }; return query; @@ -4603,6 +4629,7 @@ describe("retained native cleanup activation", () => { "bad_identity_hash", "foreign_semantic_scope", "wrong_provider_account", + "released_identity_unproven", ].includes(mode); const succeeds = [ "digest_only_command", @@ -4620,6 +4647,7 @@ describe("retained native cleanup activation", () => { "activation_commit_stalled", "empty_root", "distinct_provider_account", + "released_identity", ].includes(mode); expect(outcome.status).toBe( succeeds @@ -11817,6 +11845,8 @@ describe("runnerd provider runtime wiring", () => { "live runner", "live group", "missing process identity", + "released process identity", + "released identity without stop receipt", "active prior run", "wrong checkpoint", "active goal", @@ -11937,13 +11967,34 @@ describe("runnerd provider runtime wiring", () => { } : profile, processPid: - scenario === "missing process identity" ? null : 90000001, - processGroupId: 90000001, + scenario === "missing process identity" || + scenario.startsWith("released") + ? null + : 90000001, + processGroupId: scenario.startsWith("released") + ? null + : 90000001, contextSnapshot: { paperclipEnvironment: { driver: "local" }, }, }, ]), + orderBy: () => ({ + limit: () => + Promise.resolve( + scenario === "released process identity" + ? [ + { + eventType: "native.local_process_stopped", + payload: { + processPid: 90000001, + processGroupId: 90000001, + }, + }, + ] + : [], + ), + }), }), }), }), @@ -12043,6 +12094,7 @@ describe("runnerd provider runtime wiring", () => { state.createTransport.mockClear(); const shouldRecover = [ "quarantined", + "released process identity", "large control plane", "empty retry shell", "unsuspended current", diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index bd6aaa8b008..e749ee48313 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -48,7 +48,7 @@ import { stoppedCodexTurnIsTextOnly } from "./stopped-codex-turn.js"; import { prepareVerifiedRemoteProviderPack, useVerifiedPreinstalledRemoteProviderPack, runRemoteProviderVerification, shouldUseBundledAcpxImageAssets } from "./remote-provider-pack.js"; import { readProviderPackArchive, stageProviderPackArchive } from "./provider-pack-archive.js"; import { selectRemotePiCompanion } from "./remote-pi-companion.js"; -import { readNativeLocalProcessStop, PROCESS_START_REQUESTED } from "../native-local-process-stop.js"; +import { readNativeLocalProcessStop, PROCESS_START_REQUESTED, withReleasedNativeRunProcess } from "../native-local-process-stop.js"; import { remoteLeaseCleanupScope } from "../remote-execution-termination.js"; import { connectorToolDefinitionsForSkills } from "../connector-runtime.js"; import { activeNativeExecutionRunIds } from "./native-active-runs.js"; @@ -3002,7 +3002,7 @@ export async function reconcileRetainedNativeSessionCleanup( } | null = null; try { claim = await db.transaction(async (tx) => { - const run = await tx + const row = await tx .select() .from(heartbeatRuns) .where( @@ -3014,6 +3014,9 @@ export async function reconcileRetainedNativeSessionCleanup( .for("update", { noWait: true }) .limit(1) .then((rows) => rows[0]); + const run = row + ? await withReleasedNativeRunProcess(tx as unknown as Db, row) + : row; if ( !run || run.runtimeMode !== "native" || @@ -4655,7 +4658,7 @@ async function verifyWarmTransitionRestart(input: { now: Date.now(), }); if (!proof) return deny(); - const prior = await tx + const priorRow = await tx .select() .from(heartbeatRuns) .where( @@ -4667,6 +4670,9 @@ async function verifyWarmTransitionRestart(input: { .for("update") .limit(1) .then((rows) => rows[0]); + const prior = priorRow + ? await withReleasedNativeRunProcess(tx as unknown as Db, priorRow) + : priorRow; if ( !prior || prior.agentId !== binding.agentId || @@ -5127,6 +5133,14 @@ async function recoverQuiescentRunnerdState(input: { .then((rows) => rows[0] ?? null); if (!prior || !TERMINAL_HEARTBEAT_RUN_STATUSES.has(prior.status)) continue; + const priorProcess = + prior.processPid === null && prior.processGroupId === null + ? await readNativeLocalProcessStop( + input.db, + input.execution.binding.companyId, + identity.runId, + ) + : { processPid: prior.processPid, processGroupId: prior.processGroupId }; const profile = record(prior.runnerProfileJson); const previous = parseNativeExecutionInput(profile.nativeExecutionInput); const checkpoint = record(profile.sessionCheckpoint); @@ -5172,8 +5186,8 @@ async function recoverQuiescentRunnerdState(input: { checkpoint.activeTurnId !== null || !Array.isArray(checkpoint.pendingRuntimeRequests) || checkpoint.pendingRuntimeRequests.length !== 0 || - !localProcessDefinitelyGone(prior.processPid) || - !localProcessDefinitelyGone(prior.processGroupId, true) + !localProcessDefinitelyGone(priorProcess?.processPid) || + !localProcessDefinitelyGone(priorProcess?.processGroupId, true) ) continue; // Provider processes are normally inside the runner group. Check any @@ -5205,8 +5219,8 @@ async function recoverQuiescentRunnerdState(input: { providerBytes, runner, runId: identity.runId, - processPid: prior.processPid!, - processGroupId: prior.processGroupId!, + processPid: priorProcess!.processPid!, + processGroupId: priorProcess!.processGroupId!, }); } catch { // Unreadable, unsafe, unscoped, or unprovable state stays quarantined.