diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts index 698f8a475bc..918779e5261 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts @@ -33,6 +33,7 @@ import { decodeDurablePrpCheckpoint, captureLegacyDurablePrpCheckpoint, inspectDurablePrpCheckpointHeader, + inspectUnauthenticatedBootstrapCheckpoint, inspectUnclaimedBootstrapCheckpoint, inspectUnclaimedBootstrapState, inspectStagedRunAttachment, @@ -166,6 +167,62 @@ it("recognizes a complete unclaimed bootstrap through cold checkpoint restoratio } finally { await core.stop(); rmSync(root, { recursive: true, force: true }); } }); +it("proves a launched bootstrap never authenticated after its credentials were issued", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-unauthenticated-bootstrap-")); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest }); + let checkpoint!: DurablePrpCheckpoint; + core.setCheckpointPersistence(async value => { checkpoint = value; }); + try { + const template = { provider: { provider: "codex", command: "fixture" }, authorizedTools: {} }; + core.persistRunAttachTemplate(template); + core.queueCommand("run.prepare", template); + core.queueCommand("session.open", { reuse: "same_session" }); + core.issueBootstrapTicket(60_000); + core.issueBootstrapTicket(60_000); + await core.flushCheckpoint(); + expect(inspectUnclaimedBootstrapCheckpoint(checkpoint.records, identity)).toBeNull(); + expect(inspectUnauthenticatedBootstrapCheckpoint(checkpoint.records, identity)).toBe(true); + expect(inspectUnauthenticatedBootstrapCheckpoint(checkpoint.records, { ...identity, runId: "foreign" })).toBe(false); + for (const mutate of [ + (header: { state: Record }) => { header.state.connectionCount = 1; }, + (header: { state: Record }) => { header.state.lastLeaseId = "lease"; }, + (header: { state: Record }) => { + const tickets = header.state.tickets as Record>; + for (const ticket of Object.values(tickets)) ticket.usedAt = "2026-01-01T00:00:00.000Z"; + }, + ]) { + const changed = new Map(checkpoint.records), header = JSON.parse(changed.get("header")!); + mutate(header); changed.set("header", JSON.stringify(header)); + expect(inspectUnauthenticatedBootstrapCheckpoint(changed, identity)).toBe(false); + } + core.queueCommand("turn.start", { prompt: "queued but never delivered" }); + await core.flushCheckpoint(); + expect(inspectUnauthenticatedBootstrapCheckpoint(checkpoint.records, identity)).toBe(true); + } finally { await core.stop(); rmSync(root, { recursive: true, force: true }); } +}); + +it("hands the ticket refresh to the process owner for the first launch only", () => { + const launches: RunnerProcessLaunchSpec[] = []; + const refreshBootstrapTicket = vi.fn(async () => "fresh-ticket"); + const handle = spawnRunner({ + connection: { mode: "connect", connectUrl: "ws://127.0.0.1:43127" }, + stateDirectory: "/tmp/paperclip-runner-refresh-test", identity, ticket: "bootstrap-ticket", + maxOutboxBytes: 256 * 1024, p0ReserveBytes: 64 * 1024, + runnerVersion: expectedRunnerVersion, runnerDigest: expectedRunnerDigest, + refreshBootstrapTicket, + processLauncher: spec => { + launches.push(spec); + return { child: { pid: 42, exitCode: null, kill: () => true }, + completion: Promise.resolve({ code: 0, signal: null, stdout: "", stderr: "" }), isExitConfirmed: () => true }; + }, + }); + expect(launches[0]?.refreshBootstrapTicket).toBe(refreshBootstrapTicket); + expect(launches[0]?.environment.PAPERCLIP_RUNNER_BOOTSTRAP_TICKET).toBe("bootstrap-ticket"); + handle.restart!("replacement-ticket"); + expect(launches[1]?.refreshBootstrapTicket).toBeUndefined(); + expect(launches[1]?.environment.PAPERCLIP_RUNNER_BOOTSTRAP_TICKET).toBe("replacement-ticket"); +}); + it("retains ever-issued bootstrap history after credential pruning and cold restore", async () => { const root = mkdtempSync(resolve(tmpdir(), "runner-issued-bootstrap-")); const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest }); diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index c39bdf006ad..9e77d4c4b25 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -597,6 +597,34 @@ export function inspectUnclaimedBootstrapCheckpoint( } catch { return null; } } +function isUnauthenticatedBootstrapState(state: StoredCoreState, expected: Partial): boolean { + const history = state.bootstrapHistory; + const matches = (candidate: DurableRecoveryIdentity | undefined) => candidate !== undefined + && (Object.keys(expected) as Array).every(key => candidate[key] === expected[key]); + return !(state.schema !== coreStateSchema || state.checkpointEventPolicy !== "control-v1" + || !expected.runId || !matches(state.identity) + || history?.schema !== "paperclip.runner.bootstrap-history.v1" + || canonicalJson(history.originIdentity) !== canonicalJson(state.identity) + || history.adopted !== false + || state.warmTransition || state.completedWarmTransition || state.stagedRunAttachment + || state.connectionCount !== 0 || state.ackedSourceSeq !== 0 || state.committedEvents.length !== 0 + || Object.values(state.tickets).some(ticket => ticket.usedAt !== null || ticket.warmTransitionId !== undefined) + || Object.keys(state.leases).length !== 0 || state.lastLeaseId !== null || state.lastLeaseExpiresAt !== null + || state.replayDeliveries !== 0 || state.duplicateCommandResults !== 0 + || state.commands.length < 2 || state.commands[0]?.type !== "run.prepare" + || state.commands[1]?.type !== "session.open" + || state.commands.some(command => command.status !== "pending" || command.result !== null)); +} + +export function inspectUnauthenticatedBootstrapCheckpoint( + records: ReadonlyMap, expected: Partial, +): boolean { + try { + if ([...records.values()].reduce((bytes, value) => bytes + Buffer.byteLength(value), 0) > 8 * 1024 * 1024) return false; + return isUnauthenticatedBootstrapState(checkpointState(records).state, expected); + } catch { return false; } +} + export function inspectUnclaimedBootstrapState( contents: string, expected: DurableRecoveryIdentity, verifiedArtifact: { version: string; digest: string }, ): ReturnType { @@ -649,6 +677,7 @@ export interface RunnerProcessLaunchSpec { environment: NodeJS.ProcessEnv; /** True while the launched Runner holds an authenticated connection to its control plane. */ isRunnerConnected?: () => boolean; + refreshBootstrapTicket?: () => Promise; } function domainDigest(domain: string, parts: readonly Buffer[]): Buffer { @@ -4742,6 +4771,7 @@ export function spawnRunner(options: { environment?: NodeJS.ProcessEnv; processLauncher?: (spec: RunnerProcessLaunchSpec) => RunnerProcessHandle; isRunnerConnected?: () => boolean; + refreshBootstrapTicket?: () => Promise; diagnosticsDirectory?: string; }): RunnerProcessHandle { for (const value of [options.maxRuntimeMs, options.maxLifetimeMs]) { @@ -4884,7 +4914,7 @@ export function spawnRunner(options: { throw new Error("runner_process_termination_unconfirmed"); } if (lifetimeDeadline !== undefined && lifetimeDeadline <= Date.now()) throw new Error("runner_process_lifetime_expired"); - return spawnRunner({ ...options, ticket, + return spawnRunner({ ...options, ticket, refreshBootstrapTicket: undefined, ...(lifetimeDeadline === undefined ? {} : { maxLifetimeMs: lifetimeDeadline - Date.now() }) }); }, }); @@ -4892,6 +4922,7 @@ export function spawnRunner(options: { return withRestart( options.processLauncher({ command, args, cwd: packageRoot, environment, ...(options.isRunnerConnected ? { isRunnerConnected: options.isRunnerConnected } : {}), + ...(options.refreshBootstrapTicket ? { refreshBootstrapTicket: options.refreshBootstrapTicket } : {}), nativeRunner: { identity: options.identity, ticket: options.ticket, connection, stateDirectory: options.stateDirectory, runnerVersion: options.runnerVersion, runnerDigest: options.runnerDigest, diff --git a/packages/paperclip-runner/src/index.ts b/packages/paperclip-runner/src/index.ts index 25f2a3c0217..fcd60bbfcf9 100644 --- a/packages/paperclip-runner/src/index.ts +++ b/packages/paperclip-runner/src/index.ts @@ -51,6 +51,7 @@ export { captureLegacyDurablePrpCheckpoint, inspectDurablePrpCheckpointHeader, inspectUnclaimedBootstrapCheckpoint, + inspectUnauthenticatedBootstrapCheckpoint, inspectUnclaimedBootstrapState, type PrpWireConnection, type PrpWireAttachment, diff --git a/packages/paperclip-runner/src/live/runnerd-bootstrap-ticket-refresh.test.ts b/packages/paperclip-runner/src/live/runnerd-bootstrap-ticket-refresh.test.ts new file mode 100644 index 00000000000..914c8321893 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-bootstrap-ticket-refresh.test.ts @@ -0,0 +1,90 @@ +import { createHash } from "node:crypto"; +import { spawn } from "node:child_process"; +import { mkdir, mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { afterEach, expect, it, vi } from "vitest"; + +import type { NativeExecutionInput } from "../contracts/native-execution.js"; +import { createNativeSessionBackend } from "../index.js"; +import type { RunnerProcessLaunchSpec } from "../control-plane/durable-prp-control-plane.js"; +import { createRunnerProviderLaunch, runnerLaunchRequirementsDigest, + type RunnerProviderLaunchRequirements } from "./runner-provider-launch.js"; +import { createCapabilityRunnerdCodexTransport, defaultCapabilityRunnerdBinary } from "./runnerd-codex-transport.js"; + +const fakeCodex = resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"); +const identity = { runnerInstanceId: "runner-ticket", environmentLeaseId: "lease-ticket", runId: "run", + normalizedSessionId: "session", turnId: "turn-ticket", itemId: "item-ticket" }; +const sessionIdentity = { runId: "run", sessionId: "session", companyId: "company", issueId: "issue", agentId: "agent" }; +const STAGING_MS = 75_000; + +function execution(): NativeExecutionInput { + return { + schema: "paperclip.native-execution-input.v1", + binding: { companyId: "company", runId: "run", issueId: "issue", agentId: "agent", executionWorkspaceId: "workspace" }, + task: { identifier: "PAP-1", title: "Slow staging", description: null, prompt: "Complete the task.", workMode: "standard" }, + workspace: { cwd: "/controller/workspace", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: "session", driverKind: "codex_app_server", protocolVersion: 1, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + provider: { kind: "codex", model: null, approvalPolicy: "never" }, + completionContract: { id: "contract", sha256: "sha256", schemaVersion: "1", + contract: { revision: "revision", objective: "Complete the task.", criteria: [{ id: "objective", requirement: "Complete the task." }] } }, + interactionResponses: [], credentialBindings: [], + }; +} + +afterEach(() => { vi.restoreAllMocks(); }); + +it("authenticates a runner whose remote staging outlasts the bootstrap ticket lifetime", async () => { + const root = await mkdtemp(join(tmpdir(), "runner-ticket-refresh-")); + const cwd = join(root, "vm", "workspace"); + await mkdir(cwd, { recursive: true }); + const binary = defaultCapabilityRunnerdBinary(); + const runnerArtifact = { version: "0.3.0", digest: `sha256:${createHash("sha256").update(await readFile(binary)).digest("hex")}` }; + const prepareLaunch = vi.fn(async (requirements: RunnerProviderLaunchRequirements) => ({ + configurationId: `plugin-${requirements.identity.runId}`, requirementsDigest: runnerLaunchRequirementsDigest(requirements), + provider: createRunnerProviderLaunch(requirements, { options: { codexArgs: ["--state-file", join(root, "fake-codex-state.json")] }, + root, cwd, codexHome: join(root, "guest-codex-home"), codexCommand: fakeCodex, + providerNodeCommand: process.execPath, opencodeProxyPath: "", acpxSidecarPath: "" }) })); + const realNow = Date.now.bind(Date); + let skewMs = 0; + vi.spyOn(Date, "now").mockImplementation(() => realNow() + skewMs); + const registered = new WeakSet(); + const backend = createNativeSessionBackend(execution(), { + runnerInstanceId: identity.runnerInstanceId, environment: {}, workingDirectoryAuthority: "runner_prepared_launch", + codexTransportFactory: () => createCapabilityRunnerdCodexTransport({ + provider: "codex", stateDirectory: join(root, "state"), runnerArtifact, prepareLaunch, prpIdentity: identity, + externallySandboxed: true, + controlPlaneRegistration: async controlPlane => { + if (!registered.has(controlPlane)) { await controlPlane.start(); registered.add(controlPlane); } + return { release: () => undefined }; + }, + runnerProcessLauncher: (spec: RunnerProcessLaunchSpec) => { + let kill: (signal?: NodeJS.Signals | number) => boolean = () => false; + const child = { pid: undefined as number | undefined, exitCode: null, signalCode: null, + kill: (signal?: NodeJS.Signals | number) => kill(signal) }; + const completion = (async () => { + skewMs += STAGING_MS; + const ticket = spec.refreshBootstrapTicket + ? await spec.refreshBootstrapTicket() + : spec.environment.PAPERCLIP_RUNNER_BOOTSTRAP_TICKET; + const runner = spawn(binary, [...spec.args], { + env: { ...spec.environment, PAPERCLIP_RUNNER_BOOTSTRAP_TICKET: ticket }, stdio: "ignore" }); + child.pid = runner.pid; + kill = signal => runner.kill(signal); + return await new Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }>( + (resolveExit, reject) => { + runner.once("error", reject); + runner.once("exit", (code, signal) => resolveExit({ code, signal, stdout: "", stderr: "" })); + }); + })(); + return { child: child as never, completion }; + }, + }).transport, + }); + try { + const session = await backend.openSession({ identity: sessionIdentity, workingDirectory: cwd }); + expect((await session.snapshot()).workingDirectory).toBe(cwd); + await session.close({ reason: "test complete" }); + } finally { await rm(root, { recursive: true, force: true }); } +}, 45_000); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-startup-steps.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-startup-steps.test.ts index 3f9db93417b..d27abbf79de 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport-startup-steps.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-startup-steps.test.ts @@ -6,6 +6,7 @@ import { createCapabilityRunnerdCodexTransport, type RunnerdStartupStep } from " import { defaultCapabilityRunnerdBinary, fakeCodex, fakeCodexArgs } from "../../test/support/runnerd-transport.js"; const STEP_ORDER = ["runner.runtime.stage", "runner.process.launch", "provider.session.load"]; +const BOOTSTRAP_STEP_ORDER = ["runner.control_plane.open", ...STEP_ORDER]; const PROVIDER_DELAY_MS = 1_500; /** A slow provider must show up as session load, never as runner process launch. */ @@ -19,7 +20,7 @@ function expectProviderTimeInSessionLoad(steps: RunnerdStartupStep[]): void { } function expectOrderedSteps(steps: RunnerdStartupStep[], mode: RunnerdStartupStep["mode"]): void { - expect(steps.map((step) => step.name)).toEqual(STEP_ORDER); + expect(steps.map((step) => step.name)).toEqual(mode === "bootstrap" ? BOOTSTRAP_STEP_ORDER : STEP_ORDER); for (const [index, step] of steps.entries()) { expect(step).toMatchObject({ mode, outcome: "ok" }); expect(step.endedAtMs).toBeGreaterThanOrEqual(step.startedAtMs); @@ -61,7 +62,7 @@ it("reports local stage, launch and provider session load for bootstrap and cold const thread = opened.thread as { id: string; sessionId: string }; await first.transport.close(); // Closing a live transport is not a startup: no further steps. - expect(bootstrapSteps).toHaveLength(3); + expect(bootstrapSteps).toHaveLength(BOOTSTRAP_STEP_ORDER.length); const resumeSteps: RunnerdStartupStep[] = []; // A new run on the saved session rotates authority through run.attach, @@ -98,7 +99,7 @@ it("never lets a failing startup-step sink change startup", async () => { try { const opened = await bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] }); expect((opened.thread as { id?: string }).id).toEqual(expect.any(String)); - expect(names).toEqual(STEP_ORDER); + expect(names).toEqual(BOOTSTRAP_STEP_ORDER); } finally { await bundle.transport.close().catch(() => undefined); await rm(root, { recursive: true, force: true }); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 39cbf01287a..71a37b75020 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -1729,9 +1729,9 @@ export interface CapabilityRunnerdCodexTransportOptions extends RunnerExecutionD }) => void | Promise; } -/** One measured local startup sub-step, named for the run-log span it becomes. */ +/** One measured startup sub-step, named for the run-log span it becomes. */ export interface RunnerdStartupStep { - name: "runner.runtime.stage" | "runner.process.launch" | "provider.session.load"; + name: "runner.control_plane.open" | "runner.runtime.stage" | "runner.process.launch" | "provider.session.load"; mode: "bootstrap" | "resume"; startedAtMs: number; endedAtMs: number; @@ -5504,6 +5504,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { return operation; } + async #refreshBootstrapTicket(core: DurablePrpControlPlane): Promise { + this.#assertStartupOpen(); + const ticket = core.issueBootstrapTicket(RUNNER_BOOTSTRAP_TICKET_TTL_MS); + await core.flushCheckpoint(); + this.#assertStartupOpen(); + return ticket; + } + #assertStartupOpen(): void { if (this.#closed) throw new Error("runnerd transport closed during provider startup"); } @@ -6188,6 +6196,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { await this.#resume(); return this.#openedThreadResponse(params); } + this.#startupSteps?.begin("runner.control_plane.open"); const token = randomUUID().replaceAll("-", ""); const identity = this.options.prpIdentity ?? { runnerInstanceId: `runner_lab_${token}`, @@ -6305,6 +6314,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { diagnosticsDirectory: resolve(this.#root, "diagnostics"), processLauncher: this.options.runnerProcessLauncher, isRunnerConnected: () => core.activeRunnerConnectionCount() > 0, + refreshBootstrapTicket: () => this.#refreshBootstrapTicket(core), }); this.#handle = handle; this.#watchRunner(handle); @@ -7180,6 +7190,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { diagnosticsDirectory: resolve(this.#root, "diagnostics"), processLauncher: this.options.runnerProcessLauncher, isRunnerConnected: () => core.activeRunnerConnectionCount() > 0, + ...(warmRecovery ? {} : { refreshBootstrapTicket: () => this.#refreshBootstrapTicket(core) }), }); if (handle) { this.#handle = handle; diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index a4fa493a857..da6f970200f 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -576,6 +576,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti "provider_session_open_rejected", "provider_session_open_timeout", "local_runtime_api_listener_unavailable", + "runner_ingress_startup_failed", ]; if ( !coordinator || diff --git a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts index ebdd6c7547f..194784bf039 100644 --- a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts @@ -2,6 +2,7 @@ import { execFile, spawn } from "node:child_process"; import { nativeProviderRecoveryEvidence, nativeSessionFailureSourceCode, restoreRemoteSuspendedRunnerCheckpoint, verifyRemoteRunnerReattachment } from "./native-session-executor.js"; import { readRemoteRunnerRecoveryError } from "./remote-runner-recovery-diagnostics.js"; +import { RunnerIngressConnectionError } from "../../realtime/runner-prp-outbound.js"; import { nativeToolContractFingerprintForTarget, rebindNativeSessionCheckpoint } from "./native-session-resume.js"; import { createHash, randomUUID } from "node:crypto"; import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; @@ -249,6 +250,55 @@ describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable d expect((await evidence("local_runtime_api_listener_unavailable")).recoveryMode).toBe("ambiguous_state"); }); + it("retries a fresh remote start whose launched runner never authenticated through provider ingress", async () => { + const issueId = randomUUID(), runId = randomUUID(), runnerInstanceId = randomUUID(); + const normalizedSessionId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Slow sandbox start", assigneeAgentId: agentId, status: "in_progress" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", + nativeIssueId: issueId, nativeSessionId: normalizedSessionId, runnerInstanceId, + runnerProfileJson: { nativePreparationFencingVersion: 1 } }); + const identity = { runnerInstanceId, environmentLeaseId: randomUUID(), runId, normalizedSessionId, + turnId: `turn-${runId}`, itemId: `item-${runId}` }; + const owner = { companyId, issueId, runId, authorityRunId: runId, bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() }; + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "observed", controllerBootId: owner.bootId, + controllerGeneration: 1, leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) }); + const core = new DurablePrpControlPlane({ stateDirectory: join(root, runId), identity, + expectedRunnerVersion: "discovery-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + core.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, owner)); + const failure = new RunnerIngressConnectionError("runner_ingress_auth_failed", + "Runner ingress did not complete PRP authentication before the deadline."); + const bootstrap = { binding: { companyId, issueId, runId }, identity, transientIngressFailure: false }; + const evidence = () => nativeProviderRecoveryEvidence({ db, runId, executionTargetKind: "remote", + sourceFailureCode: nativeSessionFailureSourceCode(failure), bootstrap }); + try { + const template = { provider: { provider: "codex", command: "fixture" }, authorizedTools: {} }; + core.persistRunAttachTemplate(template); + core.queueCommand("run.prepare", template); + core.queueCommand("session.open", { reuse: "same_session" }); + core.issueBootstrapTicket(60_000); + core.issueBootstrapTicket(60_000); + await core.flushCheckpoint(); + expect(nativeSessionFailureSourceCode(failure)).toBe("runner_ingress_startup_failed"); + expect(nativeSessionFailureSourceCode(new Error("runner_ingress_unavailable: runnerd exited unexpectedly with code 1"))) + .toBe("native_runner_process_exited"); + expect(await evidence()).toMatchObject({ recoveryMode: "bootstrap_retry", providerSessionEstablished: false, + providerEventsExist: false, checkpointExists: false }); + expect(nativeSessionFailureSourceCode(new RunnerIngressConnectionError("runner_connection_lease_expired", + "The runner's connection lease expired before it was renewed.", { cause: failure }))).toBe("native_session_interrupted"); + expect(nativeSessionFailureSourceCode(new RunnerIngressConnectionError("runner_ingress_auth_failed", + "Runner ingress connection failed."))).toBe("runner_ingress_startup_failed"); + expect(await nativeProviderRecoveryEvidence({ db, runId, metadataOnly: true, + sourceFailureCode: "runner_ingress_startup_failed" })).toMatchObject({ recoveryMode: "bootstrap_retry", + providerSessionEstablished: false, providerEventsExist: false, checkpointExists: false }); + await db.update(heartbeatRuns).set({ runnerInstanceId: randomUUID() }).where(eq(heartbeatRuns.id, runId)); + expect((await evidence()).recoveryMode).toBe("ambiguous_state"); + await db.update(heartbeatRuns).set({ runnerInstanceId }).where(eq(heartbeatRuns.id, runId)); + await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 1, stream: "system", level: "info", + eventType: "session.resumed" }); + expect((await evidence()).recoveryMode).toBe("ambiguous_state"); + } finally { await core.stop(); } + }); + it("uses the database clock when an import finds an expired owner", async () => { const fixture = await seed(); const now = Date.now(); diff --git a/server/src/services/native-runtime/native-run-trace.test.ts b/server/src/services/native-runtime/native-run-trace.test.ts index bd448269e37..8954fe10442 100644 --- a/server/src/services/native-runtime/native-run-trace.test.ts +++ b/server/src/services/native-runtime/native-run-trace.test.ts @@ -677,6 +677,28 @@ describe("local runner startup sub-steps", () => { expect(names.indexOf("provider.session.load")).toBeLessThan(names.indexOf("task.provider_session.measured")); }); + it("attributes a remote runner's controller-side startup without duplicating the launcher's spans", async () => { + const events: AdapterRuntimeEvent[] = []; + const { traceContext } = createRecordingTraceContext(); + const trace = createNativeRunTrace({ runId: "remote-startup-steps", startedAtMs: 1_000, traceContext, + onEvent: async (event) => { events.push(event); } }); + const startup = trace.start("runner.session.startup", { parentName: "task.provider_session", startedAtMs: 1_000 }); + recordNativeRunnerStartupStep(trace, { name: "runner.control_plane.open", mode: "bootstrap", startedAtMs: 1_000, endedAtMs: 1_050, outcome: "ok" }, "remote"); + recordNativeRunnerStartupStep(trace, { name: "runner.runtime.stage", mode: "bootstrap", startedAtMs: 1_050, endedAtMs: 601_050, outcome: "ok" }, "remote"); + recordNativeRunnerStartupStep(trace, { name: "runner.process.launch", mode: "bootstrap", startedAtMs: 601_050, endedAtMs: 601_100, outcome: "ok" }, "remote"); + recordNativeRunnerStartupStep(trace, { name: "provider.session.load", mode: "bootstrap", startedAtMs: 601_100, endedAtMs: 601_200, outcome: "ok" }, "remote"); + await trace.end(startup, { endedAtMs: 601_200 }); + await trace.finish("ok"); + const payloads = events.map((event) => event.payload as Record); + expect(payloads.find((payload) => payload.span === "runner.control_plane.open")) + .toMatchObject({ parentSpan: "runner.session.startup", durationMs: 50, target: "remote" }); + expect(payloads.find((payload) => payload.span === "runner.launch.prepare")) + .toMatchObject({ parentSpan: "runner.session.startup", startOffsetMs: 50, durationMs: 600_000, target: "remote" }); + for (const owned of ["runner.runtime.stage", "runner.process.launch", "provider.session.load"]) { + expect(payloads.some((payload) => payload.span === owned)).toBe(false); + } + }); + it("is a no-op without a trace", () => { expect(() => recordNativeRunnerStartupStep(undefined, { name: "runner.process.launch", mode: "bootstrap", startedAtMs: 1, endedAtMs: 2, outcome: "ok" })).not.toThrow(); }); diff --git a/server/src/services/native-runtime/native-run-trace.ts b/server/src/services/native-runtime/native-run-trace.ts index 6267e985f32..9911b158cd0 100644 --- a/server/src/services/native-runtime/native-run-trace.ts +++ b/server/src/services/native-runtime/native-run-trace.ts @@ -787,9 +787,9 @@ export async function recordNativeProviderPerformanceSpans(trace: NativeRunTrace } catch { /* Diagnostics cannot replace the work's result. */ } } -/** One local runner startup sub-step measured by the runner transport. */ +/** One runner startup sub-step measured by the runner transport. */ export interface NativeRunnerStartupStep { - name: "runner.runtime.stage" | "runner.process.launch" | "provider.session.load"; + name: "runner.control_plane.open" | "runner.runtime.stage" | "runner.process.launch" | "provider.session.load"; mode: "bootstrap" | "resume"; startedAtMs: number; endedAtMs: number; @@ -802,14 +802,22 @@ export interface NativeRunnerStartupStep { * Telemetry. Detached: the run-log write never delays `turn.start`; the * trace settles it at finish. */ -export function recordNativeRunnerStartupStep(trace: NativeRunTrace | undefined, step: NativeRunnerStartupStep): void { +export function recordNativeRunnerStartupStep( + trace: NativeRunTrace | undefined, + step: NativeRunnerStartupStep, + target: "local" | "remote" = "local", +): void { + const name = target === "local" ? step.name + : step.name === "runner.runtime.stage" ? "runner.launch.prepare" + : step.name === "runner.control_plane.open" ? step.name : null; + if (!name) return; trace?.defer(() => trace.record({ - name: step.name, + name, parentName: "runner.session.startup", startedAtMs: step.startedAtMs, endedAtMs: step.endedAtMs, outcome: step.outcome, - attributes: { mode: step.mode, target: "local" }, + attributes: { mode: step.mode, target }, })); } diff --git a/server/src/services/native-runtime/native-runner-checkpoint.ts b/server/src/services/native-runtime/native-runner-checkpoint.ts index feb6e6832ac..f7dcaba7c22 100644 --- a/server/src/services/native-runtime/native-runner-checkpoint.ts +++ b/server/src/services/native-runtime/native-runner-checkpoint.ts @@ -3,7 +3,7 @@ import { NATIVE_PREPARATION_FENCING_VERSION } from "./native-active-runs.js"; import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { and, eq, exists, gt, isNull, lte, inArray, notExists, or, sql } from "drizzle-orm"; import { checkpointDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints, type Db } from "@paperclipai/db"; -import { inspectDurablePrpCheckpointHeader, inspectUnclaimedBootstrapCheckpoint, decodeDurablePrpCheckpoint, type DurablePrpCheckpoint, type DurablePrpControlPlane, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js"; +import { inspectDurablePrpCheckpointHeader, inspectUnauthenticatedBootstrapCheckpoint, inspectUnclaimedBootstrapCheckpoint, decodeDurablePrpCheckpoint, type DurablePrpCheckpoint, type DurablePrpControlPlane, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js"; import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js"; import { legacyControllerBootId } from "../legacy-controller-lease.js"; import { assertNativeSourceCheckpointConsumer, lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js"; @@ -261,6 +261,24 @@ export async function loadNativeUnclaimedBootstrapCheckpoint( } catch { return null; } } +export async function nativeRunnerNeverAuthenticated( + db: Pick, binding: Binding, + identity: Pick, +): Promise { + try { + if (binding.runId !== identity.runId) return false; + const rows = await db.select().from(nativeRunnerCheckpoints).where(and( + eq(nativeRunnerCheckpoints.companyId, binding.companyId), eq(nativeRunnerCheckpoints.issueId, binding.issueId), + eq(nativeRunnerCheckpoints.runId, binding.runId))).limit(MAX_BOOTSTRAP_CHECKPOINT_COMMANDS + 3); + if (rows.length < 4 || rows.length > MAX_BOOTSTRAP_CHECKPOINT_COMMANDS + 2 + || rows.reduce((bytes, row) => bytes + Buffer.byteLength(JSON.stringify(row.material)), 0) + > 2 * NATIVE_CHECKPOINT_INSPECTION_TOTAL_MAX_BYTES) return false; + const cipher = createLocalEncryptedCipher("read"); + const records = new Map(rows.map(row => [row.part, decryptPart(cipher, binding, row.part, row.material)] as const)); + return inspectUnauthenticatedBootstrapCheckpoint(records, identity); + } catch { return false; } +} + const BOOTSTRAP_CHECKPOINT_PARTS = new Set(["header", "cursor"]); const MAX_BOOTSTRAP_CHECKPOINT_COMMANDS = 8; /** Session-open events are few; a longer journal is never a bootstrap. */ diff --git a/server/src/services/native-runtime/native-runner-execution-driver.ts b/server/src/services/native-runtime/native-runner-execution-driver.ts index 89a5352c000..f7c6980e202 100644 --- a/server/src/services/native-runtime/native-runner-execution-driver.ts +++ b/server/src/services/native-runtime/native-runner-execution-driver.ts @@ -115,12 +115,14 @@ export function createNativeRunnerExecutionDriver(input: NativeRunnerExecutionDr } : undefined, controlPlaneRegistration: async (authority, attachmentIdentity) => { - await input.beforeConnect?.(attachmentIdentity, authority); - if (input.controllerPersistence && input.binding.issueId) { - await nativeRunnerCheckpointRegistration({ ...input.controllerPersistence, - binding: { companyId: input.binding.companyId, issueId: input.binding.issueId, runId: input.binding.runId }, - })(authority, attachmentIdentity); - } + await measureNativeRunnerSpan(input.trace, "runner.checkpoint.register", async () => { + await input.beforeConnect?.(attachmentIdentity, authority); + if (input.controllerPersistence && input.binding.issueId) { + await nativeRunnerCheckpointRegistration({ ...input.controllerPersistence, + binding: { companyId: input.binding.companyId, issueId: input.binding.issueId, runId: input.binding.runId }, + })(authority, attachmentIdentity); + } + }, { parentName: "runner.session.startup" }); return measureNativeRunnerSpan( input.trace, "runner.transport.connect", diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 81a80f99f1d..71971b8276f 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -1442,6 +1442,25 @@ describe("remote runner process supervision", () => { expect(blockedLog).toHaveBeenCalledWith("stderr", `[paperclip-runner] test bind fault failed on port ${port}\n`); }); + it("launches runnerd with a bootstrap ticket issued after remote staging", async () => { + const order: string[] = []; + const execute = vi.fn(async (request: { args: string[]; env?: Record }) => { + order.push(`execute:${request.env?.PAPERCLIP_RUNNER_BOOTSTRAP_TICKET ?? "none"}`); + return { exitCode: 1, timedOut: false, stdout: "", stderr: "" }; + }); + const handle = createRemoteRunnerProcessLauncher({ + target: { kind: "remote", transport: "sandbox", environmentId: "environment", leaseId: "lease", remoteCwd: "/workspace" }, + runner: { execute } as never, + remoteBinary: "/runtime/runnerd", processIdentityPath: "/runtime/identity", + stateDirectory: "/runtime", diagnosticsDirectory: "/runtime/diagnostics", runnerInstanceId: "runner-remote", + ensureArtifact: async () => { order.push("stage"); }, + })({ command: "/controller/runnerd", args: [], cwd: "/controller", + environment: { PAPERCLIP_RUNNER_BOOTSTRAP_TICKET: "issued-before-staging" }, + refreshBootstrapTicket: async () => { order.push("refresh"); return "issued-at-dispatch"; } }); + await expect(handle.completion).rejects.toThrow("runner_remote_process_launch_failed"); + expect(order).toEqual(["stage", "refresh", "execute:issued-at-dispatch"]); + }); + it("confirms no replacement process exists when staging fails before the launch RPC", async () => { // A full sandbox disk rejects the runtime-context upload before dispatch. const stagingError = new Error( diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 80e36428abb..78615aaa21f 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -159,7 +159,7 @@ import { import { lstat, readFile, readdir, readlink } from "node:fs/promises"; import { hostname, tmpdir, userInfo } from "node:os"; import { DatabaseSync } from "node:sqlite"; -import { NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES, loadNativeRunnerCheckpoint, loadNativeRunnerCheckpointHeader, loadNativeUnclaimedBootstrapCheckpoint, nativeRunnerCheckpointPersistence, nativeRunnerCheckpointRegistration, restoreNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; +import { NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES, loadNativeRunnerCheckpoint, loadNativeRunnerCheckpointHeader, loadNativeUnclaimedBootstrapCheckpoint, nativeRunnerCheckpointPersistence, nativeRunnerNeverAuthenticated, nativeRunnerCheckpointRegistration, restoreNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; import { REMOTE_CONTROLLER_LEASE_MS, REMOTE_RUNNER_RECONNECT_GRACE_MS, localRunnerProcessReconnectGraceMs } from "./native-runner-recovery-contract.js"; import type { NativeTaskSessionPublicationContext } from "./native-task-session-publication.js"; import { basename, dirname, join, posix, resolve } from "node:path"; @@ -8869,6 +8869,7 @@ export function nativeSessionFailureSourceCode( | "runner_task_admission_rejected" | "runner_ingress_access_rejected" | "runner_ingress_acquisition_failed" + | "runner_ingress_startup_failed" | "local_runtime_api_listener_unavailable" | typeof NATIVE_PREDECESSOR_RETIREMENT_UNPROVEN_CODE | "native_session_interrupted" { @@ -9004,9 +9005,19 @@ export function nativeSessionFailureSourceCode( if (/\blocal_runtime_api_listener_unavailable\b/.test(message)) { return "local_runtime_api_listener_unavailable"; } + if (runnerIngressConnectionFailure(error)) return "runner_ingress_startup_failed"; return "native_session_interrupted"; } +function runnerIngressConnectionFailure(error: unknown, depth = 0): boolean { + if (!error || typeof error !== "object" || depth > 5) return false; + const value = error as Record; + if (value.code === "runner_connection_lease_expired") return false; + if (typeof value.message === "string" + && /^runner_ingress_(?:auth_failed|unavailable): (?:Runner ingress did not complete PRP authentication before the deadline|Runner ingress connection failed)\.$/.test(value.message)) return true; + return runnerIngressConnectionFailure(value.cause, depth + 1); +} + const NATIVE_PROVIDER_SIDE_EFFECT_PENDING_ACTION = "The saved provider session has an unfinished provider turn, or a tool call whose result never reached the runner, so Paperclip did not replace the session or retry it. Reconcile the action: check whether it ran and record the outcome. The task then continues in a fresh provider session in the same workspace. The saved session and workspace are unchanged."; @@ -9139,7 +9150,10 @@ export async function nativeProviderRecoveryEvidence(input: { NATIVE_PROVIDER_STARTUP_REJECTED, "provider_session_open_timeout", "local_runtime_api_listener_unavailable", - ]).has(input.sourceFailureCode); + ]).has(input.sourceFailureCode) || (input.sourceFailureCode === "runner_ingress_startup_failed" + && !checkpointExists && !providerEventsExist && !!run?.nativeIssueId && !!run.nativeSessionId && !!run.runnerInstanceId + && await nativeRunnerNeverAuthenticated(input.db, { companyId: run.companyId, issueId: run.nativeIssueId, runId: run.id }, + { runId: run.id, normalizedSessionId: run.nativeSessionId, runnerInstanceId: run.runnerInstanceId })); // A closed runtime API listener means this server stopped serving (a // graceful drain) before the attempt could reach its remote runner. A // follow-up may still carry its settled predecessor's provider session, @@ -15290,6 +15304,11 @@ export function createRemoteRunnerProcessLauncher(input: { } } if (pendingSignal) return cancelledBeforeDispatch(); + const bootstrapTicket = spec.refreshBootstrapTicket ? await spec.refreshBootstrapTicket() : null; + if (pendingSignal) return cancelledBeforeDispatch(); + const launchEnvironment = bootstrapTicket + ? { ...spec.environment, PAPERCLIP_RUNNER_BOOTSTRAP_TICKET: bootstrapTicket } + : spec.environment; const launchStartedAtMs = Date.now(); // The provider's onSpawn callback is optional and some sandbox command // runners cannot report a remote pid until after the command has begun @@ -15340,7 +15359,7 @@ export function createRemoteRunnerProcessLauncher(input: { ...remoteArgs, ], cwd: input.target.remoteCwd, - env: processEnvironment(spec.environment), + env: processEnvironment(launchEnvironment), timeoutMs: 20_000, bypassSession: true, onLog: input.onLog, @@ -17820,7 +17839,8 @@ async function createRunnerdBackendWithinSessionClaim( measureOperation: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? measureRunnerOperation : undefined, recordTimingReceipt: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? recordRunnerTimingReceipt : undefined, // Remote launchers record their own stage and launch spans. - ...(remoteTarget ? {} : { onStartupStep: (step: Parameters[1]) => recordNativeRunnerStartupStep(input.trace, step) }), + onStartupStep: (step: Parameters[1]) => + recordNativeRunnerStartupStep(input.trace, step, remoteTarget ? "remote" : "local"), onSpawn: input.onSpawn, ...(input.execution.provider.kind === "codex" ? { disableCodexWebSearch: "webSearch" in input.execution.provider && input.execution.provider.webSearch === "disabled" } : {}), homeMode: preconfiguredHome ? "preconfigured" : "isolated", diff --git a/server/src/services/native-runtime/remote-launch-assets.ts b/server/src/services/native-runtime/remote-launch-assets.ts index 2d5e85e4905..6c787fa3abf 100644 --- a/server/src/services/native-runtime/remote-launch-assets.ts +++ b/server/src/services/native-runtime/remote-launch-assets.ts @@ -171,7 +171,8 @@ export async function stageRemoteLaunchAssets(input: LaunchAssetsInput): Promise const directory = input.persistenceDirectories.find(entry => entry.name === "codex-home" && entry.location === "filesystem"); if (!directory?.targetPath) throw new Error("runner_harness_state_mismatch: bootstrap_target_unavailable"); const sourcePath = resolve(input.root, "codex-home", "skills"); - await assertSelectedSkillsTree(sourcePath, input.runtimeContext?.source.skills.map(skill => skill.runtimeName) ?? []); + await measure("stage.asset.codex_skills.verify", () => + assertSelectedSkillsTree(sourcePath, input.runtimeContext?.source.skills.map(skill => skill.runtimeName) ?? [])); await measure("stage.asset.codex_skills", async () => { const cleared = await input.runner.execute({ command: "node", diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts index c528316cae7..ba3fe4e7e17 100644 --- a/server/src/vendor/paperclip-runner/index.ts +++ b/server/src/vendor/paperclip-runner/index.ts @@ -98,6 +98,7 @@ export const decodeDurablePrpCheckpoint = runner.decodeDurablePrpCheckpoint; export const captureLegacyDurablePrpCheckpoint = runner.captureLegacyDurablePrpCheckpoint; export const inspectDurablePrpCheckpointHeader = runner.inspectDurablePrpCheckpointHeader; export const inspectUnclaimedBootstrapCheckpoint = runner.inspectUnclaimedBootstrapCheckpoint; +export const inspectUnauthenticatedBootstrapCheckpoint = runner.inspectUnauthenticatedBootstrapCheckpoint; export const inspectUnclaimedBootstrapState = runner.inspectUnclaimedBootstrapState; export const RUNNER_PRP_PROTOCOL_MIN = runner.RUNNER_PRP_PROTOCOL_MIN; export const RUNNER_PRP_PROTOCOL_MAX = runner.RUNNER_PRP_PROTOCOL_MAX;