diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index 953c126de55..4171bcd5842 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -14,6 +14,31 @@ session and issue-thread surfaces, a public browser/React SDK, a standalone adapter demo, and a deterministic mock control plane. None of these surfaces imports or starts Paperclip's server, UI, CLI, or production database. +## Execution environment lifecycle + +`RunnerExecutionDriver` defines the environment hooks shared by coding harnesses +and external bot harnesses. The server's `createNativeRunnerExecutionDriver` +supplies this contract for local, SSH, and sandbox execution. Provider-specific +session setup stays separate from environment selection. + +Connection registration prepares remote resources and returns the process's +connection configuration before launch. Local Runners connect to Core over +loopback WS. SSH Runners and eligible sandboxes connect to the configured Core +WSS endpoint. Sandboxes with authorized WebSocket ingress listen inside the +sandbox; Core connects through the provider endpoint after process dispatch. +Ingress selection fails closed and does not fall back to outbound WSS. + +The returned activation, readiness, failure, checkpoint, and release hooks feed +the existing shared PRP control plane. PRP owns authentication, commands, replay, +tool authority, and run transitions. Process ownership hooks provide launch, +verified adoption, and durable state access. A lost connection does not prove +process termination, and a required checkpoint failure must prevent lease release. + +The driver is an in-process interface. Its filesystem paths and process launch +specifications are internal to built-in execution; they are not a plugin wire +contract. Existing flat transport options remain compatible through this shared +interface. This refactor does not add Launcher APIs or change harness capabilities. + Connection continuations inspect the harness descriptor's optional `toolRefreshOnResume` capability. Native Codex, Claude Managed Agents, AgentCore, OpenCode, and qualified Claude/Codex/Grok ACPX profiles expose it; unqualified diff --git a/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts new file mode 100644 index 00000000000..13d303dee05 --- /dev/null +++ b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts @@ -0,0 +1,59 @@ +import type { DurablePrpControlPlane, RunnerProcessConnection, RunnerProcessHandle, RunnerProcessLaunchSpec } from "./durable-prp-control-plane.js"; +import type { DurableRecoveryIdentity } from "../contracts/durable-recovery.js"; + +/** Environment ownership shared by local, SSH, and sandbox Runner execution. + * Preparation belongs to connection registration, before launch; activation + * happens after dispatch. PRP authority and session semantics stay with Core. + * Filesystem paths and process specifications are private to the built-in driver, + * not a serializable plugin API. + */ +export interface RunnerExecutionDriver { + /** Prepare resources and register or acquire the connection before launch. */ + controlPlaneRegistration?: ( + authority: DurablePrpControlPlane, + identity?: DurableRecoveryIdentity, + signal?: AbortSignal, + ) => Promise<{ + connectUrl?: string; + connection?: RunnerProcessConnection; + activate?: () => Promise | void; + ready?: () => Promise; + failure?: Promise; + startupFailureCode?: + | "runner_local_connect_failed" + | "runner_direct_wss_failed" + | "runner_ingress_unavailable"; + /** Persists only exact, independently verified suspended remote state. */ + checkpoint?: (settlement: "settled" | "unsettled") => Promise | void; + release: () => Promise | void; + }>; + /** Override process ownership; omission uses the SDK's local process owner. */ + runnerProcessLauncher?: ( + spec: RunnerProcessLaunchSpec, + ) => RunnerProcessHandle; + /** Durable runner state path in the process owner's filesystem. */ + runnerStateDirectory?: string; + /** Read the live durable runner state when runnerd owns a remote filesystem. */ + readRunnerState?: () => Promise>; + /** Materializes a verified external checkpoint before authority rotation. */ + prepareExternalRunnerState?: () => Promise; + /** Idempotently archives and returns the verified suspended runner binding. */ + archiveExternalRunnerState?: (input: { + archiveKey: string; + priorIdentity: DurableRecoveryIdentity; + }) => Promise>; + /** Active-connection recovery budget. Omitted for the existing local mode. */ + runnerReconnectGraceMs?: number; + /** + * A verified Runner that outlived its controller. Adoption registers + * the durable authority and waits for this exact process to reconnect; it + * never calls the process launcher while the process remains alive. + */ + adoptExistingRunner?: { + pid: number; + processGroupId: number | null; + startedAt: string; + isAlive: () => Promise | boolean; + signal?: (signal: NodeJS.Signals) => Promise | boolean; + }; +} diff --git a/packages/paperclip-runner/src/drivers/dot/runnerd-dot-driver.ts b/packages/paperclip-runner/src/drivers/dot/runnerd-dot-driver.ts index 27f5ba087b7..676738012fa 100644 --- a/packages/paperclip-runner/src/drivers/dot/runnerd-dot-driver.ts +++ b/packages/paperclip-runner/src/drivers/dot/runnerd-dot-driver.ts @@ -1,3 +1,4 @@ +import type { RunnerExecutionDriver } from "../../control-plane/runner-execution-driver.js"; import { existsSync, mkdirSync, readFileSync, statSync } from "node:fs"; import { randomUUID } from "node:crypto"; import { resolve } from "node:path"; @@ -8,27 +9,23 @@ import type { PrpEvent } from "../../protocol/replay-contract.js"; import { validatePrpStructuredRunResult } from "../../protocol/replay-contract.js"; import { DurablePrpControlPlane, spawnRunner, type RunnerProcessHandle } from "../../control-plane/durable-prp-control-plane.js"; import type { DurableRecoveryIdentity } from "../../contracts/durable-recovery.js"; -import { authorizedToolSetForProvider, defaultCapabilityRunnerdBinary, readRunnerdArtifactBinding, type CapabilityRunnerdCodexTransportOptions } from "../../live/runnerd-codex-transport.js"; +import { authorizedToolSetForProvider, defaultCapabilityRunnerdBinary, readRunnerdArtifactBinding } from "../../live/runnerd-codex-transport.js"; import { codexSemanticToolSpecs } from "../codex/codex-driver-values.js"; import { nativeSystemInstructions } from "../../backends/runtime-context.js"; import type { CodexNativeSessionBackendOptions } from "../../backends/codex-native-backend.js"; -export interface RunnerdDotDriverOptions { +export interface RunnerdDotDriverOptions extends Pick { execution: NativeExecutionInputV6; stateDirectory: string; identity: DurableRecoveryIdentity; port: ExternalProviderPort; runnerBinary?: string; - runnerStateDirectory?: string; - runnerProcessLauncher?: CapabilityRunnerdCodexTransportOptions["runnerProcessLauncher"]; /** Reads the checkpoint from the execution target. Missing state cannot authorize replay. */ readProviderState?: () => Promise | null>; - controlPlaneRegistration?: CapabilityRunnerdCodexTransportOptions["controlPlaneRegistration"]; onSpawn?: CodexNativeSessionBackendOptions["onSpawn"]; dynamicTools?: CodexNativeSessionBackendOptions["dynamicTools"]; dynamicToolHandler?: CodexNativeSessionBackendOptions["dynamicToolHandler"]; completionFeedback?: CodexNativeSessionBackendOptions["completionFeedback"]; - adoptExistingRunner?: CapabilityRunnerdCodexTransportOptions["adoptExistingRunner"]; } /** Keep the remote monitor's closed failure code without logging its stderr, diff --git a/packages/paperclip-runner/src/index.ts b/packages/paperclip-runner/src/index.ts index 32103f88d94..21329192af4 100644 --- a/packages/paperclip-runner/src/index.ts +++ b/packages/paperclip-runner/src/index.ts @@ -92,3 +92,5 @@ export { probeAcpxClaudeInstallation, probeAcpxGrokInstallation, probeAcpxPiInst export { bundledRemoteProviderPackManifestPath, bundledRemoteRunnerBinary } from "./live/bundled-remote-provider-pack.js"; export { RunnerdDotDriver, type RunnerdDotDriverOptions } from "./drivers/dot/runnerd-dot-driver.js"; + +export type { RunnerExecutionDriver } from "./control-plane/runner-execution-driver.js"; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index d824236787c..2fd55d73ddc 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -1,3 +1,4 @@ +import type { RunnerExecutionDriver } from "../control-plane/runner-execution-driver.js"; import { resolvePinnedOpenCodeCommand } from "../drivers/opencode/opencode-server-driver.js"; import { ACPX_CAPABILITY_PROFILES } from "../drivers/acpx/capability-profiles.js"; import { admittedPiThinkingLevel, resolvePiThinkingLevel } from "../drivers/acpx/pi-thinking.js"; @@ -64,7 +65,6 @@ import { waitForProcess, type RunnerProcessHandle, type RunnerProcessConnection, - type RunnerProcessLaunchSpec, type DurablePrpControlPlaneOptions, } from "../control-plane/durable-prp-control-plane.js"; import { @@ -1228,7 +1228,7 @@ export interface CapabilityRunnerdProcessEvidence { diagnostics: string[]; } -export interface CapabilityRunnerdCodexTransportOptions { +export interface CapabilityRunnerdCodexTransportOptions extends RunnerExecutionDriver { provider?: "codex" | "opencode" | "claude_managed" | "aws_agentcore" | "acpx"; opencodePermissionMode?: NativeOpenCodePermissionMode; acpxAgent?: QualifiedAcpxAgent; @@ -1354,25 +1354,6 @@ export interface CapabilityRunnerdCodexTransportOptions { turnId: string; itemId: string; }; - /** Registers the run-bound PRP authority on Paperclip's shared HTTP server. */ - controlPlaneRegistration?: ( - authority: DurablePrpControlPlane, - identity?: DurableRecoveryIdentity, - signal?: AbortSignal, - ) => Promise<{ - connectUrl?: string; - connection?: RunnerProcessConnection; - activate?: () => Promise | void; - ready?: () => Promise; - failure?: Promise; - startupFailureCode?: - | "runner_local_connect_failed" - | "runner_direct_wss_failed" - | "runner_ingress_unavailable"; - /** Persists only exact, independently verified suspended remote state. */ - checkpoint?: (settlement: "settled" | "unsettled") => Promise | void; - release: () => Promise | void; - }>; /** Existing server-owned routes only; never provision two ingress owners. */ warmTransitionRegistrationMode?: "routed_connect"; /** Rechecks the server-owned recovery claim after each asynchronous boundary. */ @@ -1383,35 +1364,6 @@ export interface CapabilityRunnerdCodexTransportOptions { onWarmTransitionRecoveryCompleted?: (input: { transitionId: string; }) => void | Promise; - /** Optional remote process owner used only by the new runner coordinator. */ - runnerProcessLauncher?: ( - spec: RunnerProcessLaunchSpec, - ) => RunnerProcessHandle; - /** Durable runner state path in the process owner's filesystem. */ - runnerStateDirectory?: string; - /** Read the live durable runner state when runnerd owns a remote filesystem. */ - readRunnerState?: () => Promise>; - /** Materializes a verified external checkpoint before authority rotation. */ - prepareExternalRunnerState?: () => Promise; - /** Idempotently archives and returns the verified suspended runner binding. */ - archiveExternalRunnerState?: (input: { - archiveKey: string; - priorIdentity: DurableRecoveryIdentity; - }) => Promise>; - /** Active-connection recovery budget. Omitted for the existing local mode. */ - runnerReconnectGraceMs?: number; - /** - * A verified local runner that outlived its controller. Adoption registers - * the durable authority and waits for this exact process to reconnect; it - * never calls the process launcher while the process remains alive. - */ - adoptExistingRunner?: { - pid: number; - processGroupId: number | null; - startedAt: string; - isAlive: () => Promise | boolean; - signal?: (signal: NodeJS.Signals) => Promise | boolean; - }; } export type RunnerdCodexTransportOptions = diff --git a/server/src/services/native-runtime/native-runner-execution-driver.test.ts b/server/src/services/native-runtime/native-runner-execution-driver.test.ts new file mode 100644 index 00000000000..830544e863b --- /dev/null +++ b/server/src/services/native-runtime/native-runner-execution-driver.test.ts @@ -0,0 +1,183 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import type { RunnerIngressEndpoint } from "@paperclipai/adapter-utils/runner-connectivity"; +import { createNativeRunnerExecutionDriver } from "./native-runner-execution-driver.js"; + +const transport = vi.hoisted(() => ({ + register: vi.fn(), + connect: vi.fn(), +})); +vi.mock("../../realtime/runner-prp-ws.js", () => ({ registerRunnerPrpAuthority: transport.register })); +vi.mock("../../realtime/runner-prp-outbound.js", () => ({ connectRunnerPrpIngress: transport.connect })); + +const binding = { companyId: "company", issueId: "issue", agentId: "agent", runId: "run-a" }; +const authority = {} as Parameters["controlPlaneRegistration"]>>[0]; +const ssh = { kind: "remote", transport: "ssh", remoteCwd: "/workspace", spec: {} } as AdapterExecutionTarget; +const sandbox = (ingress = true): AdapterExecutionTarget => ({ + kind: "remote", transport: "sandbox", remoteCwd: "/workspace", leaseId: "lease", + effectiveCapabilities: { runnerWebSocketIngress: ingress }, +} as AdapterExecutionTarget); + +function endpoint(): RunnerIngressEndpoint { + return { + kind: "authenticated_websocket", websocketUrl: "wss://sandbox.example/runner", + secretHeaders: [{ name: "Authorization", value: "test-ingress-secret" }], generation: "1", + refresh: vi.fn(), close: vi.fn(async () => {}), + }; +} + +beforeEach(() => { + vi.clearAllMocks(); + transport.register.mockResolvedValue({ connectUrl: "ws://127.0.0.1/core", release: vi.fn() }); + transport.connect.mockReturnValue({ ready: Promise.resolve(), close: vi.fn(async () => {}) }); +}); + +describe("shared Runner execution driver", () => { + it("keeps local execution on the registered loopback route", async () => { + const driver = createNativeRunnerExecutionDriver({ target: { kind: "local" }, binding }); + const connection = await driver.controlPlaneRegistration!(authority); + expect(connection.connectUrl).toBe("ws://127.0.0.1/core"); + expect(connection.startupFailureCode).toBe("runner_local_connect_failed"); + expect(transport.register).toHaveBeenCalledWith({ ...binding, authority }); + expect(transport.connect).not.toHaveBeenCalled(); + }); + + it.each([ssh, sandbox(false)])("prepares $transport before launching an outbound Runner", async target => { + const prepare = vi.fn(async () => {}); + const launch = vi.fn(); + const read = vi.fn(); + const checkpoint = vi.fn(); + const mapCaBundle = vi.fn(async () => "/remote/ca.pem"); + const driver = createNativeRunnerExecutionDriver({ + target, binding, runnerPublicUrl: "wss://core.example", prepareRemote: prepare, + process: { runnerProcessLauncher: launch, readRunnerState: read }, checkpoint, mapCaBundle, + }); + const connection = await driver.controlPlaneRegistration!(authority); + expect(prepare).toHaveBeenCalledWith("dial_wss"); + expect(connection.connection).toEqual({ mode: "connect", connectUrl: "wss://core.example/api/runner/v1/connect/run-a" }); + expect(connection.checkpoint).toBe(checkpoint); + expect(driver.runnerProcessLauncher).toBe(launch); + expect(driver.readRunnerState).toBe(read); + expect(launch).not.toHaveBeenCalled(); + expect(mapCaBundle).not.toHaveBeenCalled(); + expect(transport.connect).not.toHaveBeenCalled(); + expect(prepare.mock.invocationCallOrder[0]).toBeLessThan(transport.register.mock.invocationCallOrder[0]); + }); + + it("rejects invalid outbound connectivity before staging resources", async () => { + const prepare = vi.fn(); + const driver = createNativeRunnerExecutionDriver({ target: ssh, binding, prepareRemote: prepare }); + await expect(driver.controlPlaneRegistration!(authority)).rejects.toThrow("runner_transport_ineligible"); + expect(prepare).not.toHaveBeenCalled(); + }); + + it("rejects a missing CA file before preparing the remote environment", async () => { + const prepare = vi.fn(); + const driver = createNativeRunnerExecutionDriver({ + target: ssh, binding, prepareRemote: prepare, runnerPublicUrl: "wss://core.example", + runnerCaBundlePath: "/nonexistent/paperclip-runner-test-ca.pem", + }); + await expect(driver.controlPlaneRegistration!(authority)).rejects.toThrow("runner_direct_wss_failed"); + expect(prepare).not.toHaveBeenCalled(); + expect(transport.register).not.toHaveBeenCalled(); + }); + + it("binds an outbound CA to the process owner's filesystem", async () => { + const mapCaBundle = vi.fn(async () => "/remote/ca.pem"); + const driver = createNativeRunnerExecutionDriver({ + target: ssh, binding, prepareRemote: async () => {}, runnerPublicUrl: "wss://core.example", + runnerCaBundlePath: process.execPath, mapCaBundle, + }); + const connection = await driver.controlPlaneRegistration!(authority); + expect(mapCaBundle).toHaveBeenCalledWith(process.execPath); + expect(connection.connection).toEqual(expect.objectContaining({ caBundlePath: "/remote/ca.pem" })); + expect(mapCaBundle.mock.invocationCallOrder[0]).toBeLessThan(transport.register.mock.invocationCallOrder[0]); + }); + + it("does not register an authority route when CA staging fails", async () => { + const driver = createNativeRunnerExecutionDriver({ + target: ssh, binding, prepareRemote: async () => {}, runnerPublicUrl: "wss://core.example", + runnerCaBundlePath: process.execPath, + mapCaBundle: async () => { throw new Error("ca_copy_failed"); }, + }); + await expect(driver.controlPlaneRegistration!(authority)).rejects.toThrow("ca_copy_failed"); + expect(transport.register).not.toHaveBeenCalled(); + }); + + it("stages before acquiring ingress and waits for process dispatch before connecting", async () => { + const ingress = endpoint(); + const acquire = vi.fn(async () => ingress); + const prepare = vi.fn(async () => {}); + let started!: () => void; + const processStarted = new Promise(resolve => { started = resolve; }); + const driver = createNativeRunnerExecutionDriver({ + target: { ...sandbox(), getRunnerIngressEndpoint: acquire } as AdapterExecutionTarget, + binding, runnerIngressAuthorized: true, prepareRemote: prepare, processStarted, + }); + const connection = await driver.controlPlaneRegistration!(authority); + expect(prepare).toHaveBeenCalledWith("listen_ws"); + expect(prepare.mock.invocationCallOrder[0]).toBeLessThan(acquire.mock.invocationCallOrder[0]); + expect(connection.connection?.mode).toBe("listen"); + await connection.activate!(); + expect(transport.connect).not.toHaveBeenCalled(); + started(); + await connection.ready!(); + expect(transport.connect).toHaveBeenCalledWith(expect.objectContaining({ authority, endpoint: ingress })); + expect(transport.register).not.toHaveBeenCalled(); + await connection.release(); + expect(transport.connect.mock.results[0].value.close).toHaveBeenCalledOnce(); + }); + + it("does not fall back to outbound WSS when sandbox ingress is unauthorized", async () => { + const prepare = vi.fn(); + const acquire = vi.fn(); + const driver = createNativeRunnerExecutionDriver({ + target: { ...sandbox(), getRunnerIngressEndpoint: acquire } as AdapterExecutionTarget, + binding, runnerPublicUrl: "wss://core.example", prepareRemote: prepare, + }); + await expect(driver.controlPlaneRegistration!(authority)).rejects.toThrow("runner_ingress_unavailable"); + expect(prepare).not.toHaveBeenCalled(); + expect(acquire).not.toHaveBeenCalled(); + expect(transport.register).not.toHaveBeenCalled(); + }); + + it("closes acquired ingress when setup ends before activation", async () => { + const ingress = endpoint(); + const driver = createNativeRunnerExecutionDriver({ + target: { ...sandbox(), getRunnerIngressEndpoint: async () => ingress } as AdapterExecutionTarget, + binding, runnerIngressAuthorized: true, prepareRemote: async () => {}, processStarted: Promise.resolve(), + }); + const connection = await driver.controlPlaneRegistration!(authority); + await connection.release(); + expect(ingress.close).toHaveBeenCalledOnce(); + expect(transport.connect).not.toHaveBeenCalled(); + }); + + it("uses the newly authorized run binding during warm registration", async () => { + const beforeConnect = vi.fn(async () => {}); + const identity = { runId: "run-b" } as Parameters["controlPlaneRegistration"]>>[1]; + const driver = createNativeRunnerExecutionDriver({ target: { kind: "local" }, binding, beforeConnect }); + await driver.controlPlaneRegistration!(authority, identity); + expect(beforeConnect).toHaveBeenCalledWith(identity); + expect(transport.register).toHaveBeenCalledWith({ ...binding, runId: "run-b", authority }); + expect(beforeConnect.mock.invocationCallOrder[0]).toBeLessThan(transport.register.mock.invocationCallOrder[0]); + }); + + it("blocks registration if warm recovery authority is no longer valid", async () => { + const driver = createNativeRunnerExecutionDriver({ + target: { kind: "local" }, binding, + beforeConnect: async () => { throw new Error("recovery_denied"); }, + }); + await expect(driver.controlPlaneRegistration!(authority)).rejects.toThrow("recovery_denied"); + expect(transport.register).not.toHaveBeenCalled(); + }); + + it("requires exact process identity before admitting an external provider", async () => { + const driver = createNativeRunnerExecutionDriver({ + target: ssh, binding, runnerPublicUrl: "wss://core.example", prepareRemote: async () => {}, + processIdentityReady: Promise.reject(new Error("process_identity_invalid")), + }); + const connection = await driver.controlPlaneRegistration!(authority); + await expect(connection.ready!()).rejects.toThrow("process_identity_invalid"); + }); +}); diff --git a/server/src/services/native-runtime/native-runner-execution-driver.ts b/server/src/services/native-runtime/native-runner-execution-driver.ts new file mode 100644 index 00000000000..e6c456e4c52 --- /dev/null +++ b/server/src/services/native-runtime/native-runner-execution-driver.ts @@ -0,0 +1,301 @@ +import { existsSync } from "node:fs"; +import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import { resolvePaperclipRunnerTransport, type PaperclipRunnerTransport } from "@paperclipai/adapter-utils/runner-connectivity"; +import type { RunnerExecutionDriver } from "../../vendor/paperclip-runner/index.js"; +import { registerRunnerPrpAuthority } from "../../realtime/runner-prp-ws.js"; +import { connectRunnerPrpIngress } from "../../realtime/runner-prp-outbound.js"; +import type { NativeRunTrace } from "./native-run-trace.js"; + +type ConnectionRegistration = NonNullable; +type RunnerIdentity = Parameters[1]; + +export interface NativeRunnerExecutionDriverInput { + target: AdapterExecutionTarget; + binding: Omit[0], "authority">; + runnerIngressAuthorized?: boolean; + runnerPublicUrl?: string | null; + runnerCaBundlePath?: string | null; + /** Validate and stage environment resources before process dispatch. */ + prepareRemote?: (mode: "listen_ws" | "dial_wss") => Promise; + mapCaBundle?: (sourcePath: string) => string | Promise; + processStarted?: Promise; + processIdentityReady?: Promise; + beforeConnect?: (identity?: RunnerIdentity) => Promise; + checkpoint?: Awaited>["checkpoint"]; + process?: Omit; + trace?: NativeRunTrace; + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; +} + +async function measureNativeRunnerSpan( + trace: NativeRunTrace | undefined, + name: string, + fn: () => Promise, + options: { parentName?: string } = {}, +): Promise { + return trace ? trace.measure(name, fn, options) : fn(); +} + +/** + * Select the remote runner transport before any artifact is staged or provider + * endpoint is acquired. Sandbox ingress is available only to a run already + * authorized by native runtime selection. + */ +export function resolveRemoteRunnerTransportMode(input: { + target: AdapterExecutionTarget; + runnerIngressAuthorized: boolean; +}): "listen_ws" | "dial_wss" { + if (input.target.kind !== "remote") { + throw new Error("runner_transport_ineligible: remote target is required"); + } + const requiredMode = + input.target.transport === "sandbox" && + input.target.effectiveCapabilities?.runnerWebSocketIngress === true + ? "listen_ws" + : "dial_wss"; + if (requiredMode === "listen_ws" && !input.runnerIngressAuthorized) { + throw new Error("runner_ingress_unavailable"); + } + return requiredMode; +} + +/** Built-in environment lifecycle. PRP commands and authentication are shared; + * only environment preparation, process ownership, and connection routing vary. + */ +export function createNativeRunnerExecutionDriver(input: NativeRunnerExecutionDriverInput): RunnerExecutionDriver { + return { + ...input.process, + controlPlaneRegistration: async (authority, attachmentIdentity) => { + await input.beforeConnect?.(attachmentIdentity); + return measureNativeRunnerSpan( + input.trace, + "runner.transport.connect", + async () => { + if (input.target.kind === "local") { + const selectedAtMs = Date.now(); + await input.trace?.record({ + name: "runner.transport.selected", + parentName: "runner.transport.connect", + startedAtMs: selectedAtMs, + endedAtMs: selectedAtMs, + attributes: { + mode: "local_loopback", + connectionOwner: "runnerd", + }, + }); + await input.onLog?.( + "stderr", + "[paperclip-runner] transport mode=local_loopback state=connecting\n", + ); + const registration = await measureNativeRunnerSpan( + input.trace, + "runner.prp.route.register", + () => + registerRunnerPrpAuthority({ + ...input.binding, + runId: + attachmentIdentity?.runId ?? + input.binding.runId, + authority, + }), + ); + return { + ...registration, + startupFailureCode: "runner_local_connect_failed" as const, + }; + } + + if (!input.prepareRemote) throw new Error("runner_remote_preparation_unavailable"); + const requiredMode = resolveRemoteRunnerTransportMode({ + target: input.target, + runnerIngressAuthorized: input.runnerIngressAuthorized === true, + }); + let transport: PaperclipRunnerTransport; + if (requiredMode === "dial_wss") { + // Validate eligibility before staging any artifact. + transport = await measureNativeRunnerSpan( + input.trace, + "runner.transport.resolve", + () => + resolvePaperclipRunnerTransport({ + target: input.target, + runId: + attachmentIdentity?.runId ?? + input.binding.runId, + localConnectUrl: "ws://127.0.0.1/unused", + runnerPublicUrl: input.runnerPublicUrl, + runnerCaBundlePath: input.runnerCaBundlePath, + runnerIngressAuthorized: + input.runnerIngressAuthorized === true, + }), + ); + if ( + transport.mode === "direct_outbound" && + transport.caBundlePath && + !existsSync(transport.caBundlePath) + ) { + throw new Error( + "runner_direct_wss_failed: configured runner CA bundle is unavailable", + ); + } + await measureNativeRunnerSpan( + input.trace, + "runner.artifact.prepare", + () => input.prepareRemote!(requiredMode), + ); + } else { + await measureNativeRunnerSpan( + input.trace, + "runner.artifact.prepare", + () => input.prepareRemote!(requiredMode), + ); + // Provider endpoint acquisition happens only after runnerd is staged + // and its listener capability has been verified. + transport = await measureNativeRunnerSpan( + input.trace, + "runner.ingress.acquire", + () => + resolvePaperclipRunnerTransport({ + target: input.target, + runId: + attachmentIdentity?.runId ?? + input.binding.runId, + localConnectUrl: "ws://127.0.0.1/unused", + runnerPublicUrl: input.runnerPublicUrl, + runnerCaBundlePath: input.runnerCaBundlePath, + runnerIngressAuthorized: true, + }), + ); + } + + const selectedAtMs = Date.now(); + await input.trace?.record({ + name: "runner.transport.selected", + parentName: "runner.transport.connect", + startedAtMs: selectedAtMs, + endedAtMs: selectedAtMs, + attributes: { + mode: transport.mode, + connectionOwner: + transport.mode === "provider_ingress" + ? "paperclip" + : "runnerd", + }, + }); + + await input.onLog?.( + "stderr", + `[paperclip-runner] transport mode=${transport.mode} state=connecting\n`, + ); + + if (transport.mode === "direct_outbound") { + const caBundlePath = transport.caBundlePath && input.mapCaBundle + ? await input.mapCaBundle(transport.caBundlePath) + : transport.caBundlePath; + const inbound = await measureNativeRunnerSpan( + input.trace, + "runner.prp.route.register", + () => + registerRunnerPrpAuthority({ + ...input.binding, + runId: + attachmentIdentity?.runId ?? + input.binding.runId, + authority, + }), + { parentName: "runner.transport.connect" }, + ); + return { + connection: { + mode: "connect" as const, + connectUrl: transport.connectUrl, + ...(caBundlePath ? { caBundlePath } : {}), + }, + startupFailureCode: "runner_direct_wss_failed" as const, + checkpoint: input.checkpoint, + ready: input.processIdentityReady ? async () => { await input.processIdentityReady; } : undefined, + release: () => inbound.release(), + }; + } + + if (transport.mode !== "provider_ingress") { + throw new Error("runner_transport_mode_changed_after_dispatch"); + } + if (!input.processStarted) { + await transport.ingress.close(); + throw new Error("runner_process_start_barrier_unavailable"); + } + let outbound: ReturnType | null = + null; + let activation: Promise | null = null; + return { + connection: { + mode: "listen" as const, + listenAddress: transport.listenAddress, + listenPort: transport.listenPort, + listenPath: transport.listenPath, + }, + activate: () => { + activation = measureNativeRunnerSpan( + input.trace, + "runner.transport.activation", + async () => { + await measureNativeRunnerSpan( + input.trace, + "runner.ingress.wait_for_process", + () => input.processStarted!, + ); + outbound = connectRunnerPrpIngress({ + authority, + endpoint: transport.ingress, + onStateChange: (state, failureCode) => { + void input.onLog?.( + "stderr", + `[paperclip-runner] transport mode=provider_ingress state=${state}${failureCode ? ` failure=${failureCode}` : ""}\n`, + ); + }, + }); + }, + { parentName: "runner.session.startup" }, + ); + // Cancellation can release the sandbox while activation is + // still waiting for the remote runner process. Attach a + // rejection observer immediately; `ready()` still awaits the + // original promise and reports the same startup failure, but + // an early teardown can no longer crash the controller with + // an unhandled plugin RPC rejection. + void activation.catch(() => undefined); + }, + ready: async () => { + await measureNativeRunnerSpan( + input.trace, + "runner.transport.ready", + async () => { + await activation; + if (!outbound) + throw new Error("runner_ingress_unavailable"); + await measureNativeRunnerSpan( + input.trace, + "runner.prp.authenticate", + () => outbound!.ready, + ); + await input.processIdentityReady; + }, + { parentName: "runner.session.startup" }, + ); + }, + get failure() { + return outbound?.failure; + }, + startupFailureCode: "runner_ingress_unavailable" as const, + checkpoint: input.checkpoint, + release: async () => { + if (outbound) await outbound.close(); + else await transport.ingress.close(); + }, + }; + }, + ); + }, + }; +} diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6729158c6c7..d1da1cd7b1a 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -109,10 +109,7 @@ import { createNativeSshCommandRunner } from "./native-ssh-command-runner.js"; import { discoverRemoteExecutable, parseRemoteExecutableCandidate } from "./codex-model-fallback.js"; export { parseRemoteExecutableCandidate } from "./codex-model-fallback.js"; import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; -import { - resolvePaperclipRunnerTransport, - type PaperclipRunnerTransport, -} from "@paperclipai/adapter-utils/runner-connectivity"; +import { createNativeRunnerExecutionDriver, resolveRemoteRunnerTransportMode } from "./native-runner-execution-driver.js"; import type { Db } from "@paperclipai/db"; import { and, @@ -155,8 +152,6 @@ import { } from "./native-runner-file-handoff.js"; import { nativeRuntimeContractForProvider, nativeToolContractFingerprintForTarget } from "./native-session-resume.js"; import { verifyRetainedMaintenanceNoLaunch } from "./native-maintenance-no-launch.js"; -import { registerRunnerPrpAuthority } from "../../realtime/runner-prp-ws.js"; -import { connectRunnerPrpIngress } from "../../realtime/runner-prp-outbound.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { reportRunFailure } from "../run-failure-report.js"; import { persistActivity, publishActivity } from "../activity-log.js"; @@ -11095,28 +11090,7 @@ export function createRemoteRunnerProcessLauncher(input: { }; } -/** - * Select the remote runner transport before any artifact is staged or provider - * endpoint is acquired. Sandbox ingress is available only to a run already - * authorized by native runtime selection. - */ -export function resolveRemoteRunnerTransportMode(input: { - target: AdapterExecutionTarget; - runnerIngressAuthorized: boolean; -}): "listen_ws" | "dial_wss" { - if (input.target.kind !== "remote") { - throw new Error("runner_transport_ineligible: remote target is required"); - } - const requiredMode = - input.target.transport === "sandbox" && - input.target.effectiveCapabilities?.runnerWebSocketIngress === true - ? "listen_ws" - : "dial_wss"; - if (requiredMode === "listen_ws" && !input.runnerIngressAuthorized) { - throw new Error("runner_ingress_unavailable"); - } - return requiredMode; -} +export { resolveRemoteRunnerTransportMode } from "./native-runner-execution-driver.js"; export function remoteCheckpointIncompleteFailure( settlement: "settled" | "unsettled", @@ -11299,48 +11273,23 @@ async function prepareRemoteDotRunner({ input, target, runner, root, identity }: void handle.completion.then(() => identityFailed(new Error("runner_remote_process_identity_unavailable")), identityFailed); return handle; }; - const controlPlaneRegistration: NonNullable[1]>["dotRunnerOptions"]>["controlPlaneRegistration"]> = async authority => { - let transport: PaperclipRunnerTransport; - // Validate outbound URL before staging; provider ingress needs the staged - // listener binary before requesting the provider's private endpoint. - if (requiredMode === "dial_wss") { - transport = await resolvePaperclipRunnerTransport({ target, runId: input.execution.binding.runId, - localConnectUrl: "ws://127.0.0.1/unused", runnerPublicUrl: input.runnerPublicUrl, - runnerCaBundlePath: input.runnerCaBundlePath, runnerIngressAuthorized: input.runnerIngressAuthorized === true }); - await prepare(); - } else { - await prepare(); - transport = await resolvePaperclipRunnerTransport({ target, runId: input.execution.binding.runId, - localConnectUrl: "ws://127.0.0.1/unused", runnerIngressAuthorized: input.runnerIngressAuthorized === true }); - } - await input.onLog?.("stderr", `[paperclip-runner] Dot controller target=remote transport=${transport.mode}\n`); - if (transport.mode !== "provider_ingress") { - const registration = await registerRunnerPrpAuthority({ companyId: input.execution.binding.companyId, - issueId: input.execution.binding.issueId, agentId: input.execution.binding.agentId, - runId: input.execution.binding.runId, authority }); - let caBundlePath = transport.mode === "direct_outbound" ? transport.caBundlePath : undefined; - if (caBundlePath) { - const remoteCaBundlePath = posix.join(remoteRoot, "bin", "runner-ca-bundle.pem"); - await stageRemoteRunnerFile({ target, runner, sourcePath: caBundlePath, targetPath: remoteCaBundlePath, mode: 0o600 }); - caBundlePath = remoteCaBundlePath; - } - return { ...registration, ready: async () => { await processIdentityReady; }, connection: { mode: "connect" as const, connectUrl: transport.connectUrl, - ...(caBundlePath ? { caBundlePath } : {}) } }; - } - let outbound: ReturnType | null = null; - let activation: Promise | null = null; - return { connection: { mode: "listen" as const, listenAddress: transport.listenAddress, - listenPort: transport.listenPort, listenPath: transport.listenPath }, - activate: () => { activation = (async () => { await processStarted; - outbound = connectRunnerPrpIngress({ authority, endpoint: transport.ingress }); })(); - void activation.catch(() => {}); }, - ready: async () => { await activation; if (!outbound) throw new Error("runner_ingress_unavailable"); await outbound.ready; await processIdentityReady; }, - get failure() { return outbound?.failure; }, - startupFailureCode: "runner_ingress_unavailable" as const, - release: async () => { if (outbound) await outbound.close(); else await transport.ingress.close(); }, - }; - }; - return { runnerBinary, runnerStateDirectory, runnerProcessLauncher, readProviderState, controlPlaneRegistration, adoptExistingRunner }; + const executionDriver = createNativeRunnerExecutionDriver({ + target, binding: input.execution.binding, + runnerIngressAuthorized: input.runnerIngressAuthorized, + runnerPublicUrl: input.runnerPublicUrl, + runnerCaBundlePath: input.runnerCaBundlePath, + prepareRemote: prepare, + processStarted, processIdentityReady, + mapCaBundle: async sourcePath => { + const targetPath = posix.join(remoteRoot, "bin", "runner-ca-bundle.pem"); + await stageRemoteRunnerFile({ target, runner, sourcePath, targetPath, mode: 0o600 }); + return targetPath; + }, + process: { runnerStateDirectory, runnerProcessLauncher, adoptExistingRunner }, + onLog: input.onLog, + trace: input.trace, + }); + return { runnerBinary, readProviderState, ...executionDriver }; } async function createRunnerdBackendWithinSessionClaim( @@ -11488,9 +11437,9 @@ async function createRunnerdBackendWithinSessionClaim( runId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution), turnId: `turn-${input.execution.binding.runId}`, itemId: `item-${input.execution.binding.runId}` }, port: dotRunnerBroker(input.db).port(input.execution), - controlPlaneRegistration: remoteDot?.controlPlaneRegistration ?? (async authority => registerRunnerPrpAuthority({ companyId: input.execution.binding.companyId, - issueId: input.execution.binding.issueId, agentId: input.execution.binding.agentId, - runId: input.execution.binding.runId, authority })), + controlPlaneRegistration: remoteDot?.controlPlaneRegistration ?? createNativeRunnerExecutionDriver({ + target, binding: input.execution.binding, trace: input.trace, onLog: input.onLog, + }).controlPlaneRegistration, adoptExistingRunner: remoteDot?.adoptExistingRunner ?? (recoveredProcess ? { ...recoveredProcess, isAlive: () => verifiedRecoveryProcessIsAlive(recoveredProcess), signal: signal => signalVerifiedRecoveryProcess(recoveredProcess, signal) } : undefined), @@ -12958,25 +12907,12 @@ async function createRunnerdBackendWithinSessionClaim( } : {}), stateDirectory: root, - runnerStateDirectory: remoteStateDirectory, - readRunnerState: - remoteStateDirectory && remoteCommandRunner - ? () => - readRemoteRunnerState({ - runner: remoteCommandRunner, - stateDirectory: remoteStateDirectory, - }) - : undefined, - prepareExternalRunnerState, - archiveExternalRunnerState, runnerBinary: controllerRunnerBinary, codexCommand: remoteCodexBinary ?? undefined, sourceCodexHome: remoteTarget ? resolveSourceCodexHome(resolveNativeProviderEnvironment(input.execution.provider, input.runnerEnvironment)) : undefined, - runnerProcessLauncher: remoteProcessLauncher, runnerReconnectGraceMs: remoteTarget ? 120_000 : undefined, - adoptExistingRunner: adoptedProcess, environment: effectiveRunnerEnvironment, onDiagnostic: (message) => { void input.onLog?.( @@ -13059,253 +12995,47 @@ async function createRunnerdBackendWithinSessionClaim( recoveryPending = false; } : undefined, - controlPlaneRegistration: async (authority, attachmentIdentity) => { - if (retainedTransition && recoveryPending) { - const current = await verifyWarmTransitionRestart(input); - if ( - current.stateFingerprint !== retainedTransition.stateFingerprint - ) { - throw new Error( - "native_runner_warm_transition_recovery_unproven", - ); + ...createNativeRunnerExecutionDriver({ + process: { + runnerStateDirectory: remoteStateDirectory, + readRunnerState: + remoteStateDirectory && remoteCommandRunner + ? () => + readRemoteRunnerState({ + runner: remoteCommandRunner, + stateDirectory: remoteStateDirectory, + }) + : undefined, + prepareExternalRunnerState, + archiveExternalRunnerState, + runnerProcessLauncher: remoteProcessLauncher, + adoptExistingRunner: adoptedProcess, + }, + target, + binding: input.execution.binding, + runnerIngressAuthorized: input.runnerIngressAuthorized, + runnerPublicUrl: input.runnerPublicUrl, + runnerCaBundlePath: input.runnerCaBundlePath, + prepareRemote: prepareRemoteRunner, + processStarted: remoteRunnerProcessSpawned, + checkpoint: checkpointRemoteRunner, + trace: input.trace, + onLog: input.onLog, + mapCaBundle: sourcePath => { + if (!remoteBinary) return sourcePath; + const targetPath = posix.join(posix.dirname(remoteBinary), "runner-ca-bundle.pem"); + remoteCaBundleMapping = { sourcePath, targetPath }; + return targetPath; + }, + beforeConnect: async () => { + if (retainedTransition && recoveryPending) { + const current = await verifyWarmTransitionRestart(input); + if (current.stateFingerprint !== retainedTransition.stateFingerprint) { + throw new Error("native_runner_warm_transition_recovery_unproven"); + } } - } - return measureNativeRunnerSpan( - input.trace, - "runner.transport.connect", - async () => { - if (target.kind === "local") { - const selectedAtMs = Date.now(); - await input.trace?.record({ - name: "runner.transport.selected", - parentName: "runner.transport.connect", - startedAtMs: selectedAtMs, - endedAtMs: selectedAtMs, - attributes: { - mode: "local_loopback", - connectionOwner: "runnerd", - }, - }); - await input.onLog?.( - "stderr", - "[paperclip-runner] transport mode=local_loopback state=connecting\n", - ); - const registration = await measureNativeRunnerSpan( - input.trace, - "runner.prp.route.register", - () => - registerRunnerPrpAuthority({ - companyId: input.execution.binding.companyId, - issueId: input.execution.binding.issueId, - agentId: input.execution.binding.agentId, - runId: - attachmentIdentity?.runId ?? - input.execution.binding.runId, - authority, - }), - ); - return { - ...registration, - startupFailureCode: "runner_local_connect_failed" as const, - }; - } - - const requiredMode = resolveRemoteRunnerTransportMode({ - target, - runnerIngressAuthorized: input.runnerIngressAuthorized === true, - }); - let transport: PaperclipRunnerTransport; - if (requiredMode === "dial_wss") { - // Validate eligibility before staging any artifact. - transport = await measureNativeRunnerSpan( - input.trace, - "runner.transport.resolve", - () => - resolvePaperclipRunnerTransport({ - target, - runId: - attachmentIdentity?.runId ?? - input.execution.binding.runId, - localConnectUrl: "ws://127.0.0.1/unused", - runnerPublicUrl: input.runnerPublicUrl, - runnerCaBundlePath: input.runnerCaBundlePath, - runnerIngressAuthorized: - input.runnerIngressAuthorized === true, - }), - ); - if ( - transport.mode === "direct_outbound" && - transport.caBundlePath && - !existsSync(transport.caBundlePath) - ) { - throw new Error( - "runner_direct_wss_failed: configured runner CA bundle is unavailable", - ); - } - await measureNativeRunnerSpan( - input.trace, - "runner.artifact.prepare", - () => prepareRemoteRunner(requiredMode), - ); - } else { - await measureNativeRunnerSpan( - input.trace, - "runner.artifact.prepare", - () => prepareRemoteRunner(requiredMode), - ); - // Provider endpoint acquisition happens only after runnerd is staged - // and its listener capability has been verified. - transport = await measureNativeRunnerSpan( - input.trace, - "runner.ingress.acquire", - () => - resolvePaperclipRunnerTransport({ - target, - runId: - attachmentIdentity?.runId ?? - input.execution.binding.runId, - localConnectUrl: "ws://127.0.0.1/unused", - runnerPublicUrl: input.runnerPublicUrl, - runnerCaBundlePath: input.runnerCaBundlePath, - runnerIngressAuthorized: true, - }), - ); - } - - const selectedAtMs = Date.now(); - await input.trace?.record({ - name: "runner.transport.selected", - parentName: "runner.transport.connect", - startedAtMs: selectedAtMs, - endedAtMs: selectedAtMs, - attributes: { - mode: transport.mode, - connectionOwner: - transport.mode === "provider_ingress" - ? "paperclip" - : "runnerd", - }, - }); - - await input.onLog?.( - "stderr", - `[paperclip-runner] transport mode=${transport.mode} state=connecting\n`, - ); - - if (transport.mode === "direct_outbound") { - const inbound = await measureNativeRunnerSpan( - input.trace, - "runner.prp.route.register", - () => - registerRunnerPrpAuthority({ - companyId: input.execution.binding.companyId, - issueId: input.execution.binding.issueId, - agentId: input.execution.binding.agentId, - runId: - attachmentIdentity?.runId ?? - input.execution.binding.runId, - authority, - }), - { parentName: "runner.transport.connect" }, - ); - let caBundlePath = transport.caBundlePath; - if (caBundlePath && remoteBinary) { - const remoteCaBundlePath = posix.join( - posix.dirname(remoteBinary), - "runner-ca-bundle.pem", - ); - remoteCaBundleMapping = { - sourcePath: caBundlePath, - targetPath: remoteCaBundlePath, - }; - caBundlePath = remoteCaBundlePath; - } - return { - connection: { - mode: "connect" as const, - connectUrl: transport.connectUrl, - ...(caBundlePath ? { caBundlePath } : {}), - }, - startupFailureCode: "runner_direct_wss_failed" as const, - checkpoint: checkpointRemoteRunner, - release: () => inbound.release(), - }; - } - - if (transport.mode !== "provider_ingress") { - throw new Error("runner_transport_mode_changed_after_dispatch"); - } - let outbound: ReturnType | null = - null; - let activation: Promise | null = null; - return { - connection: { - mode: "listen" as const, - listenAddress: transport.listenAddress, - listenPort: transport.listenPort, - listenPath: transport.listenPath, - }, - activate: () => { - activation = measureNativeRunnerSpan( - input.trace, - "runner.transport.activation", - async () => { - await measureNativeRunnerSpan( - input.trace, - "runner.ingress.wait_for_process", - () => remoteRunnerProcessSpawned, - ); - outbound = connectRunnerPrpIngress({ - authority, - endpoint: transport.ingress, - onStateChange: (state, failureCode) => { - void input.onLog?.( - "stderr", - `[paperclip-runner] transport mode=provider_ingress state=${state}${failureCode ? ` failure=${failureCode}` : ""}\n`, - ); - }, - }); - }, - { parentName: "runner.session.startup" }, - ); - // Cancellation can release the sandbox while activation is - // still waiting for the remote runner process. Attach a - // rejection observer immediately; `ready()` still awaits the - // original promise and reports the same startup failure, but - // an early teardown can no longer crash the controller with - // an unhandled plugin RPC rejection. - void activation.catch(() => undefined); - }, - ready: async () => { - await measureNativeRunnerSpan( - input.trace, - "runner.transport.ready", - async () => { - await activation; - if (!outbound) - throw new Error("runner_ingress_unavailable"); - await measureNativeRunnerSpan( - input.trace, - "runner.prp.authenticate", - () => outbound!.ready, - ); - }, - { parentName: "runner.session.startup" }, - ); - }, - get failure() { - return outbound?.failure; - }, - startupFailureCode: "runner_ingress_unavailable" as const, - checkpoint: checkpointRemoteRunner, - release: async () => { - if (outbound) await outbound.close(); - else await transport.ingress.close(); - }, - }; - }, - ); - }, + }, + }), }).transport, }); const boundManagedSessions = new WeakSet(); diff --git a/server/src/services/native-runtime/setup-readiness.ts b/server/src/services/native-runtime/setup-readiness.ts index b78da1acb1a..dc83185de53 100644 --- a/server/src/services/native-runtime/setup-readiness.ts +++ b/server/src/services/native-runtime/setup-readiness.ts @@ -10,8 +10,7 @@ import { resolvePaperclipRunnerBinary, resolveRemotePaperclipRunnerBinary } from import { readLocalAiCredentialFile } from "../local-ai-credential-file.js"; import { createNativeSshCommandRunner } from "./native-ssh-command-runner.js"; import { resolvePaperclipRunnerTransport } from "@paperclipai/adapter-utils/runner-connectivity"; -import { registerRunnerPrpAuthority } from "../../realtime/runner-prp-ws.js"; -import { connectRunnerPrpIngress } from "../../realtime/runner-prp-outbound.js"; +import { createNativeRunnerExecutionDriver } from "./native-runner-execution-driver.js"; const PROBE_TRANSPORT_ENV_KEYS = ["PATH", "LANG", "LANGUAGE", "TZ", "TMPDIR", "TEMP", "TMP", "SSL_CERT_FILE", "SSL_CERT_DIR", "NODE_EXTRA_CA_CERTS", "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "no_proxy", "all_proxy"]; @@ -124,7 +123,6 @@ async function probeRemoteNativeCodex(options: { return probePreparedNativeRuntime(runtimeDirectory, async () => { const sourceCodexHome = await prepareProbeCodexSource(runtimeDirectory, options.sourceCodexHome, options.sourceAuthJson); let remoteDirectoryCreated = false; - let remoteCaBundlePath: string | undefined; let signalSpawned!: () => void; const spawned = new Promise(resolve => { signalSpawned = resolve; }); return { runtimeDirectory: runtimeDirectory, provider: "codex", model: options.model, reasoningEffort: options.reasoningEffort, @@ -132,49 +130,40 @@ async function probeRemoteNativeCodex(options: { transportOptions: { externallySandboxed: target.transport === "sandbox", runnerBinary: controllerRunnerBinary, codexCommand: remoteCodexBinary, sourceCodexHome, - runnerFilesystemRoot: remoteFilesystemRoot, runnerStateDirectory: remoteStateDirectory, + runnerFilesystemRoot: remoteFilesystemRoot, prpIdentity: { runnerInstanceId, runId: id, environmentLeaseId: `setup-${id}`, normalizedSessionId: `setup-${id}`, turnId: `turn-${id}`, itemId: `item-${id}` }, - readRunnerState: () => readRemoteRunnerState({ runner, stateDirectory: remoteStateDirectory }), - controlPlaneRegistration: async authority => { - const created = await runner.execute({ command: "sh", args: ["-c", 'set -eu; umask 077; mkdir -p -- "$1"; test -d "$1" && test ! -L "$1"; mkdir -- "$2"; install -d -m 0700 "$3" "$4" "$5"', - "paperclip-native-setup", posix.dirname(remoteRoot), remoteRoot, remoteFilesystemRoot, remoteWorkspace, posix.join(remoteFilesystemRoot, "probe-home")], cwd: target.remoteCwd, bypassSession: true, timeoutMs: 10_000 }); - requireSuccess(created, "Native setup could not claim a private runtime directory."); - remoteDirectoryCreated = true; - await artifacts.prepare(ingress ? "listen_ws" : "dial_wss"); - if (outbound?.mode === "direct_outbound" && outbound.caBundlePath) { - remoteCaBundlePath = posix.join(remoteRoot, "bin", "runner-ca-bundle.pem"); - await stageRemoteRunnerFile({ target, runner, sourcePath: outbound.caBundlePath, targetPath: remoteCaBundlePath, mode: 0o600 }); - } - if (outbound?.mode === "direct_outbound") { - const registration = await registerRunnerPrpAuthority({ companyId: options.context.companyId, runId: id, authority }); - return { connection: { mode: "connect" as const, connectUrl: outbound.connectUrl, ...(remoteCaBundlePath ? { caBundlePath: remoteCaBundlePath } : {}) }, - startupFailureCode: "runner_direct_wss_failed" as const, release: registration.release }; - } - const transport = await resolvePaperclipRunnerTransport({ target, runId: id, localConnectUrl: "ws://127.0.0.1/unused", runnerIngressAuthorized: true }); - if (transport.mode !== "provider_ingress") throw new Error("runner_transport_ineligible: expected provider ingress"); - let connection: ReturnType | undefined; - let activation: Promise | undefined; - return { - connection: { mode: "listen" as const, listenAddress: transport.listenAddress, listenPort: transport.listenPort, listenPath: transport.listenPath }, - activate: () => { - activation = spawned.then(() => { connection = connectRunnerPrpIngress({ authority, endpoint: transport.ingress }); }); - void activation.catch(() => undefined); - }, - ready: async () => { await activation; if (!connection) throw new Error("runner_ingress_unavailable"); await connection.ready; }, - get failure() { return connection?.failure; }, - startupFailureCode: "runner_ingress_unavailable" as const, - release: async () => { if (connection) await connection.close(); else await transport.ingress.close(); }, - }; - }, - runnerProcessLauncher: createRemoteRunnerProcessLauncher({ target, runner, remoteBinary, stateDirectory: remoteStateDirectory, - onRunnerProcessSpawned: signalSpawned, processIdentityPath: posix.join(remoteStateDirectory, "runner-process.identity"), diagnosticsDirectory: posix.join(remoteRoot, "diagnostics"), runnerInstanceId, - ensureArtifact: async () => { - for (const name of ["auth.json", "config.toml"] as const) { - const sourcePath = join(runtimeDirectory, "codex-home", name); - try { await stat(sourcePath); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; throw error; } - await stageRemoteRunnerFile({ target, runner, sourcePath, targetPath: posix.join(remoteFilesystemRoot, "codex-home", name), mode: 0o600 }); - } - } }), + ...createNativeRunnerExecutionDriver({ + target, binding: { companyId: options.context.companyId, runId: id }, + runnerIngressAuthorized: true, + runnerPublicUrl: process.env.PAPERCLIP_RUNNER_PUBLIC_URL?.trim() || null, + runnerCaBundlePath: process.env.PAPERCLIP_RUNNER_CA_BUNDLE_PATH?.trim() || null, + processStarted: spawned, + prepareRemote: async mode => { + const created = await runner.execute({ command: "sh", args: ["-c", 'set -eu; umask 077; mkdir -p -- "$1"; test -d "$1" && test ! -L "$1"; mkdir -- "$2"; install -d -m 0700 "$3" "$4" "$5"', + "paperclip-native-setup", posix.dirname(remoteRoot), remoteRoot, remoteFilesystemRoot, remoteWorkspace, posix.join(remoteFilesystemRoot, "probe-home")], cwd: target.remoteCwd, bypassSession: true, timeoutMs: 10_000 }); + requireSuccess(created, "Native setup could not claim a private runtime directory."); + remoteDirectoryCreated = true; + await artifacts.prepare(mode); + }, + mapCaBundle: async sourcePath => { + const targetPath = posix.join(remoteRoot, "bin", "runner-ca-bundle.pem"); + await stageRemoteRunnerFile({ target, runner, sourcePath, targetPath, mode: 0o600 }); + return targetPath; + }, + process: { + runnerStateDirectory: remoteStateDirectory, + readRunnerState: () => readRemoteRunnerState({ runner, stateDirectory: remoteStateDirectory }), + runnerProcessLauncher: createRemoteRunnerProcessLauncher({ target, runner, remoteBinary, stateDirectory: remoteStateDirectory, + onRunnerProcessSpawned: signalSpawned, processIdentityPath: posix.join(remoteStateDirectory, "runner-process.identity"), diagnosticsDirectory: posix.join(remoteRoot, "diagnostics"), runnerInstanceId, + ensureArtifact: async () => { + for (const name of ["auth.json", "config.toml"] as const) { + const sourcePath = join(runtimeDirectory, "codex-home", name); + try { await stat(sourcePath); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; throw error; } + await stageRemoteRunnerFile({ target, runner, sourcePath, targetPath: posix.join(remoteFilesystemRoot, "codex-home", name), mode: 0o600 }); + } + } }), + }, + }), }, ...(options.credentialRefreshAuthPath ? { onCodexCredentialRefresh: async (filename: string) => { if (filename !== posix.join(remoteFilesystemRoot, "codex-home", "auth.json")) throw new Error("The native Codex credential refresh handoff is invalid."); diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts index dd40545d7e5..f947d8edc68 100644 --- a/server/src/vendor/paperclip-runner/index.ts +++ b/server/src/vendor/paperclip-runner/index.ts @@ -61,6 +61,7 @@ export type { PrpVerificationReasonCode, PrpWireConnection, ReplayControlPlaneEventsInput, + RunnerExecutionDriver, RunnerProcessHandle, RunnerProcessLaunchSpec, StrictCompletionContractInput,