mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
Settle remote recovery verification failures and retry with remote authority
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
6596b09e37
commit
14163ef057
8 files changed
+242
-44
No files matched your search
@@ -147,7 +147,12 @@ runner. The runner must then authenticate to its existing durable PRP authority.
|
||||
Adoption preserves the provider attempt and does not launch another runner or
|
||||
replace its artifacts. A dead runner can restart only from verified suspended
|
||||
state. Missing or conflicting evidence preserves the controller journal and
|
||||
fails recovery. Host PID checks do not establish remote process ownership.
|
||||
fails recovery. Verification failures release the claimed execution lease and
|
||||
persist a recovery disposition. Temporary connection failures retry after 30
|
||||
seconds through the same remote verifier and consume the bounded attempt budget;
|
||||
missing or conflicting authority blocks automatic replacement. The retry keeps
|
||||
remote process identifiers and cannot launch a provider before verification.
|
||||
Host PID checks do not establish remote process ownership.
|
||||
Remote runners have a five-minute reconnect grace period for app replacement;
|
||||
liveness probes share one request in flight and run at most once per second.
|
||||
|
||||
|
||||
@@ -17137,6 +17137,12 @@ export function heartbeatService(
|
||||
);
|
||||
const claimableNativeRunIds = new Set<string>();
|
||||
for (const { run } of retryableNativeProcesses) {
|
||||
if (run.errorCode === "runner_remote_recovery_unavailable") {
|
||||
// Its PIDs belong to the sandbox. Only remote authority verification
|
||||
// can establish whether that executor survived the interruption.
|
||||
claimableNativeRunIds.add(run.id);
|
||||
continue;
|
||||
}
|
||||
if (!run.processPid && !run.processGroupId) {
|
||||
claimableNativeRunIds.add(run.id);
|
||||
continue;
|
||||
@@ -17197,6 +17203,7 @@ export function heartbeatService(
|
||||
dispatch: (claim) => {
|
||||
const execution = executeRun(claim.runId, {
|
||||
nativeLeaseOwner: claim.leaseOwner,
|
||||
nativeRestartRecovery: claim.restartRecovery,
|
||||
}).catch((error) => {
|
||||
logger.error(
|
||||
{ err: error, runId: claim.runId },
|
||||
@@ -17960,6 +17967,7 @@ export function heartbeatService(
|
||||
dispatch: (claim) => {
|
||||
const execution = executeRun(claim.runId, {
|
||||
nativeLeaseOwner: claim.leaseOwner,
|
||||
nativeRestartRecovery: claim.restartRecovery,
|
||||
}).catch((error) => {
|
||||
logger.error(
|
||||
{ err: error, runId: claim.runId },
|
||||
|
||||
@@ -15,6 +15,7 @@ import {
|
||||
workFolderRuns,
|
||||
workspaceOperations,
|
||||
} from "@paperclipai/db";
|
||||
import { claimNativeRestartRecoveries, type NativeRestartRecoveryClaim } from "./native-restart-recovery.js";
|
||||
import { finalizeNativeRun, recordNativeFinalizationFailure } from "./native-run-finalizer.js";
|
||||
import {
|
||||
commitNativeStatusDecision,
|
||||
@@ -162,7 +163,7 @@ export function resolveNativeReconciliationStatus(input: {
|
||||
throw new Error("native_reconciliation_facts_invalid");
|
||||
}
|
||||
|
||||
export type NativeSessionResumeClaim = { runId: string; leaseOwner: string };
|
||||
export type NativeSessionResumeClaim = { runId: string; leaseOwner: string; restartRecovery?: NativeRestartRecoveryClaim };
|
||||
|
||||
export async function dispatchNativeSessionResumptions(input: {
|
||||
db: Db;
|
||||
@@ -191,13 +192,15 @@ export async function claimNativeSessionResumptions(input: {
|
||||
limit?: number;
|
||||
}): Promise<NativeSessionResumeClaim[]> {
|
||||
const now = input.now ?? new Date();
|
||||
const candidates = await input.db.select({ runId: heartbeatRuns.id })
|
||||
const candidates = await input.db.select({ runId: heartbeatRuns.id, errorCode: heartbeatRuns.errorCode })
|
||||
.from(heartbeatRuns)
|
||||
.innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
|
||||
.where(and(
|
||||
eq(heartbeatRuns.runtimeMode, "native"),
|
||||
isNull(heartbeatRuns.processPid),
|
||||
isNull(heartbeatRuns.processGroupId),
|
||||
or(
|
||||
and(isNull(heartbeatRuns.processPid), isNull(heartbeatRuns.processGroupId)),
|
||||
eq(heartbeatRuns.errorCode, "runner_remote_recovery_unavailable"),
|
||||
),
|
||||
isNull(nativeRunFinalizations.resultId),
|
||||
eq(nativeRunFinalizations.phase, "retryable_failure"),
|
||||
or(isNull(nativeRunFinalizations.nextAttemptAt), lte(nativeRunFinalizations.nextAttemptAt, now)),
|
||||
@@ -212,6 +215,20 @@ export async function claimNativeSessionResumptions(input: {
|
||||
|
||||
const claims: NativeSessionResumeClaim[] = [];
|
||||
for (const candidate of candidates) {
|
||||
if (candidate.errorCode === "runner_remote_recovery_unavailable") {
|
||||
// A failed probe says nothing about the remote process's liveness. Keep
|
||||
// its identifiers and re-enter exact remote-authority reconciliation.
|
||||
const dispositions = await claimNativeRestartRecoveries({
|
||||
db: input.db, runIds: [candidate.runId], now,
|
||||
restartKind: "hard", remoteVerificationRetry: true,
|
||||
});
|
||||
for (const disposition of dispositions) {
|
||||
if (disposition.kind === "reconcile_remote_runner") {
|
||||
claims.push({ runId: disposition.runId, leaseOwner: disposition.leaseOwner, restartRecovery: disposition });
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
const leaseOwner = `${input.runnerInstanceId}:resume:${randomUUID()}`;
|
||||
let terminalRunToEmit: typeof heartbeatRuns.$inferSelect | null = null;
|
||||
const claimed = await input.db.transaction(async (tx) => {
|
||||
@@ -227,6 +244,7 @@ export async function claimNativeSessionResumptions(input: {
|
||||
if (!row) return false;
|
||||
if (
|
||||
row.run.runtimeMode !== "native"
|
||||
|| row.run.errorCode === "runner_remote_recovery_unavailable"
|
||||
|| row.run.processPid !== null
|
||||
|| row.run.processGroupId !== null
|
||||
|| row.coordinator.resultId
|
||||
|
||||
@@ -436,6 +436,7 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
controller?: NativeControllerIdentity;
|
||||
restartKind: NativeRestartKind;
|
||||
recoveryRequestId?: string | null;
|
||||
remoteVerificationRetry?: boolean;
|
||||
runIds?: string[];
|
||||
now?: Date;
|
||||
limit?: number;
|
||||
@@ -471,6 +472,10 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
),
|
||||
),
|
||||
),
|
||||
...(input.remoteVerificationRetry ? [
|
||||
eq(nativeRunFinalizations.phase, "retryable_failure"),
|
||||
eq(heartbeatRuns.errorCode, "runner_remote_recovery_unavailable"),
|
||||
] : []),
|
||||
...(input.runIds?.length
|
||||
? [inArray(heartbeatRuns.id, input.runIds)]
|
||||
: []),
|
||||
@@ -513,6 +518,13 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
reason: "recovery_rows_missing",
|
||||
} as const;
|
||||
}
|
||||
if (input.remoteVerificationRetry && (
|
||||
row.coordinator.phase !== "retryable_failure" ||
|
||||
row.run.errorCode !== "runner_remote_recovery_unavailable" ||
|
||||
(row.coordinator.nextAttemptAt && row.coordinator.nextAttemptAt > now)
|
||||
)) {
|
||||
return { kind: "blocked", runId: row.run.id, reason: "remote_verification_retry_changed" } as const;
|
||||
}
|
||||
if (row.coordinator.resultId) {
|
||||
return {
|
||||
kind: "already_finalized",
|
||||
@@ -577,8 +589,8 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
typeof environment === "object" &&
|
||||
!Array.isArray(environment) &&
|
||||
["sandbox", "ssh"].includes(String((environment as Record<string, unknown>).driver));
|
||||
const runnerPidAlive = !remote && processIsAlive(row.run.processPid);
|
||||
const runnerGroupAlive = !remote && processGroupIsAlive(row.run.processGroupId);
|
||||
const runnerPidAlive = !remote && !input.remoteVerificationRetry && processIsAlive(row.run.processPid);
|
||||
const runnerGroupAlive = !remote && !input.remoteVerificationRetry && processGroupIsAlive(row.run.processGroupId);
|
||||
const observedRunnerStart =
|
||||
row.run.processPid && runnerPidAlive
|
||||
? await observedProcessStart(row.run.processPid)
|
||||
@@ -702,14 +714,16 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
}
|
||||
}
|
||||
const providerProcesses = await evaluateNativeProviderProcesses({
|
||||
identities: (remote ? [] : providerProcessIdentities).filter(
|
||||
identities: (remote || input.remoteVerificationRetry ? [] : providerProcessIdentities).filter(
|
||||
(identity) => identity.pid !== row.run.processPid,
|
||||
),
|
||||
});
|
||||
const hasProviderEvidence =
|
||||
hasCheckpointProviderIdentity || providerEvents.length > 0;
|
||||
|
||||
const classification = remote && (environment as Record<string, unknown>).driver === "ssh"
|
||||
const classification = input.remoteVerificationRetry && !remote
|
||||
? { claimKind: null, reason: "remote_recovery_environment_missing" }
|
||||
: remote && (environment as Record<string, unknown>).driver === "ssh"
|
||||
? { claimKind: null, reason: "remote_recovery_transport_unsupported" }
|
||||
: classifyNativeRunnerRecoveryEvidence({
|
||||
remote,
|
||||
|
||||
+123
-2
@@ -1,7 +1,7 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { spawn } from "node:child_process";
|
||||
import { existsSync } from "node:fs";
|
||||
import { mkdtemp, readFile, rename, rm } from "node:fs/promises";
|
||||
import { mkdir, mkdtemp, readFile, rename, rm, writeFile } from "node:fs/promises";
|
||||
import { createServer, type Server } from "node:http";
|
||||
import { tmpdir } from "node:os";
|
||||
import { resolve } from "node:path";
|
||||
@@ -35,9 +35,14 @@ import { readProcessStartedAt } from "../hot-restart.js";
|
||||
import { prepareNativeHeartbeatRun } from "./prepare-native-run.js";
|
||||
import {
|
||||
claimNativeRestartRecoveries,
|
||||
currentNativeControllerIdentity,
|
||||
type NativeControllerIdentity,
|
||||
} from "./native-restart-recovery.js";
|
||||
|
||||
import { dispatchNativeSessionResumptions } from "./native-finalization-reconciler.js";
|
||||
import { executePaperclipNativeSession } from "./native-session-executor.js";
|
||||
import type { NativeExecutionInputV1 } from "@paperclipai/paperclip-runner";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported
|
||||
? describe
|
||||
@@ -227,6 +232,122 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
return { db, issueId, runId, native };
|
||||
}
|
||||
|
||||
it.each([
|
||||
{ scenario: "unavailable", priorAttempt: 1, phase: "retryable_failure" },
|
||||
{ scenario: "unavailable", priorAttempt: 2, phase: "terminal_failure" },
|
||||
{ scenario: "timeout", priorAttempt: 1, phase: "retryable_failure" },
|
||||
{ scenario: "mismatched", priorAttempt: 1, phase: "terminal_failure" },
|
||||
{ scenario: "missing-host-state", priorAttempt: 1, phase: "terminal_failure" },
|
||||
] as const)("settles a claimed remote verification failure ($scenario, attempt $priorAttempt)", async ({ scenario, priorAttempt, phase }) => {
|
||||
const fixture = await seedRun(`VERIFY-${scenario}-${priorAttempt}`);
|
||||
const controller = await currentNativeControllerIdentity();
|
||||
const leaseOwner = `verification-${randomUUID()}`;
|
||||
const stateBase = await mkdtemp(resolve(runtimeRoot, "remote-verification-"));
|
||||
const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR;
|
||||
process.env.PAPERCLIP_RUNNER_STATE_DIR = stateBase;
|
||||
const execution: NativeExecutionInputV1 = {
|
||||
schema: "paperclip.native-execution-input.v1",
|
||||
provider: { kind: "codex", model: null },
|
||||
binding: { companyId, runId: fixture.runId, issueId: fixture.issueId, agentId, executionWorkspaceId: "verification-workspace" },
|
||||
task: { identifier: "NRR-VERIFY", title: "Verify claimed recovery", description: null, prompt: "Must not dispatch a provider turn.", workMode: "standard" },
|
||||
workspace: { cwd: stateBase, repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: fixture.native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } },
|
||||
completionContract: { id: "verification-contract", sha256: "verification-sha", schemaVersion: "paperclip.completion-contract.v1", contract: { revision: "1", objective: "Verify recovery", criteria: [{ id: "objective", requirement: "No stranded lease" }] } },
|
||||
interactionResponses: [], credentialBindings: [],
|
||||
};
|
||||
// This canonical scope is the existing on-disk contract. Only controller
|
||||
// journal state exists locally; remote runner state must not be fabricated.
|
||||
const scope = JSON.stringify({ agentId, companyId, normalizedSessionId: fixture.native.normalizedSessionId,
|
||||
provider: { driverKind: "codex_app_server", identity: { kind: "codex" } },
|
||||
schema: "paperclip.native-session-scope.v2", workspace: { executionWorkspaceId: "verification-workspace", kind: "managed" } });
|
||||
const root = resolve(stateBase, createHash("sha256").update(scope).digest("hex"));
|
||||
const identity = { runId: fixture.runId, normalizedSessionId: fixture.native.normalizedSessionId,
|
||||
runnerInstanceId: fixture.native.runnerInstanceId, environmentLeaseId: fixture.native.environmentLeaseId };
|
||||
const journal = JSON.stringify({ schema: "paperclip.runner.durable.control-plane-state.v1", identity, committedEvents: [{ preserved: true }] });
|
||||
await mkdir(resolve(root, "control-plane"), { recursive: true });
|
||||
const journalPath = resolve(root, "control-plane", "control-plane-state.json");
|
||||
if (scenario !== "missing-host-state") await writeFile(journalPath, journal);
|
||||
await fixture.db.update(heartbeatRuns).set({
|
||||
// Deliberately use a live host PID as the remote numeric PID. The retry
|
||||
// must retain it and must not apply host process ownership checks.
|
||||
processPid: process.pid, processGroupId: process.pid,
|
||||
contextSnapshot: { issueId: fixture.issueId, paperclipEnvironment: { driver: "sandbox" } },
|
||||
runnerProfileJson: { nativeExecutionInput: execution, sessionCheckpoint: {
|
||||
providerSessionId: "original-provider-session",
|
||||
identity: { companyId, agentId, issueId: fixture.issueId, runId: fixture.runId, sessionId: fixture.native.normalizedSessionId },
|
||||
} },
|
||||
}).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
await fixture.db.update(nativeRunFinalizations).set({
|
||||
phase: "observed", attempt: priorAttempt, leaseOwner, leaseExpiresAt: new Date(Date.now() + 20 * 60_000),
|
||||
controllerBootId: controller.bootId, controllerPid: controller.pid, controllerProcessStartedAt: controller.processStartedAt,
|
||||
controllerGeneration: 2, nextAttemptAt: null,
|
||||
}).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
const execute = vi.fn(async (input: { command: string; args: string[] }) => {
|
||||
expect(input.command).toBe("node");
|
||||
if (scenario === "unavailable") throw new Error("fixture connection unavailable");
|
||||
if (scenario === "timeout") return { exitCode: null, timedOut: true, stdout: "" };
|
||||
const request = JSON.parse(input.args[2]!);
|
||||
return { exitCode: 0, timedOut: false, stdout: JSON.stringify({ identity: { ...identity, runId: randomUUID() }, stateDirectory: request.stateDirectory }) };
|
||||
});
|
||||
const onSpawn = vi.fn();
|
||||
const unavailable = scenario === "unavailable" || scenario === "timeout";
|
||||
const code = unavailable ? "runner_remote_recovery_unavailable" : "runner_remote_recovery_unverified";
|
||||
try {
|
||||
await expect(executePaperclipNativeSession({
|
||||
db: fixture.db, execution, runnerInstanceId: fixture.native.runnerInstanceId,
|
||||
useRunnerd: true, leaseOwner, onSpawn,
|
||||
restartRecovery: { kind: "reconcile_remote_runner", runId: fixture.runId, leaseOwner, controllerGeneration: 2, providerAttempt: priorAttempt, restartKind: "hard", recoveryRequestId: null },
|
||||
runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/home/daytona/repos/project", providerKey: "daytona", runner: { execute } } as never,
|
||||
})).rejects.toThrow(code);
|
||||
const [coordinator] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(coordinator).toMatchObject({ phase, attempt: priorAttempt + 1, leaseOwner: null, leaseExpiresAt: null,
|
||||
recoveryState: phase === "retryable_failure" ? "resuming_session" : "blocked",
|
||||
failureCode: !unavailable ? code : phase === "terminal_failure" ? "native_session_retry_exhausted" : "native_session_interrupted",
|
||||
failureDetail: { originalFailureCode: code },
|
||||
});
|
||||
if (phase === "retryable_failure") {
|
||||
expect(coordinator!.nextAttemptAt!.getTime()).toBeGreaterThan(Date.now() + 20_000);
|
||||
expect(coordinator!.nextAttemptAt!.getTime()).toBeLessThan(Date.now() + 31_000);
|
||||
} else expect(coordinator!.nextAttemptAt).toBeNull();
|
||||
expect(coordinator!.recoveryHistory).toEqual([expect.objectContaining({ reason: code, disposition: phase, providerAttempt: priorAttempt + 1 })]);
|
||||
const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
expect(run).toMatchObject({ nativePhase: phase, errorCode: code });
|
||||
const [issue] = await fixture.db.select().from(issues).where(eq(issues.id, fixture.issueId));
|
||||
expect(issue!.status).toBe(phase === "terminal_failure" ? "in_review" : "in_progress");
|
||||
expect(onSpawn).not.toHaveBeenCalled();
|
||||
expect(execute).toHaveBeenCalledTimes(scenario === "missing-host-state" ? 0 : 1);
|
||||
if (scenario !== "missing-host-state") expect(await readFile(journalPath, "utf8")).toBe(journal);
|
||||
expect(existsSync(resolve(stateBase, "quarantine"))).toBe(false);
|
||||
if (phase === "retryable_failure") {
|
||||
const dispatch = vi.fn();
|
||||
await dispatchNativeSessionResumptions({ db: fixture.db, runnerInstanceId: "retry-controller", runIds: [fixture.runId], dispatch });
|
||||
expect(dispatch).not.toHaveBeenCalled();
|
||||
await fixture.db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(Date.now() - 1) }).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
const claims = await dispatchNativeSessionResumptions({ db: fixture.db, runnerInstanceId: "retry-controller", runIds: [fixture.runId], dispatch });
|
||||
expect(claims).toHaveLength(1);
|
||||
expect(dispatch).toHaveBeenCalledExactlyOnceWith(claims[0]);
|
||||
expect(claims[0]!.restartRecovery).toMatchObject({ kind: "reconcile_remote_runner", providerAttempt: 2, controllerGeneration: 3 });
|
||||
const duplicate = await dispatchNativeSessionResumptions({ db: fixture.db, runnerInstanceId: "competing-controller", runIds: [fixture.runId], dispatch: vi.fn() });
|
||||
expect(duplicate).toEqual([]);
|
||||
await expect(executePaperclipNativeSession({
|
||||
db: fixture.db, execution, runnerInstanceId: fixture.native.runnerInstanceId,
|
||||
useRunnerd: true, leaseOwner: claims[0]!.leaseOwner, restartRecovery: claims[0]!.restartRecovery, onSpawn,
|
||||
runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/home/daytona/repos/project", providerKey: "daytona", runner: { execute } } as never,
|
||||
})).rejects.toThrow(code);
|
||||
const [settled] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(settled).toMatchObject({ phase: "terminal_failure", attempt: 3, leaseOwner: null, leaseExpiresAt: null, nextAttemptAt: null, failureCode: "native_session_retry_exhausted" });
|
||||
expect(await readFile(journalPath, "utf8")).toBe(journal);
|
||||
expect(onSpawn).not.toHaveBeenCalled();
|
||||
expect(execute).toHaveBeenCalledTimes(2);
|
||||
const [retained] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
expect(retained).toMatchObject({ processPid: process.pid, processGroupId: process.pid });
|
||||
}
|
||||
} finally {
|
||||
if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR;
|
||||
else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous;
|
||||
}
|
||||
});
|
||||
|
||||
function transportOptions(
|
||||
fixture: Awaited<ReturnType<typeof seedRun>>,
|
||||
stateDirectory: string,
|
||||
|
||||
@@ -2650,6 +2650,7 @@ export function nativeSessionFailureDisposition(
|
||||
) {
|
||||
const permanentFailure =
|
||||
sourceFailureCode === "native_event_replay_conflict" ||
|
||||
sourceFailureCode === "runner_remote_recovery_unverified" ||
|
||||
sourceFailureCode === "runner_remote_provider_artifact_incompatible";
|
||||
const exhausted = permanentFailure || attempt >= 3;
|
||||
return {
|
||||
@@ -2690,6 +2691,8 @@ export function nativeSessionFailureSourceCode(
|
||||
error: unknown,
|
||||
):
|
||||
| "runner_remote_provider_artifact_incompatible"
|
||||
| "runner_remote_recovery_unverified"
|
||||
| "runner_remote_recovery_unavailable"
|
||||
| "provider_process_exited"
|
||||
| "provider_stdout_closed"
|
||||
| "provider_process_output_closed"
|
||||
@@ -2705,6 +2708,12 @@ export function nativeSessionFailureSourceCode(
|
||||
| "native_event_replay_conflict"
|
||||
| "native_session_interrupted" {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
if (/runner_remote_recovery_unverified/i.test(message)) {
|
||||
return "runner_remote_recovery_unverified";
|
||||
}
|
||||
if (/runner_remote_recovery_unavailable/i.test(message)) {
|
||||
return "runner_remote_recovery_unavailable";
|
||||
}
|
||||
if (/runner_remote_provider_artifact_incompatible/i.test(message)) {
|
||||
return "runner_remote_provider_artifact_incompatible";
|
||||
}
|
||||
@@ -3853,23 +3862,37 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
await trace.end(environmentScope, { endedAtMs: environmentEndedAtMs });
|
||||
}
|
||||
let verifiedRemoteRecovery: RemoteRunnerRecovery | undefined;
|
||||
let remoteRecoveryVerificationError: Error | undefined;
|
||||
if (input.useRunnerd) {
|
||||
verifiedRemoteRecovery = await migrateRunnerdStateRootForExecution({
|
||||
db: input.db,
|
||||
execution: input.execution,
|
||||
allowVerifiedBackup:
|
||||
input.runnerExecutionTarget?.kind === "remote" &&
|
||||
input.runnerExecutionTarget.transport === "sandbox",
|
||||
// A retained warm runner is deliberately still ready rather than
|
||||
// suspended. Only the exact idle in-process owner may rotate that
|
||||
// prior-run authority; after a hard restart the map is empty and the
|
||||
// durable-state verifier continues to require a suspended runner.
|
||||
allowRetainedWarmRunner: hasIdleWarmNativeSessionOwner(input),
|
||||
restartRecovery: input.restartRecovery,
|
||||
runnerExecutionTarget: input.runnerExecutionTarget,
|
||||
});
|
||||
try {
|
||||
verifiedRemoteRecovery = await migrateRunnerdStateRootForExecution({
|
||||
db: input.db,
|
||||
execution: input.execution,
|
||||
allowVerifiedBackup:
|
||||
input.runnerExecutionTarget?.kind === "remote" &&
|
||||
input.runnerExecutionTarget.transport === "sandbox",
|
||||
// A retained warm runner is deliberately still ready rather than
|
||||
// suspended. Only the exact idle in-process owner may rotate that
|
||||
// prior-run authority; after a hard restart the map is empty and the
|
||||
// durable-state verifier continues to require a suspended runner.
|
||||
allowRetainedWarmRunner: hasIdleWarmNativeSessionOwner(input),
|
||||
restartRecovery: input.restartRecovery,
|
||||
runnerExecutionTarget: input.runnerExecutionTarget,
|
||||
});
|
||||
} catch (error) {
|
||||
if (input.restartRecovery?.kind !== "reconcile_remote_runner") throw error;
|
||||
// Restart recovery already owns a durable claim. Settle failed remote
|
||||
// verification through the same fenced failure handler as execution,
|
||||
// without loading unverified state or creating a provider session.
|
||||
remoteRecoveryVerificationError = new Error(
|
||||
nativeSessionFailureSourceCode(error) === "runner_remote_recovery_unavailable"
|
||||
? "runner_remote_recovery_unavailable"
|
||||
: "runner_remote_recovery_unverified",
|
||||
{ cause: error },
|
||||
);
|
||||
}
|
||||
}
|
||||
const durableRunnerBinding = input.useRunnerd
|
||||
const durableRunnerBinding = input.useRunnerd && !remoteRecoveryVerificationError
|
||||
? loadRunnerdDurableBinding(input.execution)
|
||||
: null;
|
||||
const effectiveRunnerInstanceId =
|
||||
@@ -3998,7 +4021,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
const nextAttempt = nextNativeProviderAttempt(
|
||||
coordinator.attempt,
|
||||
recovering?.kind === "reconcile_remote_runner" &&
|
||||
verifiedRemoteRecovery?.alive === false
|
||||
(verifiedRemoteRecovery?.alive === false || remoteRecoveryVerificationError !== undefined)
|
||||
? "resume_dead_runner"
|
||||
: recovering?.kind,
|
||||
);
|
||||
@@ -4369,7 +4392,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
);
|
||||
let existingWarmSession: NativeSession | undefined;
|
||||
let persistedWarmSession: PersistedNativeSession | null | undefined;
|
||||
if (warmSessionId !== null && warmConfigDigest !== null) {
|
||||
if (!remoteRecoveryVerificationError && warmSessionId !== null && warmConfigDigest !== null) {
|
||||
const entry = warmNativeSessions.get(warmSessionId);
|
||||
if (entry) {
|
||||
// Run-scoped broker capabilities must rotate with the process, while the
|
||||
@@ -4485,6 +4508,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
controller,
|
||||
});
|
||||
try {
|
||||
if (remoteRecoveryVerificationError) throw remoteRecoveryVerificationError;
|
||||
const runnerdBackend =
|
||||
input.useRunnerd && input.backend === undefined
|
||||
? await createRunnerdBackend({
|
||||
@@ -4762,6 +4786,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
const { exhausted } = recoveryProjection;
|
||||
const integrityFailure =
|
||||
sourceFailureCode === "native_event_replay_conflict";
|
||||
const remoteAuthorityFailure = sourceFailureCode === "runner_remote_recovery_unverified";
|
||||
const message =
|
||||
error instanceof Error
|
||||
? error.message.slice(0, 2_000)
|
||||
@@ -4852,13 +4877,15 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
nextAction:
|
||||
recoveryEvidence.recoveryMode === "ambiguous_state"
|
||||
? "Inspect the original provider failure and durable events; state is ambiguous and a replacement provider session is forbidden."
|
||||
: integrityFailure
|
||||
? "Inspect the persisted runner events and checkpoint for a source-sequence integrity conflict; automatic recovery is stopped."
|
||||
: exhausted
|
||||
? "Inspect the persisted native session after its bounded resume budget was exhausted."
|
||||
: recoveryEvidence.recoveryMode === "bootstrap_retry"
|
||||
? "Retry provider bootstrap on this same run; durable evidence proves no provider session or provider event was created."
|
||||
: "Resume this same run from its exact persisted native provider checkpoint after the retry delay.",
|
||||
: remoteAuthorityFailure
|
||||
? "Inspect the original sandbox identity and durable runner state; automatic replacement is forbidden until authority is verified."
|
||||
: integrityFailure
|
||||
? "Inspect the persisted runner events and checkpoint for a source-sequence integrity conflict; automatic recovery is stopped."
|
||||
: exhausted
|
||||
? "Inspect the persisted native session after its bounded resume budget was exhausted."
|
||||
: recoveryEvidence.recoveryMode === "bootstrap_retry"
|
||||
? "Retry provider bootstrap on this same run; durable evidence proves no provider session or provider event was created."
|
||||
: "Resume this same run from its exact persisted native provider checkpoint after the retry delay.",
|
||||
},
|
||||
nextAttemptAt,
|
||||
recoveryHistory: sql`(
|
||||
@@ -4963,13 +4990,15 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
nextAction:
|
||||
recoveryEvidence.recoveryMode === "ambiguous_state"
|
||||
? "Inspect the original provider failure and explicitly resolve the ambiguous session state; do not open a replacement provider session."
|
||||
: integrityFailure
|
||||
? "Inspect the persisted runner event collision and explicitly repair or replace the run; automatic retries are disabled."
|
||||
: exhausted
|
||||
? "Inspect the provider trace and explicitly choose a replacement run or provider configuration; automatic provider work is stopped."
|
||||
: recoveryEvidence.recoveryMode === "bootstrap_retry"
|
||||
? "Retry bootstrap on the same run without manufacturing a provider checkpoint."
|
||||
: "Resume the exact persisted native session on the same heartbeat run.",
|
||||
: remoteAuthorityFailure
|
||||
? "Inspect the original sandbox identity and durable runner state; do not launch a replacement provider session without verified authority."
|
||||
: integrityFailure
|
||||
? "Inspect the persisted runner event collision and explicitly repair or replace the run; automatic retries are disabled."
|
||||
: exhausted
|
||||
? "Inspect the provider trace and explicitly choose a replacement run or provider configuration; automatic provider work is stopped."
|
||||
: recoveryEvidence.recoveryMode === "bootstrap_retry"
|
||||
? "Retry bootstrap on the same run without manufacturing a provider checkpoint."
|
||||
: "Resume the exact persisted native session on the same heartbeat run.",
|
||||
wakePolicy: nextAttemptAt
|
||||
? {
|
||||
kind: "resume_native_run",
|
||||
|
||||
@@ -122,6 +122,6 @@ describe("remote recovery evidence", () => {
|
||||
expect(await Promise.all([adopted.isAlive(), adopted.isAlive(), adopted.isAlive()])).toEqual([true, true, true]);
|
||||
expect(execute).toHaveBeenCalledOnce();
|
||||
execute.mockRejectedValueOnce(new Error("provider unavailable"));
|
||||
await expect(adopted.signal("SIGTERM")).rejects.toThrow("provider unavailable");
|
||||
await expect(adopted.signal("SIGTERM")).rejects.toThrow("runner_remote_recovery_unavailable");
|
||||
});
|
||||
});
|
||||
@@ -97,8 +97,11 @@ async function probe(
|
||||
args: ["-e", REMOTE_RECOVERY_PROBE, JSON.stringify(request)],
|
||||
bypassSession: true,
|
||||
timeoutMs: 10_000,
|
||||
}).catch((cause: unknown) => {
|
||||
throw new Error("runner_remote_recovery_unavailable", { cause });
|
||||
});
|
||||
if (result.exitCode !== 0 || result.timedOut || result.stdout.length > 8192) {
|
||||
if (result.timedOut) throw new Error("runner_remote_recovery_unavailable");
|
||||
if (result.exitCode !== 0 || result.stdout.length > 8192) {
|
||||
throw new Error("runner_remote_recovery_unverified");
|
||||
}
|
||||
let value: RemoteRunnerRecovery;
|
||||
|
||||
Reference in new issue
Block a user