Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-10-21 08:22:20 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor(core): share Runner execution lifecycle (#16318)

Browse Source 
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Core starts Runners in local, SSH, and sandbox environments.
> - Coding runs, Dot runs, and setup probes repeat connection setup
code.
> - These paths must keep the same preparation, authentication, and
cleanup order.
> - This pull request gives them one execution lifecycle contract and
one server connection driver.
> - Core keeps the existing PRP control plane after connection.
> - The benefit is one place to maintain environment routing before
adding another execution provider.

## Linked Issues or Issue Description

**What existing behavior does this improve?**

Runner environment preparation and connection setup across local, SSH,
and sandbox execution.

**Current behavior**

The coding transport owns the environment hook type. Dot borrows that
type. The server repeats connection selection, registration, ingress
activation, and release in task execution and setup probes. Changes to
one path can miss another path.

**Proposed behavior**

Use a shared `RunnerExecutionDriver` contract. Route built-in task
execution and remote setup probes through
`createNativeRunnerExecutionDriver`. Keep provider session setup,
process supervision, PRP commands, and recovery logic on their existing
paths.

**Reason and benefit**

Environment code can prepare resources and provide a connection without
adding another PRP execution path. The same driver preserves local
loopback, remote outbound WSS, and authorized sandbox ingress.

**Breaking changes**

None intended. Existing flat SDK options remain compatible. This is an
in-process contract. It does not add a plugin wire contract or Launcher
APIs.

**Additional context**

Related: #15406 proposes plugin-provided Runner execution. This change
starts from `master` and does not depend on that PR. It extracts the
existing lifecycle before adding another provider. Related: #16125
covers remote environment support for external adapters; this PR does
not implement that feature.

## What Changed

- Extract and export `RunnerExecutionDriver` from the coding transport.
Use it for coding and Dot options.
- Share server connection negotiation across coding runs, Dot runs, and
remote setup probes.
- Preserve outbound validation before staging, capability checks before
ingress, process dispatch before ingress connection, and verified
process readiness.
- Preserve warm run registration, restart guards, checkpoint hooks, and
cleanup ownership.
- Add 13 focused driver tests and document the lifecycle boundary.

## Verification

- Targeted server execution and recovery tests: 728 passed, 35 skipped.
These cover `native-session-executor`, `native-ssh-command-runner`, the
new driver, warm restart transitions, and Runner restart recovery.
- Targeted setup and driver tests: 109 passed after the review fix.
Twelve driver tests overlap with the preceding count. The added test
verifies that failed CA staging cannot leave a PRP route behind.
- Runner connectivity tests: 11 passed.
- Runner package lifecycle tests: 287 passed across coding transport,
Dot, restart continuation, and settlement. These exercised the compiled
Rust Runner.
- `pnpm -r typecheck`: passed.
- `pnpm build`: passed.
- `git diff --check`: passed. Local secret-pattern scan: clear.
- The PRP transport runtime and remote process launcher remain
unchanged. A byte comparison against the base confirmed this.
- CI passed on `b10789dfce30a837ce890f44e352064c7a74366f`: all 55 active
checks are green, including the full test shards, Runner checks, browser
tests, typecheck, build, and aggregate gate. Greptile scored this commit
5/5 with no unresolved findings.
- An earlier broad local shard run was not green: 34,888 passed, 12
failed, and 4,125 skipped. Failures were ten timeouts and two Git
fixture assertions affected by local `init.defaultBranch=main`. Isolated
retry passed all six non-route timeout suites (427 tests), two route
suites, and both Git assertions with a test-process-only
`init.defaultBranch=master`. The agent live-run route retry still timed
out. No unrelated source changes were made. Further broad local runs
were stopped at the user's request.

Targeted commands:

```sh
cd server
node ../node_modules/vitest/vitest.mjs run src/services/native-runtime/native-session-executor.test.ts src/services/native-runtime/native-ssh-command-runner.test.ts src/services/native-runtime/native-runner-execution-driver.test.ts src/services/native-runtime/native-warm-transition-restart.integration.test.ts src/services/native-runtime/native-runner-restart-recovery.integration.test.ts --config vitest.config.ts
node ../node_modules/vitest/vitest.mjs run src/services/native-runtime/setup-readiness.test.ts src/services/native-runtime/native-runner-execution-driver.test.ts --config vitest.config.ts
cd ../packages/paperclip-runner
node ../../node_modules/vitest/vitest.mjs run src/live/runnerd-codex-transport.test.ts src/drivers/dot/runnerd-dot-driver.test.ts src/live/runnerd-restart-continuation.test.ts src/live/runnerd-codex-transport-settlement.test.ts --maxWorkers=2
cd ../..
node node_modules/vitest/vitest.mjs run packages/adapter-utils/src/runner-connectivity.test.ts
```

## Risks

- Startup ordering and connection cleanup are sensitive. Focused tests
cover local registration, outbound validation, CA mapping, ingress
ordering, release, and warm authority guards.
- No live SSH host or sandbox provider was used. Existing mocked remote
tests and real local Runner tests passed.
- No database, PRP wire protocol, or Launcher API changes are included.

## Model Used

OpenAI GPT-6 via Codex, with tool use and code execution. The served
model variant and context window were not exposed to this task.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run the stated targeted tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky LeachandPaperclip authored and GitHub committed 2026-10-14 12:58:05 -07:00
1 parent 82884443be
commit c30fb7d2b7
10 files changed
+672 -433
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/paperclip-runner/README.md
+25
View File
@@ -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
packages/paperclip-runner/src/control-plane/runner-execution-driver.ts
+59
View File
@@ -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> | void;
ready?: () => Promise<void>;
failure?: Promise<never>;
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> | void;
release: () => Promise<void> | 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<Record<string, unknown>>;
/** Materializes a verified external checkpoint before authority rotation. */
prepareExternalRunnerState?: () => Promise<void>;
/** Idempotently archives and returns the verified suspended runner binding. */
archiveExternalRunnerState?: (input: {
archiveKey: string;
priorIdentity: DurableRecoveryIdentity;
}) => Promise<Record<string, unknown>>;
/** 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> | boolean;
signal?: (signal: NodeJS.Signals) => Promise<boolean> | boolean;
};
}
packages/paperclip-runner/src/drivers/dot/runnerd-dot-driver.ts
+3 -6
View File
@@ -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<RunnerExecutionDriver, "runnerStateDirectory" | "runnerProcessLauncher" | "controlPlaneRegistration" | "adoptExistingRunner"> {
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<Record<string, unknown> | 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,
packages/paperclip-runner/src/index.ts
+2
View File
@@ -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";
packages/paperclip-runner/src/live/runnerd-codex-transport.ts
+2 -50
View File
@@ -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> | void;
ready?: () => Promise<void>;
failure?: Promise<never>;
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> | void;
release: () => Promise<void> | 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<void>;
/** 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<Record<string, unknown>>;
/** Materializes a verified external checkpoint before authority rotation. */
prepareExternalRunnerState?: () => Promise<void>;
/** Idempotently archives and returns the verified suspended runner binding. */
archiveExternalRunnerState?: (input: {
archiveKey: string;
priorIdentity: DurableRecoveryIdentity;
}) => Promise<Record<string, unknown>>;
/** 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> | boolean;
signal?: (signal: NodeJS.Signals) => Promise<boolean> | boolean;
};
}
export type RunnerdCodexTransportOptions =
server/src/services/native-runtime/native-runner-execution-driver.test.ts
+183
View File
@@ -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<NonNullable<ReturnType<typeof createNativeRunnerExecutionDriver>["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<void>(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<NonNullable<ReturnType<typeof createNativeRunnerExecutionDriver>["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");
});
});
server/src/services/native-runtime/native-runner-execution-driver.ts
+301
View File
@@ -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<RunnerExecutionDriver["controlPlaneRegistration"]>;
type RunnerIdentity = Parameters<ConnectionRegistration>[1];
export interface NativeRunnerExecutionDriverInput {
target: AdapterExecutionTarget;
binding: Omit<Parameters<typeof registerRunnerPrpAuthority>[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<void>;
mapCaBundle?: (sourcePath: string) => string | Promise<string>;
processStarted?: Promise<void>;
processIdentityReady?: Promise<void>;
beforeConnect?: (identity?: RunnerIdentity) => Promise<void>;
checkpoint?: Awaited<ReturnType<ConnectionRegistration>>["checkpoint"];
process?: Omit<RunnerExecutionDriver, "controlPlaneRegistration">;
trace?: NativeRunTrace;
onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
}
async function measureNativeRunnerSpan<T>(
trace: NativeRunTrace | undefined,
name: string,
fn: () => Promise<T>,
options: { parentName?: string } = {},
): Promise<T> {
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<typeof connectRunnerPrpIngress> | null =
null;
let activation: Promise<void> | 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();
},
};
},
);
},
};
}
server/src/services/native-runtime/native-session-executor.ts
+62 -332
View File
@@ -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<NonNullable<NonNullable<Parameters<typeof createNativeSessionBackend>[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<typeof connectRunnerPrpIngress> | null = null;
let activation: Promise<void> | 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<typeof connectRunnerPrpIngress> | null =
null;
let activation: Promise<void> | 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<NativeSession>();
server/src/services/native-runtime/setup-readiness.ts
+34 -45
View File
@@ -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<void>(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<typeof connectRunnerPrpIngress> | undefined;
let activation: Promise<void> | 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.");
server/src/vendor/paperclip-runner/index.ts
Vendored
+1
View File
@@ -61,6 +61,7 @@ export type {
PrpVerificationReasonCode,
PrpWireConnection,
ReplayControlPlaneEventsInput,
RunnerExecutionDriver,
RunnerProcessHandle,
RunnerProcessLaunchSpec,
StrictCompletionContractInput,