From b5bf86ea09042a864b60910f80c8fdf7e64e9ef0 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 23 Oct 2026 15:09:43 -0700 Subject: [PATCH] Clear process identity when settled native runs stop (#17374) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Each heartbeat run stores the process identity of its runner in `heartbeat_runs.process_pid`, `process_group_id` and `process_started_at` > - Only cancellations (`release_run_resources`), restart recovery and heartbeat recovery cleared these fields > - Failed native runs take the `schedule_retry` path, so they kept a PID after the run finished, for weeks > - Liveness checks for old runs then depend on PID-reuse defenses, and code that reads a set PID as "running" is misled > - This pull request releases the process identity of settled native runs when the process is confirmed gone, and backfills old sandbox rows > - The benefit is that a finished run no longer looks like it owns a live process, while live, warm and retrying processes keep their identity ## Linked Issues or Issue Description Refs #17358. Bug report described in-PR (bug report template). **What happened?** On one instance, 16 of 23 old native workspace sync folders belonged to failed or cancelled runs that finished more than a week earlier and still had `process_pid` set. The PID is written when the run's process starts (for remote targets it is the sandbox PID). The status arbiter clears it only through `release_run_resources`, which only cancellations emit. Failed native runs take `schedule_retry`, and other terminal paths (finalization reconciler, restart recovery terminal failures, watchdog timeouts) do not clear it either. **Expected behavior** When a run is terminal and its process is confirmed gone (or no longer owned), the process identity is cleared. A process that may still be alive keeps its identity. **Steps to reproduce** 1. Run a native task whose provider turn fails, so the arbiter decides `run_failed_partial_evidence_preserved` with a `schedule_retry` effect. 2. Wait for the run to finish and the runner to exit. 3. Read the run row: `status = 'failed'` and `process_pid` / `process_group_id` / `process_started_at` are still set. **Paperclip version or commit:** `master` at 75fcb0b3fca. **Deployment mode:** any. **Agent adapter(s) involved:** native runtime (`paperclip_runner`); local and sandbox targets. ## What Changed - `releaseSettledNativeRunProcess` (in `native-local-process-stop.ts`) clears `process_pid`, `process_group_id` and `process_started_at` of a native run under a row lock, only when all of these are true: - the status is `failed`, `cancelled`, `timed_out` or `interrupted`, and `finished_at` is set; - the finalization coordinator is absent, `committed` or `terminal_failure` (a `retryable_failure` same-run retry keeps the existing heartbeat recovery path; unfinished finalization keeps its identity); - the run has a session checkpoint or provider evidence, so native session resume never mistakes it for an unused bootstrap; - local target: every lease is released without failed cleanup, the latest `native.process_identity_recorded` event names this host and the same PID/group, and the PID and group probe absent. The release writes the existing `native.local_process_stopped` receipt in the same transaction, so `hasNativeLocalProcessStop`, `readNativeLocalProcessStop`, explicit continuation and acknowledged Stop keep their proof; - sandbox target: every lease has a remote termination receipt (sandbox PIDs are never probed on this host). - Succeeded runs keep their identity: a warm runner can serve the next turn, and chat retry matches warm predecessors by PID. - The heartbeat executor calls the release after lease and scratch cleanup at a terminal boundary (not when a same-run resume, workspace finalization or held ownership is pending). - Background work stays idle (see `rules/sleep-aware-background-work.mdc`): - Trigger: the executor settlement calls `settleNativeRunProcess`. If the release is not ready yet (process still alive, local lease not released, remote termination receipt missing), the run goes into an in-memory pending set. - Startup: the first `reapOrphanedRuns` pass after boot runs `discoverSettledNativeRunProcesses` once, in pages of 50, over the durable rows. This reconstructs work after a restart and covers terminal transitions outside an executor (reconciler, restart recovery, watchdog, timeouts). - Retry: later passes call `retryPendingSettledNativeRunProcesses` only for pending runs. It makes no database call when the set is empty. A run leaves the set when it is released, becomes ineligible, or reaches 20 attempts. - Runs that this process still executes are skipped. - Stop receipts now also record `processStartedAt`. `withReleasedNativeRunProcess` restores a released identity from the latest stop receipt. Retained session cleanup (`reconcileRetainedNativeSessionCleanup`) and warm transition restart checks use it. Quiescent runnerd state recovery reads the stop receipt when the prior run's identity was released. Without a receipt, all three still treat the identity as unproven. - Migration `0377_clear_settled_native_run_process_identity` backfills sandbox runs (without bumping `updated_at`) with the same rules: native, settled status, `finished_at` older than 1 day, settled coordinator phase, checkpoint or provider evidence, and at least one lease where every lease has a valid remote termination receipt. The snapshot is carried forward as `0377_snapshot.json`. - Not changed: legacy (non-native) runs. Case handoff release, scratch cleanup and conversation continuation read the terminal row's local PID/group as process-tree absence evidence, so clearing them needs separate work. `release_run_resources` cancellation behavior and `native_run_finalizations.controller_*` fields (they track the server controller, not the run process) are unchanged. ### Migration margin The 1-day margin is defense in depth. The real proof is the remote termination receipt: the provider confirmed that the sandbox stopped or was destroyed. One day is far longer than lease release, sandbox stop, restart recovery takeover and the finalization reconciler sweep, so no in-flight teardown or same-run recovery can still read these rows. Local rows are not backfilled in SQL: only the host that recorded the process can verify that it stopped and write the stop receipt. The runtime sweep releases them on that host. ## Verification - `pnpm exec vitest run src/__tests__/native-settled-process-release.test.ts` (new, 20 tests): release with a stop receipt and restore of the identity from it; `cancelled`/`timed_out`/`interrupted`; live process kept, then released after exit; kept for succeeded, `retryable_failure`, `workspace_finalizing`, unreleased local lease, another host, no evidence, unconfirmed sandbox, mixed leases; sandbox release with receipts; startup discovery paging and skip; pending retry; retry cap; no database access while nothing is pending; `reapOrphanedRuns` discovery, then retry after the process exits; executor release for a failed native run and no release for a succeeded run. - `native-provider-failure.integration.test.ts`: a real `schedule_retry` decision (`run_failed_partial_evidence_preserved`) leaves the failed run with its PID after finalization, and the release then clears it and the stop receipt reads back. - `native-session-executor.test.ts`: quiescent recovery and retained session cleanup accept a released identity with a stop receipt and reject one without a receipt. - `packages/db` `client.test.ts` migration rule: the backfill does not set `updated_at`. - `native-finalization-migration.test.ts`: migration 0377 clears only old settled native sandbox rows with confirmed termination (15 fixtures, including recent, succeeded, running, legacy, unsettled phases, no evidence, unconfirmed or local leases, no lease). - Related suites, all passing: `pnpm exec vitest run` on heartbeat-process-recovery, native-workspace-sync-retention(-config), native-finalization-{migration,settled-candidates,recovery}, chat-failed-run-retry-boundary, acknowledged-native-stop, explicit-native-continuation, conversation-continuation, native-session-resume, native-run-finalizer-telemetry, native-session-executor, native-safe-replacement, native-finalization-reconciler-timing, native-restart-recovery, native-finalization-recovery-state, run-lifecycle, agent-instruction/directory working copies, heartbeat-case-handoff (1920 tests across 23 files). - `pnpm run typecheck` in `server/`: passes. `pnpm run check:migrations` and the snapshot drift test in `packages/db`: pass. - Mutation checks (each reverted alone; a test failed every time): executor call, startup discovery, pending retry in `reapOrphanedRuns`, retry cap, pending-set cleanup on retry, start time in the stop receipt, retained cleanup receipt fallback, coordinator phase guard, succeeded exclusion, liveness probe, host check, remote receipt check, provider-evidence guard, local lease check, quiescent recovery receipt fallback, migration margin, migration NULL-safe receipt check, migration phase guard. The NULL-safe check came from a real bug the migration test found. The warm transition restart call site has no direct test: that path has no existing fixture. It uses the same tested helper. - Not run: the full server suite and e2e tests. ## Risks - Behavior change: more native runs now have a null `process_pid` after they settle. Every native reader was checked: explicit continuation, acknowledged Stop and agent stop use the stop receipt for null identities; cancelled-startup detection also requires no process events, so a released run is not mistaken for a cancelled startup; native session resume selection is protected by the evidence guard; chat retry reads its warm-predecessor PIDs from succeeded runs, which are not released. - A recycled local PID keeps a row unreleased (conservative), and sandbox runs from providers without termination receipts keep their identity. - Background cost: one paged discovery at startup. After that, only pending runs are checked, at most 20 times each. With nothing pending there is no recurring query. - A transition outside an executor that happens after startup, with no executor settlement, is released at the next startup discovery. - Migration: a data-only `UPDATE` on `heartbeat_runs` with a selective `WHERE`; it passes the migration safety check. It is safe to re-run. ## Model Used Claude Opus 5.5 (`claude-opus-5-5`), 1M context, through Claude Code with tool use. ## 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 - [ ] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- ...ar_settled_native_run_process_identity.sql | 43 +++ ...{0372_snapshot.json => 0377_snapshot.json} | 35 ++- packages/db/src/migrations/meta/_journal.json | 7 + .../native-finalization-migration.test.ts | 84 +++++ .../native-settled-process-release.test.ts | 287 ++++++++++++++++++ server/src/services/heartbeat.ts | 7 + server/src/services/heartbeat/recovery.ts | 16 +- .../src/services/native-local-process-stop.ts | 108 ++++++- ...ative-provider-failure.integration.test.ts | 17 ++ .../native-session-executor.test.ts | 62 +++- .../native-runtime/native-session-executor.ts | 28 +- 11 files changed, 672 insertions(+), 22 deletions(-) create mode 100644 packages/db/src/migrations/0377_clear_settled_native_run_process_identity.sql rename packages/db/src/migrations/meta/{0372_snapshot.json => 0377_snapshot.json} (99%) create mode 100644 server/src/__tests__/native-settled-process-release.test.ts 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.