mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 18:19:05 +02:00
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 75fcb0b3fc.
**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 <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip authored and GitHub committed
1 parent
4660b22f6b
commit
b5bf86ea09
11 files changed
+672
-22
No files matched your search
@@ -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<string, unknown>; 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<string, string>();
|
||||
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);
|
||||
});
|
||||
@@ -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<typeof import("../adapters/index.js")>("../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<void>(resolve => { child.once("exit", () => resolve()); child.kill("SIGKILL"); });
|
||||
return pid;
|
||||
}
|
||||
|
||||
suite("settled native run process identity", () => {
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let companyId: string;
|
||||
let agentId: string;
|
||||
const children: ReturnType<typeof startDetached>[] = [];
|
||||
|
||||
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<void>(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<void>(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<void>(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();
|
||||
});
|
||||
});
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<T extends {
|
||||
id: string; companyId: string; processPid: number | null; processGroupId: number | null; processStartedAt: Date | null;
|
||||
}>(db: Db, run: T): Promise<T> {
|
||||
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<SettledProcessRelease> {
|
||||
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<string, number>();
|
||||
|
||||
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 };
|
||||
}
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
@@ -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 });
|
||||
}
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
@@ -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<void>((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",
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in new issue
Block a user