/** * PluginWorkerManager — spawns and manages out-of-process plugin worker child * processes, routes JSON-RPC 2.0 calls over stdio, and handles lifecycle * management including crash recovery with exponential backoff. * * Each installed plugin gets one dedicated worker process. The host sends * JSON-RPC requests over the child's stdin and reads responses from stdout. * Worker stderr is captured and forwarded to the host logger. * * Process Model (from PLUGIN_SPEC.md §12): * - One worker process per installed plugin * - Failure isolation: plugin crashes do not affect the host * - Graceful shutdown: 10-second drain, then SIGTERM, then SIGKILL * - Automatic restart with exponential backoff on unexpected exits * * @see PLUGIN_SPEC.md §12 — Process Model * @see PLUGIN_SPEC.md §12.5 — Graceful Shutdown Policy * @see PLUGIN_SPEC.md §13 — Host-Worker Protocol */ import { fork, type ChildProcess } from "node:child_process"; import { randomUUID } from "node:crypto"; import { EventEmitter } from "node:events"; import { createInterface, type Interface as ReadlineInterface } from "node:readline"; import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; import { JSONRPC_VERSION, JSONRPC_ERROR_CODES, PLUGIN_RPC_ERROR_CODES, createRequest, createErrorResponse, parseMessage, serializeMessage, isJsonRpcResponse, isJsonRpcRequest, isJsonRpcNotification, isJsonRpcSuccessResponse, JsonRpcParseError, JsonRpcCallError, SETUP_TOKEN_PTY_OUTPUT_NOTIFICATION, SETUP_TOKEN_PTY_EXIT_NOTIFICATION, DUPLEX_CHANNEL_DATA_NOTIFICATION, DUPLEX_CHANNEL_EXIT_NOTIFICATION, } from "@paperclipai/plugin-sdk"; import type { JsonRpcId, PluginInvocationContext, PluginInvocationScope, JsonRpcResponse, JsonRpcRequest, JsonRpcNotification, WorkerHostCallContext, HostToWorkerMethodName, HostToWorkerMethods, WorkerToHostMethodName, WorkerToHostMethods, InitializeParams, } from "@paperclipai/plugin-sdk"; import { getActiveStepContext } from "@paperclipai/adapter-utils/acpx-engine/startup-timing"; import { CLAUDE_SETUP_TOKEN_COMMAND } from "@paperclipai/adapter-claude-local/server"; import { logger } from "../middleware/logger.js"; import { traceparentFromContextToken } from "../instrumentation.js"; // --------------------------------------------------------------------------- // Constants // --------------------------------------------------------------------------- /** Default timeout for RPC calls in milliseconds. */ const DEFAULT_RPC_TIMEOUT_MS = 30_000; /** * Upper bound for the *default* RPC timeout path (15 minutes). Explicit * caller-supplied timeouts are not subject to this cap: execute-class RPCs such * as `environmentExecute` run entire sandboxed agent sessions in one call and * their callers deliberately request multi-hour budgets (see * `resolvePluginExecuteRpcTimeoutMs` in plugin-environment-driver.ts). * Clamping those explicit budgets here killed long sandboxed runs mid-work. */ const MAX_RPC_TIMEOUT_MS = 15 * 60 * 1_000; /** * Maximum delay accepted by Node timers before Node clamps the timeout to 1ms. * Keep accepted explicit RPC budgets inside this range before calling * setTimeout, otherwise a huge timeout can expire almost immediately. */ const MAX_NODE_TIMER_TIMEOUT_MS = 2_147_483_647; /** Timeout for the initialize RPC call. */ const INITIALIZE_TIMEOUT_MS = 15_000; /** Timeout for the shutdown RPC call before escalating to SIGTERM. */ const SHUTDOWN_DRAIN_MS = 10_000; /** Time to wait after SIGTERM before sending SIGKILL. */ const SIGTERM_GRACE_MS = 5_000; /** Minimum backoff delay for crash recovery (1 second). */ const MIN_BACKOFF_MS = 1_000; /** Maximum backoff delay for crash recovery (5 minutes). */ const MAX_BACKOFF_MS = 5 * 60 * 1_000; /** Backoff multiplier on each consecutive crash. */ const BACKOFF_MULTIPLIER = 2; /** Maximum number of consecutive crashes before giving up on auto-restart. */ const MAX_CONSECUTIVE_CRASHES = 10; /** Time window in which crashes are considered consecutive (10 minutes). */ const CRASH_WINDOW_MS = 10 * 60 * 1_000; /** Maximum number of stderr characters retained for worker failure context. */ const MAX_STDERR_EXCERPT_CHARS = 8_000; /** Maximum characters accepted for one `execute.log` chunk. A larger chunk is * dropped, so a faulty or hostile worker cannot flood the host with one * unbounded notification. */ const MAX_EXECUTE_LOG_CHUNK_CHARS = 1_000_000; /** * Maximum characters accepted for one incoming worker stdout line before the * host parses it as JSON. The host drops a longer line without a parse, so a * faulty or hostile worker cannot force the host to parse an unbounded document * and exhaust memory. The bound sits far above the largest legitimate framed * message, so a real large command result still passes. A worker can override * it through `WorkerStartOptions.executeLogLimits`. */ const MAX_WORKER_MESSAGE_CHARS = 128 * 1024 * 1024; /** * Default ceiling for the total characters one execute call may stream through * `execute.log`. The host counts the delivered characters for each active * execute route and drops further chunks past this bound, so one runaway or * hostile execution cannot flood the host and the run-log sink without limit. * The final command result still delivers the complete output through its own * capture path. A worker can override it through * `WorkerStartOptions.executeLogLimits`. */ const MAX_EXECUTE_LOG_TOTAL_CHARS = 128 * 1024 * 1024; /** Maximum characters for one live login pseudo-terminal output notification. */ const MAX_SETUP_TOKEN_PTY_CHUNK_CHARS = 1_000_000; /** Maximum cumulative output characters for one login pseudo-terminal route. */ const MAX_SETUP_TOKEN_PTY_TOTAL_CHARS = 8 * 1024 * 1024; /** The default open timeout for one login pseudo-terminal route, in milliseconds. */ const SETUP_TOKEN_PTY_OPEN_TIMEOUT_MS = 30_000; /** The default close timeout for one login pseudo-terminal route, in milliseconds. */ const SETUP_TOKEN_PTY_CLOSE_TIMEOUT_MS = 10_000; /** * The fixed non-secret error a disallowed login command returns. The manager * forwards only the compile-time `CLAUDE_SETUP_TOKEN_COMMAND` to the worker * pseudo-terminal. It rejects any other command before the worker call, so a * future caller cannot spawn an arbitrary process in the sandbox. */ const SETUP_TOKEN_PTY_COMMAND_NOT_ALLOWED = "SETUP_TOKEN_PTY_COMMAND_NOT_ALLOWED"; /** The fixed non-secret error a rejected second credential open returns. */ const SETUP_TOKEN_PTY_ROUTE_BUSY = "SETUP_TOKEN_PTY_ROUTE_BUSY"; /** The fixed non-secret error a failed open returns. */ const SETUP_TOKEN_PTY_OPEN_FAILED = "SETUP_TOKEN_PTY_OPEN_FAILED"; // Bounds and timeouts for the generic duplex channel route. The route mirrors the // login pseudo-terminal route, but it carries no command allowlist and adds seven // explicit bounds the pseudo-terminal route lacks. Each bound ends the route when // it passes the limit, so a faulty or hostile worker cannot flood the host. /** The default maximum characters for one duplex channel data notification. */ const MAX_DUPLEX_CHANNEL_CHUNK_CHARS = 1_000_000; /** * The default maximum cumulative characters the host buffers for one duplex * channel route before a data listener attaches. A worker that streams data * before the consumer binds cannot grow the host buffer without limit. */ const MAX_DUPLEX_CHANNEL_PRE_BIND_CHARS = 8 * 1024 * 1024; /** * The default maximum number of data frames the host buffers for one duplex * channel route before a data listener attaches. */ const MAX_DUPLEX_CHANNEL_PRE_BIND_FRAMES = 10_000; /** * The margin the pre-bind hold ceiling keeps above the pre-bind buffered-frame * bound (`maxDuplexChannelPreBindFrames`). A worker can batch data and exit * frames with the open reply, so the host reads them before the route binds and * before it can apply the per-frame bounds. The host holds these frames and * replays them after the bind. The hold ceiling bounds that hold, so a worker * that floods frames before the open reply cannot make the host hold an * unbounded number of frames. * * The hold ceiling derives from the buffered bound, not a fixed constant. A * fixed ceiling at or below a caller-configured buffered bound would drop the * frame that must instead trip the buffered bound during replay, so the route * would never end. One frame of margin is enough. It lets the frame that * exceeds the buffered bound reach the hold, so the replay's buffered-bound * check, not the hold, ends the route. */ const DUPLEX_CHANNEL_PRE_BIND_HOLD_MARGIN_FRAMES = 1; /** * The default maximum number of in-flight host→worker requests for one duplex * channel route. A worker that never replies cannot make the host hold an * unbounded number of pending requests. */ const MAX_DUPLEX_CHANNEL_PENDING_REQUESTS = 256; /** The default maximum characters for one host→worker duplex channel write. */ const MAX_DUPLEX_CHANNEL_WRITE_CHARS = 1_000_000; /** * The default maximum number of protocol errors for one duplex channel route. * A protocol error is one malformed or mismatched data frame. The route ends * when the count passes this budget, so a flood of bad frames bounds the route. */ const MAX_DUPLEX_CHANNEL_PROTOCOL_ERRORS = 100; /** * The default maximum cumulative bytes the host forwards for one duplex channel * route over its whole life. The host counts the bytes of every inbound chunk, * before and after a data listener attaches. The route ends when the count * passes this cap, so an active route with a bound listener cannot stream an * unbounded number of bytes. */ const MAX_DUPLEX_CHANNEL_TOTAL_DATA_BYTES = 256 * 1024 * 1024; /** * The default maximum lifetime for one duplex channel route, in milliseconds. * The host starts a timer when the route opens and ends the route when the * timer expires, so a route cannot live without limit. */ const MAX_DUPLEX_CHANNEL_DURATION_MS = 60 * 60 * 1000; /** The default open timeout for one duplex channel route, in milliseconds. */ const DUPLEX_CHANNEL_OPEN_TIMEOUT_MS = 30_000; /** The default close timeout for one duplex channel route, in milliseconds. */ const DUPLEX_CHANNEL_CLOSE_TIMEOUT_MS = 10_000; /** The fixed non-secret error a rejected second duplex channel open returns. */ const DUPLEX_CHANNEL_ROUTE_BUSY = "DUPLEX_CHANNEL_ROUTE_BUSY"; /** The fixed non-secret error a failed duplex channel open returns. */ const DUPLEX_CHANNEL_OPEN_FAILED = "DUPLEX_CHANNEL_OPEN_FAILED"; // The process-scoped monotonic route-generation source. The host mints one // strictly increasing, non-reusable `hostRouteId` on each duplex channel open, // across every worker in the process. A retired generation never returns, so a // late frame for a closed pair never collides with a new open. The host owns the // value; no worker field sets it. let duplexHostRouteIdSequence = 0; /** Mint the next monotonic, non-reusable host route identifier for a duplex channel open. */ function nextDuplexHostRouteId(): string { duplexHostRouteIdSequence += 1; return `duplex-route-${duplexHostRouteIdSequence}`; } // The separator between the two identifiers of a duplex route pair key. It is a // null byte, which appears in neither a host route id nor a worker session id, so // two distinct pairs never collapse to one key. const DUPLEX_PAIR_KEY_SEPARATOR = "\u0000"; /** Build the exact-pair routing key from the host route id and the worker session id. */ function duplexPairKey(hostRouteId: string, workerSessionId: string): string { return `${hostRouteId}${DUPLEX_PAIR_KEY_SEPARATOR}${workerSessionId}`; } // The maximum number of tombstoned duplex pairs one worker retains. The worker // keeps every tombstone until it retires, so a closed pair never returns. Before // the set would exceed this bound, the host retires the worker, which drops every // route and every tombstone at once. A tombstone overflow fails closed; the host // never evicts a tombstone and lets a pair return. const MAX_DUPLEX_ROUTE_TOMBSTONES = 4096; /** Minimum time between two dropped-`execute.log` debug records. The router * rate-limits the record so a flood of dropped chunks writes at most one line * per window with a running count. */ const EXECUTE_LOG_DROP_LOG_INTERVAL_MS = 1_000; // --------------------------------------------------------------------------- // Types // --------------------------------------------------------------------------- /** * Status of a managed worker process. */ export type WorkerStatus = | "stopped" | "starting" | "running" | "stopping" | "crashed" | "backoff"; /** * Worker-to-host method handler. The host registers these to service calls * that the plugin worker makes back to the host (e.g. state.get, events.emit). */ export type WorkerToHostHandler = ( params: WorkerToHostMethods[M][0], context?: WorkerHostCallContext, ) => Promise; /** * A map of all worker-to-host method handlers provided by the host. */ export type WorkerToHostHandlers = { [M in WorkerToHostMethodName]?: WorkerToHostHandler; }; /** * Events emitted by a PluginWorkerHandle. */ export interface WorkerHandleEvents { /** Worker process started and is ready (initialize succeeded). */ "ready": { pluginId: string }; /** Worker process exited. */ "exit": { pluginId: string; code: number | null; signal: NodeJS.Signals | null }; /** Worker process crashed unexpectedly. */ "crash": { pluginId: string; code: number | null; signal: NodeJS.Signals | null; willRestart: boolean }; /** Worker process errored (e.g. spawn failure). */ "error": { pluginId: string; error: Error }; /** Worker status changed. */ "status": { pluginId: string; status: WorkerStatus; previousStatus: WorkerStatus }; } type WorkerHandleEventName = keyof WorkerHandleEvents; export function appendStderrExcerpt(current: string, chunk: string): string { const next = current ? `${current}\n${chunk}` : chunk; return next.length <= MAX_STDERR_EXCERPT_CHARS ? next : next.slice(-MAX_STDERR_EXCERPT_CHARS); } export function formatWorkerFailureMessage(message: string, stderrExcerpt: string): string { const excerpt = stderrExcerpt.trim(); if (!excerpt) return message; if (message.includes(excerpt)) return message; return `${message}\n\nWorker stderr:\n${excerpt}`; } /** * Resolve the effective timeout for an RPC call. * * An explicit, positive, finite caller-supplied timeout bypasses the 15-minute * RPC cap after normalization to Node's timer-safe integer range. Callers that * pass one (e.g. the environment driver for `environmentExecute`) own their * budget, and independent inactivity/safety guards bound hung runs. Only the * default path (no usable explicit timeout) is clamped to MAX_RPC_TIMEOUT_MS so * ordinary plugin calls stay bounded. */ export function resolveRpcCallTimeoutMs( explicitTimeoutMs: number | undefined, defaultTimeoutMs: number, ): number { if ( explicitTimeoutMs !== undefined && Number.isFinite(explicitTimeoutMs) && explicitTimeoutMs > 0 ) { return Math.min(Math.max(Math.trunc(explicitTimeoutMs), 1), MAX_NODE_TIMER_TIMEOUT_MS); } return Math.min(defaultTimeoutMs, MAX_RPC_TIMEOUT_MS); } /** * Options for starting a worker process. */ /** * The process-scoped aggregate route-slot controller. The manager injects one * shared instance into every worker handle, so the ceiling counts every * concurrent duplex route across the process, not one agent's setting. */ export interface DuplexRouteSlotController { /** Reserve one route slot. Return true when a slot was free and is now held. */ tryAcquire(): boolean; /** Release one held route slot. */ release(): void; } export interface WorkerStartOptions { /** Absolute path to the plugin worker entrypoint (CJS bundle). */ entrypointPath: string; /** Plugin manifest. */ manifest: PaperclipPluginManifestV1; /** Resolved plugin configuration. */ config: Record; /** Host instance information for the initialize call. */ instanceInfo: { instanceId: string; hostVersion: string; }; /** Host API version. */ apiVersion: number; /** Host-derived plugin database namespace, when declared. */ databaseNamespace?: string | null; /** Handlers for worker→host RPC calls. */ hostHandlers: WorkerToHostHandlers; /** Default timeout for RPC calls (ms). Defaults to 30s. */ rpcTimeoutMs?: number; /** Whether to auto-restart on crash. Defaults to true. */ autoRestart?: boolean; /** Node.js execArgv passed to the child process. */ execArgv?: string[]; /** Environment variables passed to the child process. */ env?: Record; /** * The process-scoped aggregate route-slot controller for the duplex channel * ceiling. The manager injects one shared instance into every worker handle. * When it is absent, the worker admits every duplex open unbounded (a unit test * constructs a handle this way). */ duplexRouteSlots?: DuplexRouteSlotController | null; /** * Companies this worker may act on from proactive (no-invocation) worker→host * calls — the plugin's configured companies. Seeded onto the handle at * creation, BEFORE the child process spawns, so a proactive plugin that * issues host calls during setup() (e.g. the chat gateway's one-shot * `events.subscribe`, which runs while `startWorker` is still awaiting the * initialize response) is already authorized when those calls arrive. The set * can still be replaced at runtime via `setProactiveCompanyScopes` (e.g. on a * config change). Never widens access beyond the listed companies (LOOA-695). */ proactiveCompanyScopes?: readonly string[]; /** * Callback for stream notifications from the worker (streams.open/emit/close). * The host wires this to the PluginStreamBus to fan out events to SSE clients. */ onStreamNotification?: (method: string, params: Record) => void; /** * Framing and flood limits for the `execute.log` route. The defaults bound * one incoming line before the JSON parse and the total streamed output for * one execute call. A test overrides them to exercise the drop paths without * huge inputs. */ executeLogLimits?: { /** Max characters for one incoming worker line before the JSON parse. */ maxIncomingMessageChars?: number; /** Max total characters one execute call may stream through `execute.log`. */ maxTotalCharsPerExecute?: number; }; /** * Bounds and timeouts for the login pseudo-terminal route. The * defaults bound one output notification, the cumulative output per route, and * the open and the close timeouts. A test overrides them to exercise the * terminalize paths without huge inputs or long waits. */ setupTokenPtyLimits?: { /** Max characters for one login pseudo-terminal output notification. */ maxChunkChars?: number; /** Max cumulative output characters for one login pseudo-terminal route. */ maxTotalChars?: number; /** The open timeout for one login pseudo-terminal route, in milliseconds. */ openTimeoutMs?: number; /** The close timeout for one login pseudo-terminal route, in milliseconds. */ closeTimeoutMs?: number; }; /** * Bounds and timeouts for the generic duplex channel route. The defaults bound * one data notification, the pre-bind buffer, the in-flight request count, one * host→worker write, the protocol-error budget, and the open and close * timeouts. A test overrides them to exercise each bound without huge inputs or * long waits. */ duplexChannelLimits?: { /** Max characters for one duplex channel data notification. */ maxChunkChars?: number; /** Max cumulative characters the host buffers before a data listener attaches. */ maxPreBindBufferedChars?: number; /** Max number of data frames the host buffers before a data listener attaches. */ maxPreBindBufferedFrames?: number; /** Max number of in-flight host→worker requests for one route. */ maxPendingRequests?: number; /** Max characters for one host→worker duplex channel write. */ maxWriteChars?: number; /** Max number of protocol errors for one route before the route ends. */ maxProtocolErrors?: number; /** Max cumulative bytes the host forwards for one route over its whole life. */ maxTotalDataBytes?: number; /** The maximum lifetime for one route, in milliseconds. */ maxDurationMs?: number; /** The open timeout for one duplex channel route, in milliseconds. */ openTimeoutMs?: number; /** The close timeout for one duplex channel route, in milliseconds. */ closeTimeoutMs?: number; }; } /** * A pending RPC call waiting for a response from the worker. */ interface PendingRequest { /** The request ID. */ id: JsonRpcId; /** Method name (for logging). */ method: string; /** Resolve the promise with the response. */ resolve: (response: JsonRpcResponse) => void; /** Timeout timer handle. */ timer: ReturnType; /** Timestamp when the request was sent. */ sentAt: number; /** Active host-owned invocation id attached to this host→worker call. */ invocationId?: string; } interface ActiveInvocation { scope: PluginInvocationScope; timer?: ReturnType; // The host-minted W3C `traceparent` for the active startup span, or undefined // when no startup span is active. The span host handler reads it to mint the // parentage, so a worker never supplies the parent itself. traceparent?: string; } /** * Sink for one incremental output chunk of an active `environmentExecute` call. * The host runner passes it to `call` for the execute method, and the manager * delivers each `execute.log` chunk to it. The sink may return a promise; the * caller owns the ordering. */ export type ExecuteLogSink = ( stream: "stdout" | "stderr", chunk: string, ) => void | Promise; /** * The input the manager needs to open one live login pseudo-terminal route * The manager mints the host route identifier; the caller supplies * only the sandbox scope, the provider lease id, and the fixed command. */ export interface SetupTokenPtyOpenInput { driverKey: string; companyId: string; environmentId: string; providerLeaseId: string; command: string; } /** * One live login pseudo-terminal session the manager hands to the login * transport. The shape matches the sandbox provider setup-token * pseudo-terminal session, so the transport consumes it with no adapter. */ export interface SetupTokenPtyHostSession { /** Registers the one output listener. The session streams each raw chunk in order. */ onData(listener: (chunk: string) => void): void; /** Writes raw input bytes to the pseudo-terminal. */ write(data: string): void; /** Resolves with the child exit code when the command ends or the route terminalizes. */ wait(): Promise<{ exitCode: number | null }>; /** Stops the child process. Safe to call more than one time. */ kill(): void; /** Closes the route and releases the terminal. Safe to call more than one time. */ close(): Promise; } /** * The input the manager needs to open one generic duplex channel. The manager * mints the host route identifier. The caller supplies the sandbox scope, the * provider lease id, and the command. The duplex channel carries no command * allowlist, so the caller owns the command. */ export interface DuplexChannelOpenInput { driverKey: string; companyId: string; environmentId: string; providerLeaseId: string; /** * The command argument vector the worker runs on the channel. Element 0 is the * program. The worker runs the vector with no shell, so a shell metacharacter * in an element cannot inject a command. */ command: readonly string[]; } /** * One live duplex channel the manager hands to a caller. The shape matches the * login pseudo-terminal session, so a caller consumes one live bidirectional * stream with the same methods. */ export interface DuplexChannelHostSession { /** Registers the one data listener. The session streams each raw chunk in order. */ onData(listener: (chunk: string) => void): void; /** Writes raw input bytes to the channel. */ write(data: string): void; /** Resolves with the child exit code when the command ends or the route ends. */ wait(): Promise<{ exitCode: number | null }>; /** Stops the child process. Safe to call more than one time. */ kill(): void; /** Closes the route and releases the channel. Safe to call more than one time. */ close(): Promise; } /** * Host-owned route for one active execute call. The host mints the invocation * id and stores the exact company id and log sink here. A worker never selects * this record; the host looks it up by the host-issued invocation id on the * message envelope. The company id is the single authority for the delivery * target, so an `execute.log` notification never carries a company id. */ interface ExecuteLogRoute { companyId: string; onLog: ExecuteLogSink; /** * The count of characters delivered through this route. The router bounds the * per-execute total and drops chunks past the configured ceiling. */ deliveredChars: number; /** * Latched when the router cannot bind the shared worker pipe to a single * company, because a second company's execute overlapped this one. After the * latch the router drops every further chunk for this route and lets the final * command result deliver the complete output. The latch keeps the delivered * prefix contiguous, so the run log never shows a gap. */ crossCompanyBlocked: boolean; } // --------------------------------------------------------------------------- // PluginWorkerHandle — manages a single worker process // --------------------------------------------------------------------------- /** * Handle for a single plugin worker process. * * Callers use `start()` to spawn the worker, `call()` to send RPC requests, * and `stop()` to gracefully shut down. The handle manages crash recovery * with exponential backoff automatically when `autoRestart` is enabled. */ export interface PluginWorkerHandle { /** The plugin ID this worker serves. */ readonly pluginId: string; /** Current worker status. */ readonly status: WorkerStatus; /** Start the worker process. Resolves when initialize completes. */ start(): Promise; /** * Stop the worker process gracefully. * * Sends a `shutdown` RPC call, waits up to 10 seconds for the worker to * exit, then escalates to SIGTERM, and finally SIGKILL if needed. */ stop(): Promise; /** * Restart the worker process (stop + start). */ restart(): Promise; /** * Send a typed host→worker RPC call. * * @param method - The RPC method name * @param params - Method parameters * @param timeoutMs - Optional per-call timeout override * @returns The method result * @throws {JsonRpcCallError} if the worker returns an error response * @throws {Error} if the worker is not running or the call times out */ call( method: M, params: HostToWorkerMethods[M][0], timeoutMs?: number, executeLogSink?: ExecuteLogSink, ): Promise; /** * Send a fire-and-forget notification to the worker (no response expected). */ notify(method: string, params: unknown): void; /** * Open one live login pseudo-terminal route on this worker. The * manager mints the host route identifier, reserves the route, drives the open, * binds the worker session identifier one time, and returns a session the login * transport drives. It permits one active credential pseudo-terminal per worker. */ openSetupTokenPtySession( input: SetupTokenPtyOpenInput, ): Promise; /** * Open one generic duplex channel on this worker. The manager mints the host * route identifier, reserves the route, drives the open, binds the worker * session identifier one time, and returns a session a caller drives. It * permits one active duplex channel per worker. It enforces five explicit * bounds and ends the route when a bound passes its limit. */ openDuplexChannel( input: DuplexChannelOpenInput, ): Promise; /** * Authorize the set of companies this worker may act on from proactive * (non-invocation) context. Replaces any previously-authorized set. See the * proactive-company-scope note in `createPluginWorkerHandle` for rationale. */ setProactiveCompanyScopes(companyIds: readonly string[]): void; /** Subscribe to worker events. */ on( event: K, listener: (payload: WorkerHandleEvents[K]) => void, ): void; /** Unsubscribe from worker events. */ off( event: K, listener: (payload: WorkerHandleEvents[K]) => void, ): void; /** Optional methods the worker reported during initialization. */ readonly supportedMethods: string[]; /** Get diagnostic info about the worker. */ diagnostics(): WorkerDiagnostics; } /** * Diagnostic information about a worker process. */ export interface WorkerDiagnostics { pluginId: string; status: WorkerStatus; pid: number | null; uptime: number | null; consecutiveCrashes: number; totalCrashes: number; pendingRequests: number; lastCrashAt: number | null; nextRestartAt: number | null; } // --------------------------------------------------------------------------- // PluginWorkerManager — manages all plugin workers // --------------------------------------------------------------------------- /** * The top-level manager that holds all plugin worker handles. * * Provides a registry of workers keyed by plugin ID, with convenience methods * for starting/stopping all workers and routing RPC calls. */ export interface PluginWorkerManager { /** * Register and start a worker for a plugin. * * @returns The worker handle * @throws if a worker is already registered for this plugin */ startWorker(pluginId: string, options: WorkerStartOptions): Promise; /** * Stop and unregister a specific plugin worker. */ stopWorker(pluginId: string): Promise; /** * Get the worker handle for a plugin. */ getWorker(pluginId: string): PluginWorkerHandle | undefined; /** * Check if a worker is registered and running for a plugin. */ isRunning(pluginId: string): boolean; /** * Authorize the companies a plugin's worker may act on from proactive * (non-invocation) context. No-op if the worker is not registered. */ setProactiveCompanyScopes(pluginId: string, companyIds: readonly string[]): void; /** * Stop all managed workers. Called during server shutdown. */ stopAll(): Promise; /** * Get diagnostic info for all workers. */ diagnostics(): WorkerDiagnostics[]; /** * Send an RPC call to a specific plugin worker. * * @throws if the worker is not running */ call( pluginId: string, method: M, params: HostToWorkerMethods[M][0], timeoutMs?: number, executeLogSink?: ExecuteLogSink, ): Promise; /** * Open one live login pseudo-terminal route on a specific plugin worker * See {@link PluginWorkerHandle.openSetupTokenPtySession}. * * @throws if the worker is not registered. */ openSetupTokenPtySession( pluginId: string, input: SetupTokenPtyOpenInput, ): Promise; } // --------------------------------------------------------------------------- // Implementation: createPluginWorkerHandle // --------------------------------------------------------------------------- /** * Create a handle for a single plugin worker process. * * @internal Exported for testing; consumers should use `createPluginWorkerManager`. */ export function createPluginWorkerHandle( pluginId: string, options: WorkerStartOptions, ): PluginWorkerHandle { const log = logger.child({ service: "plugin-worker", pluginId }); const emitter = new EventEmitter(); /** * Higher than default (10) to accommodate multiple subscribers to * crash/ready/exit events during integration tests and runtime monitoring. */ emitter.setMaxListeners(50); // Worker process state let childProcess: ChildProcess | null = null; let readline: ReadlineInterface | null = null; let stderrReadline: ReadlineInterface | null = null; let status: WorkerStatus = "stopped"; let startedAt: number | null = null; let stderrExcerpt = ""; // Pending RPC requests awaiting a response const pendingRequests = new Map(); let nextRequestId = 1; const activeInvocations = new Map(); // Host-owned execute routes, keyed by the host-issued invocation id. Only an // `environmentExecute` call with a log sink registers a route here. The // `execute.log` router delivers only through this map — never through the // generic `activeInvocations` record — so a non-execute call can never become // a log target. const activeExecuteRoutes = new Map(); // Rate-limit state for dropped `execute.log` notifications. The debug record // never carries chunk bytes. let executeLogDropCount = 0; let executeLogDropLoggedAtMs = 0; // Rate-limit state for dropped oversized worker lines. The warn record carries // only the length, never the line bytes. let oversizedLineDropCount = 0; let oversizedLineLoggedAtMs = 0; // Framing and flood limits for the `execute.log` route. The defaults bound one // incoming line before the JSON parse and the total streamed output for one // execute call. A caller (a test) can lower them. const maxIncomingMessageChars = options.executeLogLimits?.maxIncomingMessageChars ?? MAX_WORKER_MESSAGE_CHARS; const maxExecuteLogTotalChars = options.executeLogLimits?.maxTotalCharsPerExecute ?? MAX_EXECUTE_LOG_TOTAL_CHARS; // Bounds and timeouts for the login pseudo-terminal route. A caller // (a test) can lower them to exercise the terminalize paths. const maxSetupTokenPtyChunkChars = options.setupTokenPtyLimits?.maxChunkChars ?? MAX_SETUP_TOKEN_PTY_CHUNK_CHARS; const maxSetupTokenPtyTotalChars = options.setupTokenPtyLimits?.maxTotalChars ?? MAX_SETUP_TOKEN_PTY_TOTAL_CHARS; const setupTokenPtyOpenTimeoutMs = options.setupTokenPtyLimits?.openTimeoutMs ?? SETUP_TOKEN_PTY_OPEN_TIMEOUT_MS; const setupTokenPtyCloseTimeoutMs = options.setupTokenPtyLimits?.closeTimeoutMs ?? SETUP_TOKEN_PTY_CLOSE_TIMEOUT_MS; // Bounds and timeouts for the generic duplex channel route. A caller (a test) // can lower them to exercise each bound and the terminalize paths. const maxDuplexChannelChunkChars = options.duplexChannelLimits?.maxChunkChars ?? MAX_DUPLEX_CHANNEL_CHUNK_CHARS; const maxDuplexChannelPreBindChars = options.duplexChannelLimits?.maxPreBindBufferedChars ?? MAX_DUPLEX_CHANNEL_PRE_BIND_CHARS; const maxDuplexChannelPreBindFrames = options.duplexChannelLimits?.maxPreBindBufferedFrames ?? MAX_DUPLEX_CHANNEL_PRE_BIND_FRAMES; // The pre-bind hold ceiling stays one frame above the buffered bound, so the // replay, not the hold, ends the route on the buffered bound. See // DUPLEX_CHANNEL_PRE_BIND_HOLD_MARGIN_FRAMES for why the margin must hold. const maxDuplexChannelPreBindHoldFrames = maxDuplexChannelPreBindFrames + DUPLEX_CHANNEL_PRE_BIND_HOLD_MARGIN_FRAMES; const maxDuplexChannelPendingRequests = options.duplexChannelLimits?.maxPendingRequests ?? MAX_DUPLEX_CHANNEL_PENDING_REQUESTS; const maxDuplexChannelWriteChars = options.duplexChannelLimits?.maxWriteChars ?? MAX_DUPLEX_CHANNEL_WRITE_CHARS; const maxDuplexChannelProtocolErrors = options.duplexChannelLimits?.maxProtocolErrors ?? MAX_DUPLEX_CHANNEL_PROTOCOL_ERRORS; const maxDuplexChannelTotalDataBytes = options.duplexChannelLimits?.maxTotalDataBytes ?? MAX_DUPLEX_CHANNEL_TOTAL_DATA_BYTES; const maxDuplexChannelDurationMs = options.duplexChannelLimits?.maxDurationMs ?? MAX_DUPLEX_CHANNEL_DURATION_MS; const duplexChannelOpenTimeoutMs = options.duplexChannelLimits?.openTimeoutMs ?? DUPLEX_CHANNEL_OPEN_TIMEOUT_MS; const duplexChannelCloseTimeoutMs = options.duplexChannelLimits?.closeTimeoutMs ?? DUPLEX_CHANNEL_CLOSE_TIMEOUT_MS; // ------------------------------------------------------------------ // Proactive company scopes (LOOA-629) // ------------------------------------------------------------------ // A proactive plugin (e.g. the chat gateway) does company-scoped work from // its own timers/loops — not inside a host-issued top-level invocation // (onEvent/performAction/executeTool/configChanged). Those worker→host calls // carry no `paperclipInvocationId`, so the governed-access gate // (host-client-factory.ts) rejects any company-scoped request with // "company context is required" (regression class from #9557). The host // authorizes a bounded set of companies — the plugin's configured companies, // set by the loader after startup config delivery — for such proactive work. // A no-invocation call that references one of these companies resolves to // that company's scope; a call referencing any other company stays denied, // and in-invocation calls keep their strict single-company match. // // Seeded from options at handle creation — before the child process is // spawned — so a proactive plugin's setup()-time host calls (which land while // `startWorker` is still awaiting initialize) are authorized in time. The // loader used to call setProactiveCompanyScopes only AFTER startWorker // resolved, which was too late for the gateway's one-shot events.subscribe // and left outbound push permanently dead (LOOA-695). const proactiveCompanyScopes = new Set(); for (const id of options.proactiveCompanyScopes ?? []) { const trimmed = readNonEmptyString(id); if (trimmed) proactiveCompanyScopes.add(trimmed); } // Optional methods reported by the worker during initialization let supportedMethods: string[] = []; // Crash tracking for exponential backoff let consecutiveCrashes = 0; let totalCrashes = 0; let lastCrashAt: number | null = null; let backoffTimer: ReturnType | null = null; let nextRestartAt: number | null = null; // Track open stream channels so we can emit synthetic close on crash. // Maps channel → companyId. const openStreamChannels = new Map(); // Shutdown coordination let intentionalStop = false; const rpcTimeoutMs = options.rpcTimeoutMs ?? DEFAULT_RPC_TIMEOUT_MS; const autoRestart = options.autoRestart ?? true; // ----------------------------------------------------------------------- // Status management // ----------------------------------------------------------------------- function setStatus(newStatus: WorkerStatus): void { const prev = status; if (prev === newStatus) return; status = newStatus; log.debug({ from: prev, to: newStatus }, "worker status change"); emitter.emit("status", { pluginId, status: newStatus, previousStatus: prev }); } // ----------------------------------------------------------------------- // JSON-RPC message sending // ----------------------------------------------------------------------- function sendMessage(message: unknown): void { if (!childProcess?.stdin?.writable) { throw new Error(`Worker process for plugin "${pluginId}" is not writable`); } const serialized = serializeMessage(message as any); childProcess.stdin.write(serialized); } function errorCodeForWorkerHostError(err: unknown): number { const code = (err as { code?: unknown } | null)?.code; const pluginErrorCodes: readonly number[] = Object.values(PLUGIN_RPC_ERROR_CODES); return typeof code === "number" && pluginErrorCodes.includes(code) ? code : JSONRPC_ERROR_CODES.INTERNAL_ERROR; } // ----------------------------------------------------------------------- // Incoming message handling // ----------------------------------------------------------------------- function handleLine(line: string): void { if (!line.trim()) return; // Enforce the framing bound BEFORE the JSON parse. A line longer than the // limit is dropped without a parse, so a faulty or hostile worker cannot // force the host to parse an unbounded document and exhaust memory. if (line.length > maxIncomingMessageChars) { dropOversizedLine(line.length); return; } let message: unknown; try { message = parseMessage(line); } catch (err) { if (err instanceof JsonRpcParseError) { log.warn({ rawLine: line.slice(0, 200) }, "unparseable message from worker"); } else { log.warn({ err }, "error parsing worker message"); } return; } if (isJsonRpcResponse(message)) { handleResponse(message); } else if (isJsonRpcRequest(message)) { handleWorkerRequest(message as JsonRpcRequest); } else if (isJsonRpcNotification(message)) { handleWorkerNotification(message as JsonRpcNotification); } else { log.warn("unknown message type from worker"); } } /** * Handle a JSON-RPC response from the worker (matching a pending request). */ function handleResponse(response: JsonRpcResponse): void { const id = response.id; if (id === null || id === undefined) { log.warn("received response with null/undefined id"); return; } const pending = pendingRequests.get(id); if (!pending) { log.warn({ id }, "received response for unknown request id"); return; } clearTimeout(pending.timer); pendingRequests.delete(id); pending.resolve(response); } function readNonEmptyString(value: unknown): string | null { return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; } function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } function deriveInvocationScope( method: HostToWorkerMethodName | string, params: unknown, ): PluginInvocationScope | null { if (!isRecord(params)) return null; const directCompanyId = readNonEmptyString(params.companyId); if (directCompanyId) return { companyId: directCompanyId }; if (method === "performAction" && isRecord(params.actorContext)) { const companyId = readNonEmptyString(params.actorContext.companyId); return companyId ? { companyId } : null; } if (method === "executeTool" && isRecord(params.runContext)) { const companyId = readNonEmptyString(params.runContext.companyId); return companyId ? { companyId } : null; } if (method === "onEvent" && isRecord(params.event)) { const companyId = readNonEmptyString(params.event.companyId); return companyId ? { companyId } : null; } return null; } function registerInvocation(scope: PluginInvocationScope, ttlMs?: number): PluginInvocationContext { // Mint a W3C `traceparent` from the active startup span, so the worker's // provider span can parent to it. The host keeps the value on its own record // (below) and never trusts the worker to supply the parent. Outside a // measured startup step there is no active span, so this is undefined. const activeStep = getActiveStepContext(); const traceparent = activeStep ? traceparentFromContextToken(activeStep.parentContext) : undefined; const invocation: PluginInvocationContext = { id: randomUUID(), scope, ...(traceparent ? { traceparent } : {}), }; const entry: ActiveInvocation = { scope, traceparent }; if (ttlMs !== undefined) { entry.timer = setTimeout(() => { activeInvocations.delete(invocation.id); }, ttlMs); if (entry.timer.unref) entry.timer.unref(); } activeInvocations.set(invocation.id, entry); return invocation; } function clearInvocation(invocation: PluginInvocationContext | null): void { if (!invocation) return; const entry = activeInvocations.get(invocation.id); if (entry?.timer) clearTimeout(entry.timer); activeInvocations.delete(invocation.id); } // Store the host-owned execute route for one active execute call. The host // holds the exact company id and log sink; the worker never supplies them. function registerExecuteRoute( invocationId: string, companyId: string, onLog: ExecuteLogSink, ): void { activeExecuteRoutes.set(invocationId, { companyId, onLog, deliveredChars: 0, crossCompanyBlocked: false, }); } function clearExecuteRoute(invocationId: string | undefined): void { if (invocationId) activeExecuteRoutes.delete(invocationId); } // Drop an oversized incoming worker line before the JSON parse. Write a // rate-limited warn record with the length and a running drop count. The // record never carries the line bytes. function dropOversizedLine(lineLength: number): void { oversizedLineDropCount += 1; const nowMs = Date.now(); if (nowMs - oversizedLineLoggedAtMs >= EXECUTE_LOG_DROP_LOG_INTERVAL_MS) { log.warn( { lineLength, maxIncomingMessageChars, droppedSinceLastLog: oversizedLineDropCount }, "dropping oversized worker line before JSON parse", ); oversizedLineLoggedAtMs = nowMs; oversizedLineDropCount = 0; } } // Drop an `execute.log` notification. Write a rate-limited debug record with // the reason and a running drop count. The record never carries the chunk // bytes, the company id, or command data. function dropExecuteLogNotification(reason: string): void { executeLogDropCount += 1; const nowMs = Date.now(); if (nowMs - executeLogDropLoggedAtMs >= EXECUTE_LOG_DROP_LOG_INTERVAL_MS) { log.debug( { reason, droppedSinceLastLog: executeLogDropCount }, "dropping execute.log notification", ); executeLogDropLoggedAtMs = nowMs; executeLogDropCount = 0; } } // Route one `execute.log` notification to its host-owned execute route. The // route is the single authority for the delivery target and the company // binding. This never reads a company id from the notification and never // routes through the generic active-invocation record. // // Complete mediation: the host and the worker share one stdio pipe, and the // worker process sees every active invocation id. So the host cannot prove // which concurrent invocation produced a notification, and it must NOT treat // the worker-supplied `paperclipInvocationId` alone as proof of origin. The // host validates the exact company scope instead: it delivers only while every // active execute route on this worker belongs to ONE company. When a second // company's execute overlaps, the host fails closed — it latches the active // routes and drops the chunk — so a worker that runs company A can never forge // company B's active id and inject output into B's route. The final command // result still delivers the complete output, so no byte is lost; only the live // stream pauses while two companies overlap. function routeExecuteLogNotification(notification: JsonRpcNotification): void { const invocationId = readNonEmptyString( (notification as { paperclipInvocationId?: unknown }).paperclipInvocationId, ); const params = isRecord(notification.params) ? notification.params : {}; const stream = params.stream; const chunk = params.chunk; // Runtime-validate the payload. Drop invalid input without a throw. if (stream !== "stdout" && stream !== "stderr") { dropExecuteLogNotification("invalid-stream"); return; } if ( typeof chunk !== "string" || chunk.length === 0 || chunk.length > MAX_EXECUTE_LOG_CHUNK_CHARS ) { dropExecuteLogNotification("invalid-chunk"); return; } if (!invocationId) { dropExecuteLogNotification("missing-invocation"); return; } const route = activeExecuteRoutes.get(invocationId); if (!route) { // No active execute route for this id: a late chunk after settlement or // timeout, a non-execute invocation, or an unknown id. Drop it. dropExecuteLogNotification("no-active-route"); return; } // The route already lost single-company attribution earlier in its life, so // it stays closed for the rest of the call. if (route.crossCompanyBlocked) { dropExecuteLogNotification("cross-company-scope"); return; } // Validate the exact company scope. Deliver only while every active execute // route on this worker belongs to one company. A second company's active // route makes the shared pipe ambiguous, so the host fails closed: it // latches every active route and drops the chunk. let onlyCompanyId: string | null = null; let crossCompany = false; for (const active of activeExecuteRoutes.values()) { if (onlyCompanyId === null) { onlyCompanyId = active.companyId; } else if (onlyCompanyId !== active.companyId) { crossCompany = true; break; } } if (crossCompany) { for (const active of activeExecuteRoutes.values()) { active.crossCompanyBlocked = true; } dropExecuteLogNotification("cross-company-scope"); return; } // Bound the total characters one execute call may stream. Past the ceiling // the host drops further chunks, so one runaway or hostile execution cannot // flood the host and the run-log sink without limit. if (route.deliveredChars + chunk.length > maxExecuteLogTotalChars) { dropExecuteLogNotification("execute-output-cap"); return; } route.deliveredChars += chunk.length; try { const delivery = route.onLog(stream, chunk); if (delivery && typeof (delivery as Promise).then === "function") { void (delivery as Promise).catch((err) => { log.error( { err: err instanceof Error ? err.message : String(err) }, "execute.log delivery failed", ); }); } } catch (err) { log.error( { err: err instanceof Error ? err.message : String(err) }, "execute.log delivery threw", ); } } // ----------------------------------------------------------------------- // Host-owned setup-token login pseudo-terminal route gate // ----------------------------------------------------------------------- // The manager owns one live login pseudo-terminal route per worker. It mints a // host-owned opaque route identifier, carries it in the open call, and keys the // close on it, so it closes a worker-created terminal even when the open reply // was lost and no worker session identifier arrived. It binds the worker // session identifier one time while the route is `opening`, for output only. It // never trusts a worker-supplied identifier as proof of origin: it delivers // output only while the route is `open` and the notification carries the exact // bound identifier and valid bounded bytes, and it never logs the raw bytes. It // terminalizes the route exactly once on every open failure path, closes the // terminal by the host route identifier, and admits a new open only after it // verifies a close acknowledgement bound to that identifier; it retires the // worker on an unconfirmed close. // A single-consumer route state. The login pseudo-terminal route and the // generic duplex channel route share it. type RouteState = "reserved" | "opening" | "open" | "closed"; // Shared route-binding helpers. The login pseudo-terminal route and the duplex // channel route both use them, so the two routes bind and settle one way. // Settle the route wait exactly once. Replace the settler with a no-op, so a // later exit or terminalize never settles the wait a second time. function settleRouteWait( route: { settleWait: (value: { exitCode: number | null }) => void }, value: { exitCode: number | null }, ): void { const settle = route.settleWait; route.settleWait = () => {}; settle(value); } // Read the worker session identifier from an open reply, but only when the // route can still bind. Return null for a malformed reply, or for a route that // already left `opening` or terminalized. A late or a duplicate reply never // binds, revives, or reopens a route. function readBindableWorkerSessionId( route: { state: RouteState; terminalized: boolean }, openResult: unknown, ): string | null { const workerSessionId = readNonEmptyString( isRecord(openResult) ? openResult.workerSessionId : null, ); if (!workerSessionId || route.state !== "opening" || route.terminalized) { return null; } return workerSessionId; } type SetupTokenPtyRouteState = RouteState; interface SetupTokenPtyRoute { hostRouteId: string; state: SetupTokenPtyRouteState; workerSessionId: string | null; listener: ((chunk: string) => void) | null; buffered: string[]; deliveredChars: number; terminalized: boolean; settleWait: (value: { exitCode: number | null }) => void; } // At most one active credential pseudo-terminal per worker. A non-null route // blocks a second open until the manager confirms the first route's close. let setupTokenPtyRoute: SetupTokenPtyRoute | null = null; // Close the worker terminal by the host route identifier and verify the bound // acknowledgement. Return true only when the worker returns an acknowledgement // that carries the exact host route identifier. An absent, malformed, // mismatched, or timed-out acknowledgement returns false, so the caller fails // closed. async function closeSetupTokenPtyTerminal(hostRouteId: string): Promise { try { const ack = await callInternal( "setupTokenPtyClose", { hostRouteId }, setupTokenPtyCloseTimeoutMs, ); return isRecord(ack) && readNonEmptyString(ack.hostRouteId) === hostRouteId; } catch { return false; } } // Terminalize the route exactly once. Resolve the login wait, close the worker // terminal by the host route identifier, and free the per-worker slot only // after the close resolves. Retire the worker when the close is unconfirmed. async function terminalizeSetupTokenPtyRoute(route: SetupTokenPtyRoute): Promise { if (route.terminalized) return; route.terminalized = true; route.state = "closed"; route.listener = null; route.buffered = []; // A terminalized route reports a null exit code, which the runner treats as a // failure. settleRouteWait(route, { exitCode: null }); const confirmed = await closeSetupTokenPtyTerminal(route.hostRouteId); if (setupTokenPtyRoute === route) setupTokenPtyRoute = null; if (!confirmed) { // The worker did not acknowledge the close, so the host cannot prove the // terminal is gone. Fail closed: retire the worker before any reuse. log.error( { pluginId }, "setup-token login pseudo-terminal close not acknowledged; retiring worker", ); void killProcess(); } } // Route one login pseudo-terminal output notification to the per-session // listener. Deliver only while the route is `open` and the notification carries // the exact bound worker session identifier and valid bounded bytes. Drop an // unknown, late, malformed, or mismatched notification. Never log the raw bytes. function routeSetupTokenPtyOutput(notification: JsonRpcNotification): void { const route = setupTokenPtyRoute; if (!route || route.state !== "open") return; const params = isRecord(notification.params) ? notification.params : {}; const workerSessionId = readNonEmptyString(params.workerSessionId); if (!workerSessionId || workerSessionId !== route.workerSessionId) return; const chunk = params.chunk; if ( typeof chunk !== "string" || chunk.length === 0 || chunk.length > maxSetupTokenPtyChunkChars ) { return; } if (route.deliveredChars + chunk.length > maxSetupTokenPtyTotalChars) { // The cumulative output passed the per-route bound. Terminalize the route. void terminalizeSetupTokenPtyRoute(route); return; } route.deliveredChars += chunk.length; if (route.listener) route.listener(chunk); else route.buffered.push(chunk); } // Route one login pseudo-terminal exit notification to the login wait. Resolve // only while the route is `open` and the notification carries the exact bound // worker session identifier. function routeSetupTokenPtyExit(notification: JsonRpcNotification): void { const route = setupTokenPtyRoute; if (!route || route.state !== "open") return; const params = isRecord(notification.params) ? notification.params : {}; const workerSessionId = readNonEmptyString(params.workerSessionId); if (!workerSessionId || workerSessionId !== route.workerSessionId) return; const exitCode = typeof params.exitCode === "number" ? params.exitCode : null; settleRouteWait(route, { exitCode }); } // Close the one route on a worker exit. The worker is gone, so the manager // resolves the login wait with the fixed non-secret exit and clears the route // one time. The pending pseudo-terminal calls reject through `rejectAllPending`. function closeSetupTokenPtyRouteOnWorkerExit(): void { const route = setupTokenPtyRoute; if (!route) return; setupTokenPtyRoute = null; route.terminalized = true; route.state = "closed"; route.listener = null; route.buffered = []; settleRouteWait(route, { exitCode: null }); } // Open one live login pseudo-terminal route. Reserve the route // before the open call, bind the worker session identifier one time on the // first successful open reply, and return a session the login transport drives. // Terminalize the route on every open failure path. async function openSetupTokenPtySession( input: SetupTokenPtyOpenInput, ): Promise { if (input.command !== CLAUDE_SETUP_TOKEN_COMMAND) { // Allowlist the login command. Only the fixed `CLAUDE_SETUP_TOKEN_COMMAND` // may run in the sandbox pseudo-terminal. Reject any other command with one // fixed non-secret error before the worker call, so a caller cannot spawn // an arbitrary process in the sandbox pseudo-terminal. throw new Error(SETUP_TOKEN_PTY_COMMAND_NOT_ALLOWED); } if (setupTokenPtyRoute) { // A route for this worker is not yet closed and confirmed. Reject the // second open with one fixed non-secret error before it reaches the worker. throw new Error(SETUP_TOKEN_PTY_ROUTE_BUSY); } const hostRouteId = randomUUID(); let settleWait: (value: { exitCode: number | null }) => void = () => {}; const waitPromise = new Promise<{ exitCode: number | null }>((resolve) => { settleWait = resolve; }); const route: SetupTokenPtyRoute = { hostRouteId, state: "reserved", workerSessionId: null, listener: null, buffered: [], deliveredChars: 0, terminalized: false, settleWait, }; setupTokenPtyRoute = route; route.state = "opening"; let openResult: HostToWorkerMethods["setupTokenPtyOpen"][1]; try { openResult = await callInternal( "setupTokenPtyOpen", { hostRouteId, driverKey: input.driverKey, companyId: input.companyId, environmentId: input.environmentId, providerLeaseId: input.providerLeaseId, command: input.command, }, setupTokenPtyOpenTimeoutMs, ); } catch (err) { // A send failure, an RPC rejection, or an open timeout. Terminalize the // route exactly once and fail closed. await terminalizeSetupTokenPtyRoute(route); throw err instanceof Error ? err : new Error(SETUP_TOKEN_PTY_OPEN_FAILED); } const workerSessionId = readBindableWorkerSessionId(route, openResult); if (!workerSessionId) { // A malformed reply, or a route that already left `opening`. A late or a // duplicate reply never binds, revives, or reopens a route. await terminalizeSetupTokenPtyRoute(route); throw new Error(SETUP_TOKEN_PTY_OPEN_FAILED); } // Bind the worker session identifier one time and move the route to `open`. route.workerSessionId = workerSessionId; route.state = "open"; return { onData(listener: (chunk: string) => void): void { route.listener = listener; if (route.buffered.length > 0) { const pending = route.buffered; route.buffered = []; for (const chunk of pending) listener(chunk); } }, write(data: string): void { const sid = route.workerSessionId; if (route.state !== "open" || !sid) return; void callInternal( "setupTokenPtyInput", { workerSessionId: sid, data }, setupTokenPtyOpenTimeoutMs, ).catch(() => {}); }, wait(): Promise<{ exitCode: number | null }> { return waitPromise; }, kill(): void { const sid = route.workerSessionId; if (!sid) return; void callInternal( "setupTokenPtyStop", { workerSessionId: sid }, setupTokenPtyOpenTimeoutMs, ).catch(() => {}); }, async close(): Promise { await terminalizeSetupTokenPtyRoute(route); }, }; } // ----------------------------------------------------------------------- // Host-owned generic duplex channel route // ----------------------------------------------------------------------- // The duplex channel route mirrors the login pseudo-terminal route model. The // host owns the route identifier, binds the worker session identifier one time // on a valid open reply, and keys the close on the host route identifier. It // rejects a late or a duplicate open reply, and it retires the worker on an // unconfirmed close. The duplex channel carries no command allowlist, so the // caller owns the command. // // The route adds seven explicit bounds the pseudo-terminal route lacks. Each // bound ends the route when it passes its limit: // 1. pre-bind buffered bytes — the cumulative characters the host buffers // before a data listener attaches; // 2. pre-bind buffered frame count — the number of data frames the host // buffers before a data listener attaches; // 3. pending request count — the number of in-flight host→worker requests; // 4. host→worker write size — the characters for one write; // 5. protocol error rate — the count of malformed or mismatched data frames; // 6. total data bytes — the cumulative inbound bytes over the whole life, // counted before and after a data listener attaches; // 7. route lifetime — the milliseconds from the open to the terminal end. interface DuplexChannelRoute { hostRouteId: string; state: RouteState; workerSessionId: string | null; listener: ((chunk: string) => void) | null; buffered: string[]; bufferedChars: number; pendingRequests: number; protocolErrors: number; totalDataBytes: number; lifetimeTimer: ReturnType | null; terminalized: boolean; settleWait: (value: { exitCode: number | null }) => void; // The data frames that arrived before the bind, held in order. The bind // replays them through the exact-pair routing, so an early frame is never // lost and a frame whose pair does not match the bound pair still fails // closed. The hold ceiling is `maxDuplexChannelPreBindHoldFrames`, one frame // above the buffered bound, so the replay's buffered-bound check ends the // route, not the hold. preBind: JsonRpcNotification[]; // The single exit frame that arrived before the bind. An exit never consumes // a data hold slot, so a worker that batches an exit among enough data frames // to fill the hold cannot crowd out a data frame. The bind replays the held // data frames first, then this exit last. preBindExit: JsonRpcNotification | null; } // The live duplex routes on this worker, keyed by the exact // `{ hostRouteId, workerSessionId }` pair. The host binds one pair once, at // open, and routes each data, exit, and close frame only to the route that owns // its exact pair. The host never routes by the worker session id alone. const liveDuplexRoutes = new Map(); // The reserved-or-opening routes, keyed by the host route id. A route lives here // from the open call until the worker session id binds. The host then moves it // to `liveDuplexRoutes` under the exact pair key. const openingDuplexRoutes = new Map(); // The tombstoned pairs on this worker, keyed by the exact pair key. The host // installs a tombstone atomically when it removes a live binding, and retains it // until the worker retires. A late frame for a tombstoned pair reaches no // listener. A closed pair never returns. const duplexPairTombstones = new Set(); // The process-scoped aggregate route-slot controller. The manager injects it, so // the ceiling counts every duplex route across every worker in the process, not // one agent's setting. When it is absent, the worker admits every open, so the // handle runs unbounded in isolation (a unit test constructs it this way). const duplexRouteSlots = options.duplexRouteSlots ?? null; // The routes that currently hold one aggregate slot. The host releases a slot one // time per route, so a double terminalize never releases two slots. const duplexRouteSlotHolders = new Set(); // Try to reserve one aggregate route slot for a route. Return true when the // route holds a slot after the call. When no controller is present, the route // always holds a slot. function acquireDuplexRouteSlot(route: DuplexChannelRoute): boolean { if (!duplexRouteSlots) return true; if (!duplexRouteSlots.tryAcquire()) return false; duplexRouteSlotHolders.add(route); return true; } // Release the aggregate route slot a route holds, one time. A route that never // held a slot releases nothing. function releaseDuplexRouteSlot(route: DuplexChannelRoute): void { if (!duplexRouteSlotHolders.delete(route)) return; duplexRouteSlots?.release(); } // Retire the worker at once. The host kills the process, so every live route, // every reserved route, and every tombstone drops together. The host uses this // as the fail-closed response to an ownership violation and to a tombstone-set // overflow. The host never logs the raw frame content on a violation. function retireDuplexWorkerOnViolation(reason: string): void { log.error({ pluginId }, `duplex channel ownership violation (${reason}); retiring worker`); void killProcess(); } // Install a tombstone for a bound pair atomically with the removal of its live // binding. This runs in one synchronous step, before any slot release or reuse, // so a late frame for the pair reaches no listener. Before the set would exceed // its hard bound, retire the worker, which drops every route and every tombstone // at once. A tombstone overflow fails closed; the host never evicts a tombstone // and lets a pair return. function tombstoneBoundDuplexPair(route: DuplexChannelRoute): void { if (route.workerSessionId === null) return; const pairKey = duplexPairKey(route.hostRouteId, route.workerSessionId); liveDuplexRoutes.delete(pairKey); if (!duplexPairTombstones.has(pairKey)) { if (duplexPairTombstones.size >= MAX_DUPLEX_ROUTE_TOMBSTONES) { // The set is full. Do not evict a tombstone. Retire the worker, which // drops every route and every tombstone together, so no closed pair ever // returns. retireDuplexWorkerOnViolation("tombstone set overflow"); return; } duplexPairTombstones.add(pairKey); } } // Close the worker channel by the host route identifier and verify the bound // acknowledgement. Return true only when the worker returns an acknowledgement // for the exact pair. Before a session binds, the acknowledgement carries the // host route identifier only, so a lost open reply still permits a route-only // close. After a session binds, the acknowledgement must echo both the host // route identifier and the bound worker session identifier. An absent, // malformed, mismatched, or timed-out acknowledgement returns false, so the // caller fails closed. async function closeDuplexChannelTerminal( hostRouteId: string, boundWorkerSessionId: string | null, ): Promise { try { const ack = await callInternal( "duplexChannelClose", { hostRouteId }, duplexChannelCloseTimeoutMs, ); if (!isRecord(ack) || readNonEmptyString(ack.hostRouteId) !== hostRouteId) { return false; } if (boundWorkerSessionId !== null) { // A bound close: the acknowledgement must echo the exact worker session // identifier. A missing or a mismatched identifier is invalid. return readNonEmptyString(ack.workerSessionId) === boundWorkerSessionId; } return true; } catch { return false; } } // Terminalize the route exactly once. Resolve the wait, close the worker // channel by the host route identifier, and free the per-worker slot only // after the close resolves. Retire the worker when the close is unconfirmed. // Clear the route lifetime timer one time. Every terminal path and the // worker-exit path calls this, so a timer never fires after the route ends. function clearDuplexChannelLifetimeTimer(route: DuplexChannelRoute): void { if (route.lifetimeTimer) { clearTimeout(route.lifetimeTimer); route.lifetimeTimer = null; } } async function terminalizeDuplexChannelRoute(route: DuplexChannelRoute): Promise { if (route.terminalized) return; route.terminalized = true; route.state = "closed"; route.listener = null; // Keep the buffered chunks the host accepted before the route ended, so a // listener that attaches after the end still drains them. A frame can end the // route during the pre-bind replay, before a listener attaches, and the // chunks the host accepted before that frame are valid data the listener must // still receive. The buffered bytes stay bounded by the pre-bind buffered // bound, and `onData` clears them once it drains them. Drop the held pre-bind // frames, which never bound to a listener. route.preBind = []; route.preBindExit = null; clearDuplexChannelLifetimeTimer(route); // Remove the live binding and install the tombstone in one synchronous step, // before the worker close and before any reuse. A reserved route that never // bound leaves the opening map only; it has no pair to tombstone, and the // monotonic host route id never returns. openingDuplexRoutes.delete(route.hostRouteId); tombstoneBoundDuplexPair(route); releaseDuplexRouteSlot(route); // A terminalized route reports a null exit code, which the caller treats as a // failure. settleRouteWait(route, { exitCode: null }); const confirmed = await closeDuplexChannelTerminal(route.hostRouteId, route.workerSessionId); if (!confirmed) { // The worker did not acknowledge the close, so the host cannot prove the // channel is gone. Fail closed: retire the worker before any reuse. log.error( { pluginId }, "duplex channel close not acknowledged; retiring worker", ); void killProcess(); } } // Count one protocol error for the route. End the route when the count passes // the per-route budget, so a flood of malformed or mismatched frames bounds the // route. function recordDuplexChannelProtocolError(route: DuplexChannelRoute): void { route.protocolErrors += 1; if (route.protocolErrors > maxDuplexChannelProtocolErrors) { void terminalizeDuplexChannelRoute(route); } } // Deliver one duplex channel chunk to the bound listener in isolation. A // listener that throws must not escape the worker stdout notification handler // or the buffered replay, so a throw here breaks neither the notification // dispatch loop nor the pre-bind drain. The manager catches the error and logs // it without the raw bytes. This mirrors the `execute.log` delivery isolation. function deliverDuplexChannelChunk( listener: (chunk: string) => void, chunk: string, ): void { try { listener(chunk); } catch (err) { log.error( { err: err instanceof Error ? err.message : String(err) }, "duplex channel data delivery threw", ); } } // Route one duplex channel data notification to the per-session listener. // Deliver only while the route is `open` and the notification carries the exact // bound worker session identifier and a valid chunk. Count a mismatched or // malformed frame as a protocol error. End the route at once when one chunk is // larger than the per-chunk limit or when the cumulative bytes pass the total // cap. Buffer a valid frame under the pre-bind bounds when no listener has // attached yet. Never log the raw bytes. // Resolve one inbound duplex frame to its owning route by the exact pair. Return // the live route on an exact live-pair match. Return `"tombstoned"` for a late // frame whose exact pair the host already closed; that frame reaches no listener // and changes no state. Return `"violation"` for any unknown, foreign, // duplicate, ambiguous, or malformed pair; the caller fails closed and retires // the worker. The lookup never routes by the worker session id alone. function resolveDuplexRouteByPair( hostRouteId: string | null, workerSessionId: string | null, ): DuplexChannelRoute | "tombstoned" | "opening" | "violation" { if (!hostRouteId || !workerSessionId) return "violation"; const pairKey = duplexPairKey(hostRouteId, workerSessionId); const live = liveDuplexRoutes.get(pairKey); if (live && live.state === "open") return live; if (duplexPairTombstones.has(pairKey)) return "tombstoned"; // A frame whose host route id names a route that is still opening arrived // before the bind. It is the host's own reserved route, not a foreign frame. // The caller defers it, so the bind replays it through the exact-pair routing. if (openingDuplexRoutes.has(hostRouteId)) return "opening"; return "violation"; } function routeDuplexChannelData(notification: JsonRpcNotification): void { const params = isRecord(notification.params) ? notification.params : {}; const hostRouteId = readNonEmptyString(params.hostRouteId); const workerSessionId = readNonEmptyString(params.workerSessionId); const resolved = resolveDuplexRouteByPair(hostRouteId, workerSessionId); if (resolved === "tombstoned") { // A late frame for a closed pair. It reaches no listener and changes no // state. Never log the raw frame content. return; } if (resolved === "violation") { // An unknown, foreign, duplicate, ambiguous, or malformed pair. Fail closed: // retire the worker. Never log the raw frame content. retireDuplexWorkerOnViolation("data frame pair"); return; } if (resolved === "opening") { // The frame arrived before the bind. Hold it; the bind replays it through the // exact-pair routing. bufferPreBindDuplexFrame(hostRouteId, notification); return; } const route = resolved; const chunk = params.chunk; if (typeof chunk !== "string" || chunk.length === 0) { // The exact pair matches, but the chunk is malformed. Count one per-route // protocol error. recordDuplexChannelProtocolError(route); return; } if (chunk.length > maxDuplexChannelChunkChars) { // One inbound chunk is larger than the per-chunk limit. End the route at // once. Do not count the chunk as a protocol error. void terminalizeDuplexChannelRoute(route); return; } // Count the bytes of the chunk. Enforce the cumulative total-byte cap before // and after a listener attaches. End the route when the cap is exceeded, so a // bound listener cannot receive data past the cap. const chunkBytes = Buffer.byteLength(chunk); if (route.totalDataBytes + chunkBytes > maxDuplexChannelTotalDataBytes) { void terminalizeDuplexChannelRoute(route); return; } route.totalDataBytes += chunkBytes; if (route.listener) { deliverDuplexChannelChunk(route.listener, chunk); return; } // No listener attached yet. Buffer the frame under the pre-bind bounds. End // the route when the cumulative bytes or the frame count passes the bound. if ( route.buffered.length + 1 > maxDuplexChannelPreBindFrames || route.bufferedChars + chunk.length > maxDuplexChannelPreBindChars ) { void terminalizeDuplexChannelRoute(route); return; } route.buffered.push(chunk); route.bufferedChars += chunk.length; } // Route one duplex channel exit notification to the wait. Resolve only while // the route is `open` and the notification carries the exact bound worker // session identifier. function routeDuplexChannelExit(notification: JsonRpcNotification): void { const params = isRecord(notification.params) ? notification.params : {}; const hostRouteId = readNonEmptyString(params.hostRouteId); const workerSessionId = readNonEmptyString(params.workerSessionId); const resolved = resolveDuplexRouteByPair(hostRouteId, workerSessionId); if (resolved === "tombstoned") { // A late exit for a closed pair. It reaches no wait and changes no state. return; } if (resolved === "violation") { // An unknown, foreign, duplicate, ambiguous, or malformed pair. Fail closed: // retire the worker. retireDuplexWorkerOnViolation("exit frame pair"); return; } if (resolved === "opening") { // The exit arrived before the bind. Hold it; the bind replays it through the // exact-pair routing. bufferPreBindDuplexFrame(hostRouteId, notification); return; } const exitCode = typeof params.exitCode === "number" ? params.exitCode : null; settleRouteWait(resolved, { exitCode }); } // Hold one frame that arrived before its route bound, in order. The bind replays // the held frames through the exact-pair routing. An exit frame never consumes a // data hold slot; the host holds it in the single `preBindExit` slot instead, so // a worker that batches an exit among enough data frames to fill the hold cannot // crowd out a data frame. The data hold ceiling stays one frame above the // buffered bound, so the replay's buffered-bound check, not the hold, ends the // route. The host counts one protocol error for each data frame past the ceiling, // so an early flood bounds the hold instead of growing without limit. function bufferPreBindDuplexFrame( hostRouteId: string | null, notification: JsonRpcNotification, ): void { if (!hostRouteId) return; const route = openingDuplexRoutes.get(hostRouteId); if (!route) return; if (notification.method === DUPLEX_CHANNEL_EXIT_NOTIFICATION) { route.preBindExit = notification; return; } if (route.preBind.length >= maxDuplexChannelPreBindHoldFrames) { recordDuplexChannelProtocolError(route); return; } route.preBind.push(notification); } // Replay the frames a route held before it bound. The route is live now, so the // exact-pair routing delivers each held frame or fails closed on a mismatch. // Replay the held data frames first, in order, then the held exit last, so the // data delivers before the exit resolves the wait, which matches a real worker's // order. A frame that ends the route terminalizes it, and every later frame in // the replay is a no-op, because the routing functions drop a frame for a route // that is not `open`. function replayPreBindDuplexFrames(route: DuplexChannelRoute): void { const held = route.preBind; route.preBind = []; for (const notification of held) { if (route.terminalized) break; routeDuplexChannelData(notification); } const heldExit = route.preBindExit; route.preBindExit = null; if (heldExit && !route.terminalized) { routeDuplexChannelExit(heldExit); } } // Close every route on a worker exit. The worker is gone, so the manager // resolves each wait with the fixed non-secret exit, releases each aggregate // slot, and clears the live, opening, and tombstone state. A worker exit is a // worker retirement, so the host drops every tombstone; the monotonic host route // id never returns, so no closed pair can revive on a restart. The pending // channel calls reject through `rejectAllPending`. function closeDuplexChannelRouteOnWorkerExit(): void { const routes = [...openingDuplexRoutes.values(), ...liveDuplexRoutes.values()]; openingDuplexRoutes.clear(); liveDuplexRoutes.clear(); duplexPairTombstones.clear(); for (const route of routes) { if (route.terminalized) continue; route.terminalized = true; route.state = "closed"; route.listener = null; route.buffered = []; route.bufferedChars = 0; route.preBind = []; route.preBindExit = null; clearDuplexChannelLifetimeTimer(route); releaseDuplexRouteSlot(route); settleRouteWait(route, { exitCode: null }); } } // Open one live generic duplex channel route. Reserve the route before the open // call, bind the worker session identifier one time on the first successful // open reply, and return a session a caller drives. Terminalize the route on // every open failure path. async function openDuplexChannel( input: DuplexChannelOpenInput, ): Promise { const hostRouteId = nextDuplexHostRouteId(); let settleWait: (value: { exitCode: number | null }) => void = () => {}; const waitPromise = new Promise<{ exitCode: number | null }>((resolve) => { settleWait = resolve; }); const route: DuplexChannelRoute = { hostRouteId, state: "reserved", workerSessionId: null, listener: null, buffered: [], bufferedChars: 0, pendingRequests: 0, protocolErrors: 0, totalDataBytes: 0, lifetimeTimer: null, terminalized: false, settleWait, preBind: [], preBindExit: null, }; // Reserve one aggregate route slot before any work. When the process-scoped // ceiling is full, reject with the fixed route-busy error and open nothing, so // an active channel never downgrades and the ceiling never overcommits. if (!acquireDuplexRouteSlot(route)) { throw new Error(DUPLEX_CHANNEL_ROUTE_BUSY); } openingDuplexRoutes.set(hostRouteId, route); route.state = "opening"; let openResult: HostToWorkerMethods["duplexChannelOpen"][1]; try { openResult = await callInternal( "duplexChannelOpen", { hostRouteId, driverKey: input.driverKey, companyId: input.companyId, environmentId: input.environmentId, providerLeaseId: input.providerLeaseId, command: input.command, }, duplexChannelOpenTimeoutMs, ); } catch (err) { // A send failure, an RPC rejection, or an open timeout. Terminalize the // route exactly once and fail closed. await terminalizeDuplexChannelRoute(route); throw err instanceof Error ? err : new Error(DUPLEX_CHANNEL_OPEN_FAILED); } const workerSessionId = readBindableWorkerSessionId(route, openResult); // Verify the worker echoed the exact host route id the open request carried. // A reply with a missing or a mismatched host route id never binds, so the // host binds only a reply that proves the worker holds the exact pair. const echoedHostRouteId = isRecord(openResult) ? readNonEmptyString(openResult.hostRouteId) : null; if (!workerSessionId || echoedHostRouteId !== hostRouteId) { // A malformed reply, a mismatched host route id, or a route that already // left `opening`. A late or a duplicate reply never binds, revives, or // reopens a route. await terminalizeDuplexChannelRoute(route); throw new Error(DUPLEX_CHANNEL_OPEN_FAILED); } const pairKey = duplexPairKey(hostRouteId, workerSessionId); if (liveDuplexRoutes.has(pairKey) || duplexPairTombstones.has(pairKey)) { // The worker returned a pair that is already live or already tombstoned. Fail // closed: terminalize this route and retire the worker before any reuse. await terminalizeDuplexChannelRoute(route); retireDuplexWorkerOnViolation("duplicate bound pair"); throw new Error(DUPLEX_CHANNEL_OPEN_FAILED); } // Bind the worker session identifier one time and move the route from the // opening map to the live map under the exact pair key. route.workerSessionId = workerSessionId; route.state = "open"; openingDuplexRoutes.delete(hostRouteId); liveDuplexRoutes.set(pairKey, route); // Start the route lifetime timer now the route is open. The route ends when // the timer expires. Every terminal path and the worker-exit path clears the // timer. Unreference the timer so it never blocks the host process shutdown. route.lifetimeTimer = setTimeout(() => { void terminalizeDuplexChannelRoute(route); }, maxDuplexChannelDurationMs); route.lifetimeTimer.unref?.(); // Replay any frame that arrived before the bind. The route is live now, so the // exact-pair routing delivers each held frame or fails closed on a mismatch. replayPreBindDuplexFrames(route); // Send one host→worker request under the pending-request bound. End the route // when too many requests are in-flight, so a worker that never replies cannot // make the host hold an unbounded number of pending requests. const sendBoundedRequest = < M extends "duplexChannelWrite" | "duplexChannelStop", >( method: M, params: HostToWorkerMethods[M][0], ): void => { if (route.state !== "open") return; if (route.pendingRequests >= maxDuplexChannelPendingRequests) { void terminalizeDuplexChannelRoute(route); return; } route.pendingRequests += 1; void callInternal(method, params, duplexChannelOpenTimeoutMs) .catch(() => {}) .finally(() => { route.pendingRequests -= 1; }); }; return { onData(listener: (chunk: string) => void): void { route.listener = listener; if (route.buffered.length > 0) { const pending = route.buffered; route.buffered = []; route.bufferedChars = 0; for (const chunk of pending) deliverDuplexChannelChunk(listener, chunk); } }, write(data: string): void { const sid = route.workerSessionId; if (route.state !== "open" || !sid) return; if (data.length > maxDuplexChannelWriteChars) { // The write is larger than the size bound. End the route before the // write reaches the worker. void terminalizeDuplexChannelRoute(route); return; } sendBoundedRequest("duplexChannelWrite", { hostRouteId: route.hostRouteId, workerSessionId: sid, data, }); }, wait(): Promise<{ exitCode: number | null }> { return waitPromise; }, kill(): void { const sid = route.workerSessionId; if (!sid) return; sendBoundedRequest("duplexChannelStop", { hostRouteId: route.hostRouteId, workerSessionId: sid }); }, async close(): Promise { await terminalizeDuplexChannelRoute(route); }, }; } /** * Extract the single company a worker→host call references, mirroring the SDK * governed-access gate's own derivation (host-client-factory.ts * `requestedCompanyScope`) so a proactive call resolves to exactly the company * the gate would require: * - explicit `params.companyId`; * - a company-scoped state key (`scopeKind: "company"` + `scopeId`); * - `events.subscribe`'s `params.filter.companyId` (how the SDK's * `ctx.events.on(name, { companyId }, fn)` issues its subscribe). * * Returns null whenever the gate treats the call as a wildcard (`companies.list`, * a `scopeKind: "company"` key with no `scopeId`) or as referencing no company * (instance-scoped state, an unfiltered subscribe). A wildcard is deliberately * NOT granted proactively: proactive resolution only ever admits a single, * explicit company, never "all". This keeps the resolver and the gate in * lockstep in the functional direction (LOOA-693 AC#4 / LOOA-695). */ function referencedCompanyId(method: string, params: unknown): string | null { // Gate returns { kind: "all" } for companies.list regardless of params — // never a single company — so proactive access declines it here. if (method === "companies.list") return null; if (!isRecord(params)) return null; const direct = readNonEmptyString(params.companyId); if (direct) return direct; if (params.scopeKind === "company") { // scopeId present → that company; absent → wildcard ("all") in the gate, // which we never grant proactively → null. return readNonEmptyString(params.scopeId); } if (method === "events.subscribe" && isRecord(params.filter)) { return readNonEmptyString(params.filter.companyId); } return null; } function contextForWorkerMessage(message: JsonRpcRequest | JsonRpcNotification): WorkerHostCallContext { const invocationId = readNonEmptyString( (message as { paperclipInvocationId?: unknown }).paperclipInvocationId, ); if (!invocationId) { // No host-issued invocation is being echoed. This is a genuinely // proactive worker→host call (timer/loop). If it references a company the // plugin is authorized to act on proactively, resolve it to that // company's scope so the governed-access gate admits it. This never // widens access beyond the plugin's configured companies, and only // applies when the worker is NOT inside a host-issued invocation (which // would carry an id and keep its strict single-company match below). const proactiveCompanyId = referencedCompanyId( message.method, (message as { params?: unknown }).params, ); if (proactiveCompanyId && proactiveCompanyScopes.has(proactiveCompanyId)) { return { invocationScope: { companyId: proactiveCompanyId } }; } const hasActiveInvocation = activeInvocations.size > 0 || Array.from(pendingRequests.values()).some((pending) => pending.invocationId); return hasActiveInvocation ? { invalidInvocationScope: true } : {}; } const entry = activeInvocations.get(invocationId); if (!entry) return { invalidInvocationScope: true }; return { invocationScope: entry.scope, traceparent: entry.traceparent }; } /** * Handle a JSON-RPC request from the worker (worker→host call). */ async function handleWorkerRequest(request: JsonRpcRequest): Promise { const method = request.method as WorkerToHostMethodName; const handler = options.hostHandlers[method] as | ((params: unknown, context?: WorkerHostCallContext) => Promise) | undefined; if (!handler) { log.warn({ method }, "worker called unregistered host method"); try { sendMessage( createErrorResponse( request.id, JSONRPC_ERROR_CODES.METHOD_NOT_FOUND, `Host does not handle method "${method}"`, ), ); } catch { // Worker may have exited, ignore send error } return; } try { const result = await handler(request.params, contextForWorkerMessage(request)); sendMessage({ jsonrpc: JSONRPC_VERSION, id: request.id, result: result ?? null, }); } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); log.error({ method, err: errorMessage }, "host handler error"); try { sendMessage( createErrorResponse( request.id, errorCodeForWorkerHostError(err), errorMessage, ), ); } catch { // Worker may have exited, ignore send error } } } /** * Handle a JSON-RPC notification from the worker (fire-and-forget). * * The `log` notification is the primary case — worker `ctx.logger` calls * arrive here. We append structured plugin context (pluginId, timestamp, * level) so that every log entry is queryable per the spec (§26.1). */ function handleWorkerNotification(notification: JsonRpcNotification): void { if (notification.method === "log") { const params = notification.params as { level?: string; message?: string; meta?: Record; } | null; const level = params?.level ?? "info"; const msg = params?.message ?? ""; const meta = params?.meta; // Build a structured log object that includes the plugin context fields // required by §26.1: pluginId, timestamp, level, message, and metadata. // The child logger already carries `pluginId` in its bindings, but we // add explicit `pluginLogLevel` and `pluginTimestamp` so downstream // consumers (log storage, UI queries) can filter without parsing. const logFields: Record = { ...meta, pluginLogLevel: level, pluginTimestamp: new Date().toISOString(), }; if (level === "error") { log.error(logFields, `[plugin] ${msg}`); } else if (level === "warn") { log.warn(logFields, `[plugin] ${msg}`); } else if (level === "debug") { log.debug(logFields, `[plugin] ${msg}`); } else { log.info(logFields, `[plugin] ${msg}`); } return; } // Execute-log notifications: deliver one incremental output chunk to the // host-owned execute route for the active execute call. if (notification.method === "execute.log") { routeExecuteLogNotification(notification); return; } // Setup-token login pseudo-terminal notifications: deliver output // and the exit to the one host-owned login route, bound by the worker session // identifier while the route is open. if (notification.method === SETUP_TOKEN_PTY_OUTPUT_NOTIFICATION) { routeSetupTokenPtyOutput(notification); return; } if (notification.method === SETUP_TOKEN_PTY_EXIT_NOTIFICATION) { routeSetupTokenPtyExit(notification); return; } // Duplex channel notifications: deliver data and the exit to the one // host-owned duplex route, bound by the worker session identifier while the // route is open. if (notification.method === DUPLEX_CHANNEL_DATA_NOTIFICATION) { routeDuplexChannelData(notification); return; } if (notification.method === DUPLEX_CHANNEL_EXIT_NOTIFICATION) { routeDuplexChannelExit(notification); return; } // Stream notifications: forward to the stream bus via callback if ( notification.method === "streams.open" || notification.method === "streams.emit" || notification.method === "streams.close" ) { const params = (notification.params ?? {}) as Record; const companyId = String(params.companyId ?? ""); const context = contextForWorkerMessage(notification); if (context.invalidInvocationScope) { log.warn( { method: notification.method, companyId }, "dropping plugin stream notification with invalid invocation scope", ); return; } const allowedCompanyId = context.invocationScope?.companyId; if (allowedCompanyId && companyId !== allowedCompanyId) { log.warn( { method: notification.method, companyId, allowedCompanyId }, "dropping plugin stream notification outside invocation company scope", ); return; } // Track open channels so we can emit synthetic close on crash if (notification.method === "streams.open") { const ch = String(params.channel ?? ""); if (ch) openStreamChannels.set(ch, companyId); } else if (notification.method === "streams.close") { openStreamChannels.delete(String(params.channel ?? "")); } if (options.onStreamNotification) { try { options.onStreamNotification(notification.method, params); } catch (err) { log.error( { method: notification.method, err: err instanceof Error ? err.message : String(err), }, "stream notification handler failed", ); } } return; } log.debug({ method: notification.method }, "received notification from worker"); } // ----------------------------------------------------------------------- // Process lifecycle // ----------------------------------------------------------------------- function spawnProcess(): ChildProcess { // Security: Do NOT spread process.env into the worker. Plugins should only // receive a minimal, controlled environment to prevent leaking host // secrets (like DATABASE_URL, internal API keys, etc.). const workerEnv: Record = { ...options.env, PATH: process.env.PATH ?? "", NODE_PATH: process.env.NODE_PATH ?? "", PAPERCLIP_PLUGIN_ID: pluginId, NODE_ENV: process.env.NODE_ENV ?? "production", TZ: process.env.TZ ?? "UTC", }; const child = fork(options.entrypointPath, [], { stdio: ["pipe", "pipe", "pipe", "ipc"], execArgv: options.execArgv ?? [], env: workerEnv, // Don't let the child keep the parent alive detached: false, }); return child; } function attachStdioHandlers(child: ChildProcess): void { // Read NDJSON from stdout if (child.stdout) { readline = createInterface({ input: child.stdout }); readline.on("line", handleLine); } // Capture stderr for logging if (child.stderr) { stderrReadline = createInterface({ input: child.stderr }); stderrReadline.on("line", (line: string) => { stderrExcerpt = appendStderrExcerpt(stderrExcerpt, line); log.warn({ stream: "stderr" }, `[plugin stderr] ${line}`); }); } // Handle process exit child.on("exit", (code, signal) => { handleProcessExit(code, signal); }); // Handle process errors (e.g. spawn failure) child.on("error", (err) => { log.error({ err: err.message }, "worker process error"); if (emitter.listenerCount("error") > 0) { emitter.emit("error", { pluginId, error: err }); } if (status === "starting") { setStatus("crashed"); rejectAllPending( new Error(formatWorkerFailureMessage( `Worker process failed to start: ${err.message}`, stderrExcerpt, )), ); } }); } function handleProcessExit( code: number | null, signal: NodeJS.Signals | null, ): void { const wasIntentional = intentionalStop; // Clean up readline interfaces if (readline) { readline.close(); readline = null; } if (stderrReadline) { stderrReadline.close(); stderrReadline = null; } childProcess = null; startedAt = null; // Reject all pending requests rejectAllPending( new Error(formatWorkerFailureMessage( `Worker process exited (code=${code}, signal=${signal})`, stderrExcerpt, )), ); // Close the one login pseudo-terminal route with a fixed non-secret exit and // clear the route one time. The pending pseudo-terminal calls // already rejected through `rejectAllPending`. closeSetupTokenPtyRouteOnWorkerExit(); // Close the one duplex channel route the same way. The pending channel calls // already rejected through `rejectAllPending`. closeDuplexChannelRouteOnWorkerExit(); // Emit synthetic close for any orphaned stream channels so SSE clients // are notified instead of hanging indefinitely. if (openStreamChannels.size > 0 && options.onStreamNotification) { for (const [channel, companyId] of openStreamChannels) { try { options.onStreamNotification("streams.close", { channel, companyId }); } catch { // Best-effort cleanup — don't let it interfere with exit handling } } openStreamChannels.clear(); } emitter.emit("exit", { pluginId, code, signal }); if (wasIntentional) { // Graceful stop — status is already "stopping" or will be set to "stopped" setStatus("stopped"); log.info({ code, signal }, "worker process stopped"); return; } // Unexpected exit — crash recovery totalCrashes++; const now = Date.now(); // Reset consecutive crash counter if enough time passed if (lastCrashAt !== null && now - lastCrashAt > CRASH_WINDOW_MS) { consecutiveCrashes = 0; } consecutiveCrashes++; lastCrashAt = now; log.error( { code, signal, consecutiveCrashes, totalCrashes }, "worker process crashed", ); const willRestart = autoRestart && consecutiveCrashes <= MAX_CONSECUTIVE_CRASHES; setStatus("crashed"); emitter.emit("crash", { pluginId, code, signal, willRestart }); if (willRestart) { scheduleRestart(); } else { log.error( { consecutiveCrashes, maxCrashes: MAX_CONSECUTIVE_CRASHES }, "max consecutive crashes reached, not restarting", ); } } function rejectAllPending(error: Error): void { for (const [id, pending] of pendingRequests) { clearTimeout(pending.timer); pending.resolve( createErrorResponse( pending.id, PLUGIN_RPC_ERROR_CODES.WORKER_UNAVAILABLE, error.message, ) as JsonRpcResponse, ); } pendingRequests.clear(); for (const invocation of activeInvocations.values()) { if (invocation.timer) clearTimeout(invocation.timer); } activeInvocations.clear(); } // ----------------------------------------------------------------------- // Crash recovery with exponential backoff // ----------------------------------------------------------------------- function computeBackoffMs(): number { // Exponential backoff: MIN_BACKOFF * MULTIPLIER^(consecutiveCrashes - 1) const delay = MIN_BACKOFF_MS * Math.pow(BACKOFF_MULTIPLIER, consecutiveCrashes - 1); // Add jitter: ±25% const jitter = delay * 0.25 * (Math.random() * 2 - 1); return Math.min(Math.round(delay + jitter), MAX_BACKOFF_MS); } function scheduleRestart(): void { const delay = computeBackoffMs(); nextRestartAt = Date.now() + delay; setStatus("backoff"); log.info( { delayMs: delay, consecutiveCrashes }, "scheduling restart with backoff", ); backoffTimer = setTimeout(async () => { backoffTimer = null; nextRestartAt = null; try { await startInternal(); } catch (err) { log.error( { err: err instanceof Error ? err.message : String(err) }, "restart after backoff failed", ); } }, delay); } function cancelPendingRestart(): void { if (backoffTimer !== null) { clearTimeout(backoffTimer); backoffTimer = null; nextRestartAt = null; } } // ----------------------------------------------------------------------- // Start / Stop // ----------------------------------------------------------------------- async function startInternal(): Promise { if (status === "running" || status === "starting") { throw new Error(`Worker for plugin "${pluginId}" is already ${status}`); } intentionalStop = false; setStatus("starting"); stderrExcerpt = ""; const child = spawnProcess(); childProcess = child; attachStdioHandlers(child); startedAt = Date.now(); // Send the initialize RPC call const initParams: InitializeParams = { manifest: options.manifest, config: options.config, instanceInfo: options.instanceInfo, apiVersion: options.apiVersion, databaseNamespace: options.databaseNamespace ?? null, }; try { const result = await callInternal( "initialize", initParams, INITIALIZE_TIMEOUT_MS, ) as { ok?: boolean; supportedMethods?: string[] } | undefined; if (!result || !result.ok) { throw new Error("Worker initialize returned ok=false"); } supportedMethods = result.supportedMethods ?? []; } catch (err) { // Initialize failed — kill the process and propagate const msg = err instanceof Error ? err.message : String(err); log.error({ err: msg }, "worker initialize failed"); await killProcess(); setStatus("crashed"); throw new Error(`Worker initialize failed for "${pluginId}": ${msg}`); } // Reset crash counter on successful start consecutiveCrashes = 0; setStatus("running"); emitter.emit("ready", { pluginId }); log.info({ pid: child.pid }, "worker process started and initialized"); } async function stopInternal(): Promise { cancelPendingRestart(); if (status === "stopped" || status === "stopping") { return; } intentionalStop = true; setStatus("stopping"); if (!childProcess) { setStatus("stopped"); return; } // Step 1: Send shutdown RPC and wait for the worker to exit gracefully. // We race the shutdown call against a timeout. The worker should process // the shutdown and exit on its own within the drain period. try { await Promise.race([ callInternal("shutdown", {} as Record, SHUTDOWN_DRAIN_MS), waitForExit(SHUTDOWN_DRAIN_MS), ]); } catch { // Shutdown call failed or timed out — proceed to kill log.warn("shutdown RPC failed or timed out, escalating to SIGTERM"); } // Give the process a brief moment to exit after the shutdown response if (childProcess) { await waitForExit(500); } // Check if process already exited if (!childProcess) { setStatus("stopped"); return; } // Step 2: Send SIGTERM and wait log.info("worker did not exit after shutdown RPC, sending SIGTERM"); await killWithSignal("SIGTERM", SIGTERM_GRACE_MS); if (!childProcess) { setStatus("stopped"); return; } // Step 3: Forcefully kill with SIGKILL log.warn("worker did not exit after SIGTERM, sending SIGKILL"); await killWithSignal("SIGKILL", 2_000); if (childProcess) { log.error("worker process still alive after SIGKILL — this should not happen"); } setStatus("stopped"); } /** * Wait for the child process to exit, up to `timeoutMs`. * Resolves immediately if the process is already gone. */ function waitForExit(timeoutMs: number): Promise { return new Promise((resolve) => { if (!childProcess) { resolve(); return; } let settled = false; const timer = setTimeout(() => { if (settled) return; settled = true; resolve(); }, timeoutMs); childProcess.once("exit", () => { if (settled) return; settled = true; clearTimeout(timer); resolve(); }); }); } function killWithSignal( signal: NodeJS.Signals, waitMs: number, ): Promise { return new Promise((resolve) => { if (!childProcess) { resolve(); return; } const timer = setTimeout(() => { resolve(); }, waitMs); childProcess.once("exit", () => { clearTimeout(timer); resolve(); }); try { childProcess.kill(signal); } catch { clearTimeout(timer); resolve(); } }); } async function killProcess(): Promise { if (!childProcess) return; intentionalStop = true; try { childProcess.kill("SIGKILL"); } catch { // Process may already be dead } // Wait briefly for exit event await new Promise((resolve) => { if (!childProcess) { resolve(); return; } const timer = setTimeout(() => { resolve(); }, 1_000); childProcess.once("exit", () => { clearTimeout(timer); resolve(); }); }); } // ----------------------------------------------------------------------- // RPC call implementation // ----------------------------------------------------------------------- function callInternal( method: M, params: HostToWorkerMethods[M][0], timeoutMs?: number, executeLogSink?: ExecuteLogSink, ): Promise { const rpcPromise = new Promise((resolve, reject) => { if (!childProcess?.stdin?.writable) { reject( new Error( `Cannot call "${method}" — worker for "${pluginId}" is not running`, ), ); return; } const id = nextRequestId++; const timeout = resolveRpcCallTimeoutMs(timeoutMs, rpcTimeoutMs); const invocationScope = deriveInvocationScope(method, params); const invocation = invocationScope ? registerInvocation(invocationScope) : null; // Register the host-owned execute route only for an execute call that // carries a log sink. The company id comes from the host-derived // invocation scope, never from the worker. This binds the sink to the // exact company for the life of the call. if (invocation && invocationScope && executeLogSink && method === "environmentExecute") { registerExecuteRoute(invocation.id, invocationScope.companyId, executeLogSink); } // Guard against double-settlement. When a process exits all pending // requests are rejected via rejectAllPending(), but the timeout timer // may still be running. Without this guard the timer's reject fires on // an already-settled promise, producing an unhandled rejection. let settled = false; const settle = (fn: (value: T) => void, value: T): void => { if (settled) return; settled = true; clearTimeout(timer); pendingRequests.delete(id); clearInvocation(invocation); clearExecuteRoute(invocation?.id); fn(value); }; const timer = setTimeout(() => { settle( reject, new JsonRpcCallError({ code: PLUGIN_RPC_ERROR_CODES.TIMEOUT, message: `RPC call "${method}" timed out after ${timeout}ms`, }), ); }, timeout); const pending: PendingRequest = { id, method, resolve: (response: JsonRpcResponse) => { if (isJsonRpcSuccessResponse(response)) { settle(resolve, response.result as HostToWorkerMethods[M][1]); } else if ("error" in response && response.error) { settle(reject, new JsonRpcCallError(response.error)); } else { settle(reject, new Error(`Unexpected response format for "${method}"`)); } }, timer, sentAt: Date.now(), invocationId: invocation?.id, }; pendingRequests.set(id, pending); try { const request = { ...createRequest(method, params, id), ...(invocation ? { paperclipInvocation: invocation } : {}), }; sendMessage(request); } catch (err) { clearTimeout(timer); pendingRequests.delete(id); clearInvocation(invocation); clearExecuteRoute(invocation?.id); reject( new Error( `Failed to send "${method}" to worker: ${ err instanceof Error ? err.message : String(err) }`, ), ); } }); // Some call sites hand these promises across async boundaries before // attaching their own handlers. Mark the promise as handled here so a // worker-side JSON-RPC error can fail the caller without killing the host // process via an unhandled rejection. void rpcPromise.catch(() => undefined); return rpcPromise; } // ----------------------------------------------------------------------- // Public API // ----------------------------------------------------------------------- const handle: PluginWorkerHandle = { get pluginId() { return pluginId; }, get status() { return status; }, get supportedMethods() { return supportedMethods; }, async start() { await startInternal(); }, async stop() { await stopInternal(); }, async restart() { await stopInternal(); await startInternal(); }, call( method: M, params: HostToWorkerMethods[M][0], timeoutMs?: number, executeLogSink?: ExecuteLogSink, ): Promise { if (status !== "running" && status !== "starting") { return Promise.reject( new Error( `Cannot call "${method}" — worker for "${pluginId}" is ${status}`, ), ); } return callInternal(method, params, timeoutMs, executeLogSink); }, openSetupTokenPtySession(input: SetupTokenPtyOpenInput) { if (status !== "running" && status !== "starting") { return Promise.reject( new Error( `Cannot open a login pseudo-terminal — worker for "${pluginId}" is ${status}`, ), ); } return openSetupTokenPtySession(input); }, openDuplexChannel(input: DuplexChannelOpenInput) { if (status !== "running" && status !== "starting") { return Promise.reject( new Error( `Cannot open a duplex channel — worker for "${pluginId}" is ${status}`, ), ); } return openDuplexChannel(input); }, notify(method: string, params: unknown) { if (status !== "running") return; const invocationScope = deriveInvocationScope(method, params); // Notifications have no response to settle on, so the invocation scope // is GC'd by TTL. Call-path invocations are registered without a TTL and // cleared on settlement, so they survive arbitrarily long call timeouts. const invocation = invocationScope ? registerInvocation(invocationScope, MAX_RPC_TIMEOUT_MS) : null; try { sendMessage({ jsonrpc: JSONRPC_VERSION, method, params, ...(invocation ? { paperclipInvocation: invocation } : {}), }); } catch { clearInvocation(invocation); log.warn({ method }, "failed to send notification to worker"); } }, on( event: K, listener: (payload: WorkerHandleEvents[K]) => void, ) { emitter.on(event, listener); }, off( event: K, listener: (payload: WorkerHandleEvents[K]) => void, ) { emitter.off(event, listener); }, setProactiveCompanyScopes(companyIds: readonly string[]): void { proactiveCompanyScopes.clear(); for (const id of companyIds) { const trimmed = readNonEmptyString(id); if (trimmed) proactiveCompanyScopes.add(trimmed); } }, diagnostics(): WorkerDiagnostics { return { pluginId, status, pid: childProcess?.pid ?? null, uptime: startedAt !== null && status === "running" ? Date.now() - startedAt : null, consecutiveCrashes, totalCrashes, pendingRequests: pendingRequests.size, lastCrashAt, nextRestartAt, }; }, }; return handle; } // --------------------------------------------------------------------------- // Implementation: createPluginWorkerManager // --------------------------------------------------------------------------- /** * Options for creating a PluginWorkerManager. */ export interface PluginWorkerManagerOptions { /** * Optional callback invoked when a worker emits a lifecycle event * (crash, restart). Used by the server to publish global live events. */ onWorkerEvent?: (event: { type: "plugin.worker.crashed" | "plugin.worker.restarted"; pluginId: string; code?: number | null; signal?: string | null; willRestart?: boolean; }) => void; /** * The process-scoped aggregate ceiling for concurrent duplex channel routes, * across every worker in the process. The manager builds one shared slot * controller from it and injects it into every worker handle, so one tenant can * never exhaust the manager-wide resource. The manager validates it and falls * back to {@link DEFAULT_MAX_CONCURRENT_DUPLEX_ROUTES} for an absent or an * invalid value. It is not the per-agent `heartbeat.maxConcurrentRuns`, which * stays upstream admission only. */ maxConcurrentDuplexRoutes?: number | null; } /** * The default process-scoped aggregate ceiling for concurrent duplex channel * routes. It caps the manager-wide resource, not one agent's run budget. The host * reports an explicit route-busy outcome when the ceiling is full. */ export const DEFAULT_MAX_CONCURRENT_DUPLEX_ROUTES = 128; /** * Build one process-scoped aggregate route-slot controller. The controller holds a * strictly positive integer ceiling and a live count. `tryAcquire` reserves one * slot only when a slot is free, so the count never passes the ceiling. */ export function createDuplexRouteSlotController(maxRoutes?: number | null): DuplexRouteSlotController { const ceiling = typeof maxRoutes === "number" && Number.isInteger(maxRoutes) && maxRoutes > 0 ? maxRoutes : DEFAULT_MAX_CONCURRENT_DUPLEX_ROUTES; let active = 0; return { tryAcquire(): boolean { if (active >= ceiling) return false; active += 1; return true; }, release(): void { if (active > 0) active -= 1; }, }; } /** * Create a new PluginWorkerManager. * * The manager holds all plugin worker handles and provides a unified API for * starting, stopping, and communicating with plugin workers. * * @example * ```ts * const manager = createPluginWorkerManager(); * * const handle = await manager.startWorker("acme.linear", { * entrypointPath: "/path/to/worker.cjs", * manifest, * config: resolvedConfig, * instanceInfo: { instanceId: "inst-1", hostVersion: "1.0.0" }, * apiVersion: 1, * hostHandlers: { "config.get": async () => resolvedConfig, ... }, * }); * * // Send RPC call to the worker * const health = await manager.call("acme.linear", "health", {}); * * // Shutdown all workers on server exit * await manager.stopAll(); * ``` */ export function createPluginWorkerManager( managerOptions?: PluginWorkerManagerOptions, ): PluginWorkerManager { const log = logger.child({ service: "plugin-worker-manager" }); const workers = new Map(); /** Per-plugin startup locks to prevent concurrent spawn races. */ const startupLocks = new Map>(); // The one shared, process-scoped aggregate route-slot controller. The manager // injects it into every worker handle, so the duplex route ceiling counts every // concurrent route across the process, not one agent's setting. const duplexRouteSlots = createDuplexRouteSlotController( managerOptions?.maxConcurrentDuplexRoutes, ); return { async startWorker( pluginId: string, options: WorkerStartOptions, ): Promise { // Mutex: if a start is already in-flight for this plugin, wait for it const inFlight = startupLocks.get(pluginId); if (inFlight) { log.warn({ pluginId }, "concurrent startWorker call — waiting for in-flight start"); return inFlight; } const existing = workers.get(pluginId); if (existing && existing.status !== "stopped") { throw new Error( `Worker already registered for plugin "${pluginId}" (status: ${existing.status})`, ); } const handle = createPluginWorkerHandle(pluginId, { // Inject the shared process-scoped route-slot controller, unless the caller // already supplied one (a test may inject its own). duplexRouteSlots, ...options, }); workers.set(pluginId, handle); // Subscribe to crash/ready events for live event forwarding if (managerOptions?.onWorkerEvent) { const notify = managerOptions.onWorkerEvent; handle.on("crash", (payload) => { notify({ type: "plugin.worker.crashed", pluginId: payload.pluginId, code: payload.code, signal: payload.signal, willRestart: payload.willRestart, }); }); handle.on("ready", (payload) => { // Only emit restarted if this was a crash recovery (totalCrashes > 0) const diag = handle.diagnostics(); if (diag.totalCrashes > 0) { notify({ type: "plugin.worker.restarted", pluginId: payload.pluginId, }); } }); } log.info({ pluginId }, "starting plugin worker"); // Set the lock before awaiting start() to prevent concurrent spawns const startPromise = handle.start().then(() => handle).finally(() => { startupLocks.delete(pluginId); }); startupLocks.set(pluginId, startPromise); return startPromise; }, async stopWorker(pluginId: string): Promise { const handle = workers.get(pluginId); if (!handle) { log.warn({ pluginId }, "no worker registered for plugin, nothing to stop"); return; } log.info({ pluginId }, "stopping plugin worker"); await handle.stop(); workers.delete(pluginId); }, getWorker(pluginId: string): PluginWorkerHandle | undefined { return workers.get(pluginId); }, isRunning(pluginId: string): boolean { const handle = workers.get(pluginId); return handle?.status === "running"; }, setProactiveCompanyScopes(pluginId: string, companyIds: readonly string[]): void { workers.get(pluginId)?.setProactiveCompanyScopes(companyIds); }, async stopAll(): Promise { log.info({ count: workers.size }, "stopping all plugin workers"); const promises = Array.from(workers.values()).map(async (handle) => { try { await handle.stop(); } catch (err) { log.error( { pluginId: handle.pluginId, err: err instanceof Error ? err.message : String(err), }, "error stopping worker during shutdown", ); } }); await Promise.all(promises); workers.clear(); }, diagnostics(): WorkerDiagnostics[] { return Array.from(workers.values()).map((h) => h.diagnostics()); }, call( pluginId: string, method: M, params: HostToWorkerMethods[M][0], timeoutMs?: number, executeLogSink?: ExecuteLogSink, ): Promise { const handle = workers.get(pluginId); if (!handle) { return Promise.reject( new Error(`No worker registered for plugin "${pluginId}"`), ); } return handle.call(method, params, timeoutMs, executeLogSink); }, openSetupTokenPtySession(pluginId: string, input: SetupTokenPtyOpenInput) { const handle = workers.get(pluginId); if (!handle) { return Promise.reject( new Error(`No worker registered for plugin "${pluginId}"`), ); } return handle.openSetupTokenPtySession(input); }, }; }