mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-07 06:16:06 -10:00
fix(recovery): resume a continuation that never launched its runner in its own sandbox
A remote chat continuation inherits its predecessor's saved session and, for a warm session, its runner process identity. When it failed before requesting a runner of its own, the due same-run retry reattached to a runner that never existed and failed with native_remote_recovery_lease_mismatch. Recovery now proves that shape from durable state, retires the inherited identity, and uses the ordinary same-run resume, which keeps the run's retained sandbox through the exact-lease path. Runs with any runner start, controller checkpoint, provider event or pending request of their own keep verified reattachment, as do rows saved without the required facts. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
1 parent
84908d176b
commit
e977509c0a
7 files changed
+325
-4
No files matched your search
@@ -1478,6 +1478,30 @@ differed, for example `lease_status`, `adapter_type`, `provider_lease_id`, `remo
|
||||
`identity_run_id`. The detail contains no identifiers. Consumers match the code
|
||||
exactly or as the `code: ` prefix.
|
||||
|
||||
A continuation run starts with its predecessor's saved provider session. For a
|
||||
warm session it also carries the predecessor's runner process identity. When the
|
||||
run fails before it asks for a runner of its own (for example while it stages
|
||||
its provider pack), a due same-run retry does not reattach: there is no runner
|
||||
of this run to reattach to. The server retires the inherited process identity
|
||||
and uses the ordinary same-run resume, which is the path the failed attempt
|
||||
took. The resume prepares the run again in its own retained sandbox when that
|
||||
sandbox lease is still active, and acquires one as an ordinary resume does when
|
||||
it is gone. The server does this only when all of these are true:
|
||||
|
||||
- the failed attempt is a result-less exact-checkpoint retry that recorded
|
||||
`providerEventsExist: false`, and no controller owns it;
|
||||
- the run has no runner start, process identity, adapter invocation, tool start,
|
||||
runner event, controller checkpoint, provider event or provider session of
|
||||
its own;
|
||||
- the saved session has no active turn and no pending runtime request; and
|
||||
- a settled predecessor of the same native session recorded the exact same
|
||||
process identity.
|
||||
|
||||
It records an `inherited_runner_never_launched` lifecycle event. Any other run,
|
||||
and any run saved by an older release without these facts, keeps verified
|
||||
reattachment. A reattachment that loses its runner still requests the user's
|
||||
Retry as described above.
|
||||
|
||||
When a native attempt fails before launch because its predecessor's session is
|
||||
quarantined, explicit task Retry verifies the retained journal's actual owner.
|
||||
That owner must belong to the same company, task, agent, workspace, native session,
|
||||
|
||||
@@ -369,6 +369,88 @@ describePostgres("heartbeat recovery module ownership", () => {
|
||||
} finally { kill.mockRestore(); try { await f.cleanup(); } finally { await child.stop(); } }
|
||||
});
|
||||
|
||||
// Staging shape: a chat continuation inherited its predecessor's saved
|
||||
// provider session and warm runner identity, then failed while staging its
|
||||
// provider pack, before it ever asked for a runner of its own. Its workspace
|
||||
// reference is saved and its own sandbox lease is still active.
|
||||
type InheritedCase = "staging" | "own_turn" | "own_launch" | "own_checkpoint" | "no_predecessor"
|
||||
| "pending_side_effect" | "legacy_detail";
|
||||
async function inheritedRunnerRetry(pid: number, condition: InheritedCase) {
|
||||
const f = await remoteRetryFixture(pid, "remote_workspace");
|
||||
const run = await f.current();
|
||||
// The PID came from the predecessor, so this run recorded no process identity.
|
||||
await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id));
|
||||
if (condition !== "own_checkpoint") await db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.run.id));
|
||||
const profile = run.runnerProfileJson ?? {};
|
||||
const checkpoint = profile.sessionCheckpoint as Record<string, unknown>;
|
||||
await db.update(agents).set({ status: "active" }).where(eq(agents.id, f.run.agentId));
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeExecutionInput: { fixture: true },
|
||||
remoteWorkspace: { ...(profile.remoteWorkspace as Record<string, unknown>), state: "saved" },
|
||||
sessionCheckpoint: { ...checkpoint, activeTurnId: null,
|
||||
pendingRuntimeRequests: condition === "pending_side_effect" ? [{ requestId: "tool-1", kind: "tool_call" }] : [] },
|
||||
} }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
await db.update(environmentLeases).set({ executionWorkspaceId: null }).where(eq(environmentLeases.id, f.lease.id));
|
||||
await db.update(nativeRunFinalizations).set({ nextAttemptAt: null, failureDetail: {
|
||||
message: "runner_remote_provider_archive_invalid", recoveryMode: "exact_checkpoint_resume", checkpointExists: true,
|
||||
providerSessionEstablished: true, ...(condition === "legacy_detail" ? {} : { providerEventsExist: false }),
|
||||
} }).where(eq(nativeRunFinalizations.runId, f.run.id));
|
||||
const predecessorId = randomUUID();
|
||||
if (condition !== "no_predecessor") await db.insert(heartbeatRuns).values({ id: predecessorId,
|
||||
companyId: f.run.companyId, agentId: f.run.agentId, issueId: f.run.issueId, invocationSource: "automation",
|
||||
status: "succeeded", runtimeMode: "native", nativeSessionId: run.nativeSessionId, createdAt: new Date(Date.now() - 600_000),
|
||||
finishedAt: new Date(Date.now() - 300_000), processPid: run.processPid, processGroupId: run.processGroupId,
|
||||
processStartedAt: run.processStartedAt });
|
||||
if (condition === "own_turn") await appendHeartbeatRunEvent(db, { companyId: f.run.companyId, runId: f.run.id,
|
||||
agentId: f.run.agentId, eventType: "turn.started", stream: "system", level: "info", message: "turn.started" });
|
||||
if (condition === "own_launch") await appendHeartbeatRunEvent(db, { companyId: f.run.companyId, runId: f.run.id,
|
||||
agentId: f.run.agentId, eventType: PROCESS_START_REQUESTED, stream: "system", level: "info" });
|
||||
return { ...f, predecessorId, async cleanupAll() {
|
||||
try { await f.cleanup(); } finally { await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, predecessorId)); }
|
||||
} };
|
||||
}
|
||||
|
||||
it("resumes a continuation that never launched its own runner in its retained sandbox, without reattaching", async () => {
|
||||
const child = await ownedProcess();
|
||||
const f = await inheritedRunnerRetry(child.pid, "staging");
|
||||
try {
|
||||
native.claims.mockClear();
|
||||
await createHeartbeatRecovery(db, f.deps).resumeRetryableNativeProcessSessions({ runId: f.run.id });
|
||||
await Promise.all([...f.deps.activeRunExecutionPromises]);
|
||||
expect(native.claims.mock.calls.some(([input]) => input?.runIds?.includes(f.run.id))).toBe(false);
|
||||
expect(f.deps.executeRun).toHaveBeenCalledExactlyOnceWith(f.run.id, { nativeLeaseOwner: expect.any(String) });
|
||||
const current = await f.current();
|
||||
expect(current).toMatchObject({ processPid: null, processGroupId: null, processStartedAt: null });
|
||||
expect(current.runnerProfileJson?.nativeInheritedRunnerRetirement).toEqual({
|
||||
schema: "paperclip.native-inherited-runner-retirement.v1", predecessorRunId: f.predecessorId,
|
||||
retainedSandbox: { leaseId: f.lease.id, providerLeaseId: f.lease.providerLeaseId, remoteCwd: "/work" } });
|
||||
const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id));
|
||||
expect(events.find(event => event.payload?.reason === "inherited_runner_never_launched")?.payload)
|
||||
.toMatchObject({ predecessorRunId: f.predecessorId, retainedSandbox: true });
|
||||
// The retained sandbox lease stays the run's only lease; nothing new was acquired.
|
||||
expect(await db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, f.run.id)))
|
||||
.toMatchObject([{ id: f.lease.id, status: "active" }]);
|
||||
expect(f.deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled();
|
||||
expect(child.alive()).toBe(true);
|
||||
} finally { try { await f.cleanupAll(); } finally { await child.stop(); } }
|
||||
});
|
||||
|
||||
it.each(["own_turn", "own_launch", "own_checkpoint", "no_predecessor", "pending_side_effect", "legacy_detail"] as const)(
|
||||
"keeps verified reattachment when the run's own runner may hold work or the proof is incomplete: %s", async condition => {
|
||||
const child = await ownedProcess();
|
||||
const f = await inheritedRunnerRetry(child.pid, condition);
|
||||
try {
|
||||
native.claims.mockClear();
|
||||
await createHeartbeatRecovery(db, f.deps).resumeRetryableNativeProcessSessions({ runId: f.run.id });
|
||||
await Promise.all([...f.deps.activeRunExecutionPromises]);
|
||||
expect(native.claims.mock.calls.some(([input]) => input?.runIds?.includes(f.run.id))).toBe(true);
|
||||
expect(f.deps.executeRun).not.toHaveBeenCalledWith(f.run.id, { nativeLeaseOwner: expect.any(String) });
|
||||
const current = await f.current();
|
||||
expect(current.processPid).toBe(child.pid);
|
||||
expect(current.runnerProfileJson).not.toHaveProperty("nativeInheritedRunnerRetirement");
|
||||
} finally { try { await f.cleanupAll(); } finally { await child.stop(); } }
|
||||
},
|
||||
);
|
||||
|
||||
it("routes a due plugin-task retry through its existing remote lease", async () => {
|
||||
const child = await ownedProcess();
|
||||
const f = await remoteRetryFixture(child.pid, "plugin_task");
|
||||
|
||||
@@ -28,6 +28,7 @@ import {
|
||||
} from "../sandbox-stop-and-retain.js";
|
||||
import { discoverSettledNativeRunProcesses, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, recordNativeLocalProcessStopInTransaction, releaseSettledNativeRunProcess, retryPendingSettledNativeRunProcesses } from "../native-local-process-stop.js";
|
||||
import { pendingCaseHandoffForRun } from "./case-handoff-lifecycle.js";
|
||||
import { retireInheritedRemoteRunnerIdentity } from "../native-runtime/native-inherited-runner-retry.js";
|
||||
import {
|
||||
legacyControllerBootId,
|
||||
liveLegacyControllerExpiresAt,
|
||||
@@ -2202,7 +2203,10 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
if (activeRunExecutions.has(run.id) || runningProcesses.has(run.id)) continue;
|
||||
const profile = parseObject(run.runnerProfileJson);
|
||||
if (profile.nativeWorkspaceSync != null || profile.remoteWorkspace != null) {
|
||||
remoteRetryRunIds.push(run.id);
|
||||
// A continuation that never started its own runner holds only its
|
||||
// predecessor's identity: resume it the way its failed attempt ran.
|
||||
if (await retireInheritedRemoteRunnerIdentity(db, run.id)) claimableNativeRunIds.add(run.id);
|
||||
else remoteRetryRunIds.push(run.id);
|
||||
} else {
|
||||
const [taskLease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
|
||||
@@ -2065,6 +2065,51 @@ describe.skipIf(!support.supported && !externalDatabaseUrl)("heartbeat workspace
|
||||
expect(input.nativeRunnerPreparationSpans).toMatchObject([{ name: "environment.acquire", outcome: "failed" }]);
|
||||
});
|
||||
|
||||
// A same-run resume whose inherited runner identity was retired keeps the
|
||||
// run's own retained sandbox. When that sandbox is gone, it acquires as an
|
||||
// ordinary resume does.
|
||||
function retiredInheritedRunner(input: Awaited<ReturnType<typeof fixture>>["input"]) {
|
||||
const location = { leaseId: randomUUID(), providerLeaseId: "retained-sandbox", remoteCwd: "/retained" };
|
||||
input.run.runnerProfileJson = { ...(input.run.runnerProfileJson ?? {}),
|
||||
remoteWorkspace: { schema: "paperclip.remote-workspace-run.v1", workspaceId: randomUUID(), epoch: 2,
|
||||
state: "saved", snapshotId: randomUUID(), ...location },
|
||||
nativeInheritedRunnerRetirement: { schema: "paperclip.native-inherited-runner-retirement.v1",
|
||||
predecessorRunId: randomUUID(), retainedSandbox: location } };
|
||||
input.reuse.runOptions = { nativeLeaseOwner: "resume-owner" };
|
||||
return location;
|
||||
}
|
||||
|
||||
it("resumes a retired inherited runner in its own retained sandbox", async () => {
|
||||
const { input, envOrchestrator } = await fixture();
|
||||
const location = retiredInheritedRunner(input);
|
||||
const reached = new Error("Reached the retained sandbox");
|
||||
const acquire = vi.spyOn(envOrchestrator, "acquireForRun").mockRejectedValue(reached);
|
||||
await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(reached);
|
||||
expect(acquire).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ reattachRemoteLease: location }));
|
||||
});
|
||||
|
||||
it("acquires as an ordinary resume when the retained sandbox is gone", async () => {
|
||||
const { input, envOrchestrator } = await fixture();
|
||||
retiredInheritedRunner(input);
|
||||
const reached = new Error("Reached ordinary acquisition");
|
||||
const acquire = vi.spyOn(envOrchestrator, "acquireForRun")
|
||||
.mockRejectedValueOnce(new Error("native_remote_recovery_lease_mismatch: stage=environment_acquire field=lease_status"))
|
||||
.mockRejectedValueOnce(reached);
|
||||
await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(reached);
|
||||
expect(acquire).toHaveBeenCalledTimes(2);
|
||||
expect(acquire.mock.calls[1]![0]).not.toHaveProperty("reattachRemoteLease");
|
||||
});
|
||||
|
||||
it("does not reuse a retained sandbox without a retired inherited runner", async () => {
|
||||
const { input, envOrchestrator } = await fixture();
|
||||
retiredInheritedRunner(input);
|
||||
delete (input.run.runnerProfileJson as Record<string, unknown>).nativeInheritedRunnerRetirement;
|
||||
const reached = new Error("Reached ordinary acquisition");
|
||||
const acquire = vi.spyOn(envOrchestrator, "acquireForRun").mockRejectedValue(reached);
|
||||
await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(reached);
|
||||
expect(acquire.mock.calls[0]![0]).not.toHaveProperty("reattachRemoteLease");
|
||||
});
|
||||
|
||||
it("binds a replacement workspace from realization and snapshots the restore lease", async () => {
|
||||
const { input, envOrchestrator, issue } = await fixture();
|
||||
const realize = envOrchestrator.realizeForRun;
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { remoteRecoveryLeaseMismatch } from "../native-runtime/remote-recovery-lease-mismatch.js";
|
||||
import { NATIVE_REMOTE_RECOVERY_LEASE_MISMATCH, remoteRecoveryLeaseMismatch } from "../native-runtime/remote-recovery-lease-mismatch.js";
|
||||
import { readInheritedRunnerRetainedSandbox } from "../native-runtime/native-inherited-runner-retry.js";
|
||||
import { resolveProviderApiPlacement } from "../native-runtime/provider-api-placement.js";
|
||||
import { assertEphemeralWorkspaceExecution, readRemoteWorkspaceAuthority, RemoteWorkspaceWriterActiveError, reserveRemoteWorkspace, shouldUseRemoteWorkspaces } from "../remote-workspaces.js";
|
||||
import { WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, WorkspaceBusyDeferral } from "./retries.js";
|
||||
@@ -1306,7 +1307,22 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace
|
||||
: saved.remoteCwd !== sandboxRecovery.remoteCwd ? "remote_cwd" as const : null;
|
||||
if (mismatch) throw remoteRecoveryLeaseMismatch("workspace_preparation", mismatch);
|
||||
}
|
||||
acquiredEnvironment = await timeWorkspace("workspace_environment_acquire", () => envOrchestrator.acquireForRun({
|
||||
// A same-run resume whose inherited runner identity was retired keeps the
|
||||
// run's own retained sandbox, so its workspace and allocation carry over.
|
||||
const retainedSandbox = runOptions.nativeLeaseOwner && !runOptions.nativeRestartRecovery
|
||||
? readInheritedRunnerRetainedSandbox(parseObject(run.runnerProfileJson)) : null;
|
||||
const acquireRetainedOrNew = async (input: Parameters<typeof envOrchestrator.acquireForRun>[0]) => {
|
||||
if (!retainedSandbox) return envOrchestrator.acquireForRun(input);
|
||||
try {
|
||||
return await envOrchestrator.acquireForRun({ ...input, reattachRemoteLease: retainedSandbox });
|
||||
} catch (error) {
|
||||
// The retained sandbox is gone. Acquire as an ordinary resume does.
|
||||
if (!(error instanceof Error) || !error.message.startsWith(`${NATIVE_REMOTE_RECOVERY_LEASE_MISMATCH}: `)) throw error;
|
||||
logger.warn({ runId: run.id, detail: error.message }, "retained sandbox unavailable for same-run resume; acquiring");
|
||||
return envOrchestrator.acquireForRun(input);
|
||||
}
|
||||
};
|
||||
acquiredEnvironment = await timeWorkspace("workspace_environment_acquire", () => acquireRetainedOrNew({
|
||||
companyId: agent.companyId,
|
||||
selectedEnvironmentId: selectedEnvironmentId ?? localEnvironment.id,
|
||||
localEnvironmentId: localEnvironment.id,
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
import { and, desc, eq, inArray, isNull, lt, ne, or, sql } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared";
|
||||
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
|
||||
import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
|
||||
import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js";
|
||||
import { hasNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { nativeSessionProviderEvidence } from "./native-session-resume.js";
|
||||
import { hasNativeExecutionStartEvidence } from "./native-restart-recovery.js";
|
||||
import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
|
||||
/**
|
||||
* A continuation run starts with its predecessor's saved provider session and,
|
||||
* for a warm session, the predecessor's runner process identity. When such a
|
||||
* run fails before it ever asks for its own runner (for example while staging
|
||||
* its provider pack), the recorded process identity is the predecessor's, not
|
||||
* its own. A due same-run retry used to reattach to "this run's" runner, which
|
||||
* never existed, and always failed with `native_remote_recovery_lease_mismatch`.
|
||||
*
|
||||
* This module proves that shape from durable state and retires the inherited
|
||||
* identity, so the retry takes the ordinary same-run resume: the same path the
|
||||
* failed attempt took. It also records the run's own retained sandbox lease,
|
||||
* so preparation resumes that exact sandbox instead of acquiring another one.
|
||||
*/
|
||||
export const NATIVE_INHERITED_RUNNER_RETIREMENT_KEY = "nativeInheritedRunnerRetirement";
|
||||
const RETIREMENT_SCHEMA = "paperclip.native-inherited-runner-retirement.v1";
|
||||
|
||||
const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"];
|
||||
|
||||
export type InheritedRunnerRetainedSandbox = { leaseId: string; providerLeaseId: string; remoteCwd: string };
|
||||
|
||||
function record(value: unknown): Record<string, unknown> {
|
||||
return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
}
|
||||
|
||||
function sameTime(left: Date | null, right: Date | null) {
|
||||
return (left?.getTime() ?? null) === (right?.getTime() ?? null);
|
||||
}
|
||||
|
||||
/** The run's saved remote location: a native workspace sync or a remote workspace reference. */
|
||||
function savedRemoteLocation(profile: Record<string, unknown>): InheritedRunnerRetainedSandbox | null {
|
||||
const sync = readNativeWorkspaceSyncReference(profile.nativeWorkspaceSync);
|
||||
if (sync) return { leaseId: sync.leaseId, providerLeaseId: sync.providerLeaseId, remoteCwd: sync.remoteCwd };
|
||||
const remote = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace);
|
||||
return remote.success
|
||||
? { leaseId: remote.data.leaseId, providerLeaseId: remote.data.providerLeaseId, remoteCwd: remote.data.remoteCwd }
|
||||
: null;
|
||||
}
|
||||
|
||||
/** The retained sandbox recorded when the inherited identity was retired, if it still matches the saved location. */
|
||||
export function readInheritedRunnerRetainedSandbox(profile: Record<string, unknown> | null | undefined): InheritedRunnerRetainedSandbox | null {
|
||||
const marker = record(profile?.[NATIVE_INHERITED_RUNNER_RETIREMENT_KEY]);
|
||||
const sandbox = record(marker.retainedSandbox);
|
||||
const location = savedRemoteLocation(record(profile));
|
||||
if (marker.schema !== RETIREMENT_SCHEMA || !location || sandbox.leaseId !== location.leaseId
|
||||
|| sandbox.providerLeaseId !== location.providerLeaseId || sandbox.remoteCwd !== location.remoteCwd) return null;
|
||||
return location;
|
||||
}
|
||||
|
||||
/**
|
||||
* Retires the inherited remote runner identity of a due same-run retry when
|
||||
* durable state proves all of these:
|
||||
*
|
||||
* - the failed attempt is a result-less exact-checkpoint retry that recorded
|
||||
* `providerEventsExist: false`, and no controller owns it;
|
||||
* - this run never requested or identified a runner, invoked the adapter,
|
||||
* started a tool, or committed a runner event, has no controller checkpoint
|
||||
* of its own, and has no provider event or provider session of its own;
|
||||
* - its saved session has no active turn and no pending runtime request; and
|
||||
* - its exact process identity is the one a settled predecessor of the same
|
||||
* native session recorded.
|
||||
*
|
||||
* Anything else, including rows saved by a release that did not record these
|
||||
* facts, keeps the existing verified reattachment path.
|
||||
*/
|
||||
export async function retireInheritedRemoteRunnerIdentity(db: Db, runId: string): Promise<boolean> {
|
||||
return db.transaction(async transaction => {
|
||||
const tx = transaction as unknown as Db;
|
||||
await tx.execute(sql`select set_config('lock_timeout', '2000', true)`);
|
||||
const [candidate] = await tx.select({ companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.nativeIssueId })
|
||||
.from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).limit(1);
|
||||
if (!candidate?.issueId) return false;
|
||||
// Same lock order as same-run resumption: issue, coordinator, run.
|
||||
const [issue] = await tx.select({ id: issues.id }).from(issues).where(and(
|
||||
eq(issues.companyId, candidate.companyId), eq(issues.id, candidate.issueId))).for("update").limit(1);
|
||||
if (!issue) return false;
|
||||
const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, candidate.companyId), eq(nativeRunFinalizations.runId, runId),
|
||||
eq(nativeRunFinalizations.issueId, issue.id))).for("update").limit(1);
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, runId))).for("update").limit(1);
|
||||
if (!coordinator || !run || run.runtimeMode !== "native" || isNativeRunnerOwnershipHeld(run)
|
||||
|| !["running", "failed"].includes(run.status) || run.sessionIdAfter
|
||||
|| (run.processPid === null && run.processGroupId === null) || !run.nativeSessionId) return false;
|
||||
const detail = record(coordinator.failureDetail);
|
||||
if (coordinator.phase !== "retryable_failure" || coordinator.resultId !== null || coordinator.leaseOwner !== null
|
||||
|| detail.recoveryMode !== "exact_checkpoint_resume" || detail.providerEventsExist !== false
|
||||
|| typeof detail.successorRunId === "string") return false;
|
||||
|
||||
const profile = record(run.runnerProfileJson);
|
||||
const location = savedRemoteLocation(profile);
|
||||
const checkpoint = record(profile.sessionCheckpoint);
|
||||
const pending = checkpoint.pendingRuntimeRequests;
|
||||
if (!location || checkpoint.activeTurnId != null
|
||||
|| (pending !== undefined && pending !== null && (!Array.isArray(pending) || pending.length > 0))) return false;
|
||||
if (await hasNativeExecutionStartEvidence(tx, run)) return false;
|
||||
if ((await nativeSessionProviderEvidence(tx, [run.id])).has(run.id)) return false;
|
||||
if (await hasNativeRunnerCheckpoint(tx, { companyId: run.companyId, issueId: issue.id, runId: run.id })) return false;
|
||||
|
||||
// The identity must be exactly the one a settled predecessor recorded.
|
||||
const [predecessor] = await tx.select({ id: heartbeatRuns.id, processStartedAt: heartbeatRuns.processStartedAt })
|
||||
.from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId),
|
||||
eq(heartbeatRuns.nativeSessionId, run.nativeSessionId), ne(heartbeatRuns.id, run.id),
|
||||
lt(heartbeatRuns.createdAt, run.createdAt), inArray(heartbeatRuns.status, TERMINAL_RUN_STATUSES),
|
||||
run.processPid === null ? isNull(heartbeatRuns.processPid) : eq(heartbeatRuns.processPid, run.processPid),
|
||||
run.processGroupId === null ? isNull(heartbeatRuns.processGroupId) : eq(heartbeatRuns.processGroupId, run.processGroupId),
|
||||
)).orderBy(desc(heartbeatRuns.createdAt)).limit(1);
|
||||
if (!predecessor || !sameTime(predecessor.processStartedAt, run.processStartedAt)) return false;
|
||||
|
||||
// Keep the run's own sandbox when its lease is still live.
|
||||
const [lease] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.id, location.leaseId),
|
||||
eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.providerLeaseId, location.providerLeaseId),
|
||||
eq(environmentLeases.status, "active"), isNull(environmentLeases.releasedAt), isNull(environmentLeases.cleanupStatus),
|
||||
or(isNull(environmentLeases.expiresAt), sql`${environmentLeases.expiresAt} > now()`))).limit(1);
|
||||
|
||||
await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id, agentId: run.agentId });
|
||||
await tx.update(heartbeatRuns).set({
|
||||
processPid: null, processGroupId: null, processStartedAt: null,
|
||||
runnerProfileJson: { ...profile, [NATIVE_INHERITED_RUNNER_RETIREMENT_KEY]: {
|
||||
schema: RETIREMENT_SCHEMA, predecessorRunId: predecessor.id,
|
||||
...(lease ? { retainedSandbox: location } : {}),
|
||||
} },
|
||||
updatedAt: new Date(),
|
||||
}).where(eq(heartbeatRuns.id, run.id));
|
||||
await appendHeartbeatRunEvent(tx, {
|
||||
companyId: run.companyId, runId: run.id, agentId: run.agentId,
|
||||
eventType: "lifecycle", stream: "system", level: "info",
|
||||
message: "Retrying this run in its own sandbox: it never started a runner, so there was nothing to reattach to.",
|
||||
payload: { reason: "inherited_runner_never_launched", predecessorRunId: predecessor.id,
|
||||
retainedSandbox: Boolean(lease) },
|
||||
});
|
||||
return true;
|
||||
}).catch(error => {
|
||||
logger.warn({ err: error, runId }, "failed to retire an inherited remote runner identity");
|
||||
return false;
|
||||
});
|
||||
}
|
||||
@@ -714,7 +714,7 @@ export function classifyNativeRunnerRecoveryEvidence(input: {
|
||||
|
||||
/** Any record that execution reached a runner: a requested or identified
|
||||
* process, an adapter invocation, a tool start, or a runner-sourced event. */
|
||||
async function hasNativeExecutionStartEvidence(
|
||||
export async function hasNativeExecutionStartEvidence(
|
||||
tx: Pick<Db, "select">, run: { companyId: string; id: string },
|
||||
): Promise<boolean> {
|
||||
const [executionEvent] = await tx
|
||||
|
||||
Reference in new issue
Block a user