feat(runner): add bounded Codex app-server transport (#12362)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The first production runner provider is Codex > - Codex app-server communicates with JSON-RPC over a child process > - The process boundary needs explicit resource, environment, and diagnostic limits > - This pull request adds only the bounded transport layer > - A later pull request will add the Codex event-mapping driver > - The benefit is a small reviewable process boundary before provider behavior is added ## Linked Issues or Issue Description **Subsystem affected** `packages/paperclip-runner` Codex transport. **Problem or motivation** The runner needs a safe way to start and communicate with Codex app-server. Unbounded lines, queued notifications, pending requests, and inherited credentials would make that boundary unsafe. **Proposed solution** Add a JSON-RPC process transport with explicit byte and count limits, sanitized environment construction, process-group metadata, server-request handling, redacted diagnostics, and deterministic closure. **Alternatives considered** Adding transport and the full Codex driver in one change would create a much larger review surface. This pull request isolates the child-process boundary. **Roadmap alignment** This supports the Codex-first experimental runner. It does not enable the runner adapter. ## What Changed - Added bounded JSON-RPC request and notification queues. - Added malformed-message and oversized-line fail-closed behavior. - Added sanitized Codex environment construction. - Added process-group tracking and deterministic shutdown. - Added redacted diagnostics and server-request replies. ## Verification - `pnpm --filter @paperclipai/paperclip-runner test:typescript` - `pnpm -r typecheck` - `pnpm build` - The focused transport test has 6 passing cases. ## Risks The main risk is child-process cleanup. Tests cover closure, pending request rejection, queue bounds, process metadata, and malformed input. ## Model Used OpenAI Codex with GPT-5.6 and repository tool use. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
Dotta
GitHub
parent
ea86e780af
commit
83243d4b5d
|
|
@ -0,0 +1,248 @@
|
|||
import { ChildProcess } from "node:child_process";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { ProcessCodexAppServerTransport, redactCodexDiagnostic } from "./app-server-transport.js";
|
||||
|
||||
function nodeTransport(
|
||||
source: string,
|
||||
options: ConstructorParameters<typeof ProcessCodexAppServerTransport>[0] = {},
|
||||
) {
|
||||
return new ProcessCodexAppServerTransport({
|
||||
...options,
|
||||
command: process.execPath,
|
||||
args: ["-e", source],
|
||||
environment: { PATH: process.env.PATH },
|
||||
});
|
||||
}
|
||||
|
||||
describe("Codex app-server transport limits", () => {
|
||||
it("redacts real Basic credentials without corrupting ordinary question copy", () => {
|
||||
expect(redactCodexDiagnostic("Authorization: Basic dXNlcjpwYXNz"))
|
||||
.toBe("Authorization: Basic [REDACTED]");
|
||||
expect(redactCodexDiagnostic("Basic API foundation"))
|
||||
.toBe("Basic API foundation");
|
||||
});
|
||||
|
||||
it("reports restart-safe process-group ownership", async () => {
|
||||
const transport = nodeTransport("process.stdin.resume()", { processGroup: true });
|
||||
const info = transport.processInfo();
|
||||
|
||||
expect(info.pid).toBeGreaterThan(0);
|
||||
expect(info.processGroupId).toBe(process.platform === "win32" ? null : info.pid);
|
||||
expect(new Date(info.startedAt).toISOString()).toBe(info.startedAt);
|
||||
|
||||
await transport.close();
|
||||
});
|
||||
|
||||
it("installs cleanup handlers before synchronously reporting process ownership", async () => {
|
||||
const originalSpawn = ChildProcess.prototype.spawn;
|
||||
let spawnedProcess: ChildProcess | undefined;
|
||||
const spawnSpy = vi
|
||||
.spyOn(ChildProcess.prototype, "spawn")
|
||||
.mockImplementation(function (this: ChildProcess, ...args) {
|
||||
spawnedProcess = this;
|
||||
return originalSpawn.apply(this, args);
|
||||
});
|
||||
let transport: ProcessCodexAppServerTransport | undefined;
|
||||
let observedInitialProcess = false;
|
||||
|
||||
try {
|
||||
transport = nodeTransport("process.stdin.resume()", {
|
||||
onProcess: (info) => {
|
||||
if (info.exited) return;
|
||||
observedInitialProcess = true;
|
||||
expect(spawnedProcess?.pid).toBe(info.pid);
|
||||
expect(spawnedProcess?.listenerCount("error")).toBeGreaterThan(0);
|
||||
expect(spawnedProcess?.listenerCount("exit")).toBeGreaterThan(0);
|
||||
expect(
|
||||
spawnedProcess?.stdout?.listenerCount("data"),
|
||||
).toBeGreaterThan(0);
|
||||
expect(
|
||||
spawnedProcess?.stdout?.listenerCount("end"),
|
||||
).toBeGreaterThan(0);
|
||||
expect(
|
||||
spawnedProcess?.stderr?.listenerCount("data"),
|
||||
).toBeGreaterThan(0);
|
||||
expect(
|
||||
spawnedProcess?.stdin?.listenerCount("error"),
|
||||
).toBeGreaterThan(0);
|
||||
spawnedProcess?.emit(
|
||||
"error",
|
||||
new Error("synchronous process callback cleanup"),
|
||||
);
|
||||
},
|
||||
});
|
||||
expect(observedInitialProcess).toBe(true);
|
||||
await expect(
|
||||
transport.request("after-callback-cleanup", {}),
|
||||
).rejects.toThrow("codex app-server transport is closed");
|
||||
} finally {
|
||||
spawnSpy.mockRestore();
|
||||
await transport?.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects an oversized line before buffering the complete hostile payload", async () => {
|
||||
const diagnostics: string[] = [];
|
||||
const transport = nodeTransport(
|
||||
"setTimeout(() => process.stdout.write('x'.repeat(1048576)), 50); setInterval(() => {}, 1000)",
|
||||
{ maxLineBytes: 128, onDiagnostic: (message) => diagnostics.push(message) },
|
||||
);
|
||||
await expect(transport.notifications()[Symbol.asyncIterator]().next()).rejects.toThrow(
|
||||
"codex app-server line exceeded 128 bytes",
|
||||
);
|
||||
expect(diagnostics).toEqual(["codex app-server line exceeded 128 bytes"]);
|
||||
await transport.close();
|
||||
});
|
||||
|
||||
it("bounds pending requests", async () => {
|
||||
const transport = nodeTransport("process.stdin.resume()", { maxPendingRequests: 1 });
|
||||
const first = transport.request("first", {});
|
||||
await expect(transport.request("second", {})).rejects.toThrow(
|
||||
"codex app-server pending request limit 1 exceeded",
|
||||
);
|
||||
const firstRejected = expect(first).rejects.toThrow("codex app-server transport closed");
|
||||
await transport.close();
|
||||
await firstRejected;
|
||||
});
|
||||
|
||||
it("fails closed when queued notifications exceed their count bound", async () => {
|
||||
let transport: ProcessCodexAppServerTransport;
|
||||
const diagnostic = new Promise<string>((resolve) => {
|
||||
const lines = [1, 2].map((id) => JSON.stringify({
|
||||
method: "item/completed",
|
||||
params: { threadId: "thread-1", turnId: "turn-1", item: { id } },
|
||||
})).join("\n");
|
||||
transport = nodeTransport(
|
||||
`setTimeout(() => process.stdout.write(${JSON.stringify(`${lines}\n`)}), 50); setInterval(() => {}, 1000)`,
|
||||
{
|
||||
maxQueuedNotifications: 1,
|
||||
maxQueuedNotificationBytes: 1024,
|
||||
onDiagnostic: (message) => resolve(message),
|
||||
},
|
||||
);
|
||||
});
|
||||
await expect(diagnostic).resolves.toBe("codex app-server notification queue limit exceeded");
|
||||
await transport!.close();
|
||||
});
|
||||
|
||||
it("rejects malformed JSON-RPC messages", async () => {
|
||||
const transport = nodeTransport(
|
||||
"setTimeout(() => process.stdout.write('{not-json}\\n'), 50); setInterval(() => {}, 1000)",
|
||||
);
|
||||
await expect(transport.request("pending", {})).rejects.toThrow(
|
||||
"codex app-server emitted malformed JSON",
|
||||
);
|
||||
await transport.close();
|
||||
});
|
||||
|
||||
it("fails pending and future requests when stdout closes cleanly", async () => {
|
||||
const transport = nodeTransport(
|
||||
"process.stdin.resume(); process.stdout.end(); setInterval(() => {}, 1000)",
|
||||
);
|
||||
const pending = transport.request("pending", {});
|
||||
await expect(pending).rejects.toThrow(
|
||||
"codex app-server stdout ended before transport closure",
|
||||
);
|
||||
await expect(transport.request("after-stdout-end", {})).rejects.toThrow(
|
||||
"codex app-server transport is closed",
|
||||
);
|
||||
await transport.close();
|
||||
});
|
||||
|
||||
it("fails closed when stdout is destroyed without a usable response channel", async () => {
|
||||
const transport = nodeTransport(
|
||||
'process.stdin.resume(); require("node:fs").closeSync(1); setInterval(() => {}, 1000)',
|
||||
);
|
||||
await expect(transport.request("pending", {})).rejects.toThrow(/stdout (ended|closed)/u);
|
||||
await expect(transport.request("after-stdout-close", {})).rejects.toThrow(
|
||||
"codex app-server transport is closed",
|
||||
);
|
||||
await transport.close();
|
||||
});
|
||||
|
||||
it("fails closed when outbound buffering exceeds its bound", async () => {
|
||||
const diagnostics: string[] = [];
|
||||
const transport = nodeTransport("process.stdin.pause(); setInterval(() => {}, 1000)", {
|
||||
maxLineBytes: 1_024,
|
||||
maxBufferedOutputBytes: 64,
|
||||
onDiagnostic: (message) => diagnostics.push(message),
|
||||
});
|
||||
|
||||
expect(() => transport.notify("large", { value: "x".repeat(100) })).toThrow(
|
||||
"outbound codex JSON-RPC buffer exceeded 64 bytes",
|
||||
);
|
||||
expect(diagnostics).toContain(
|
||||
"outbound codex JSON-RPC buffer exceeded 64 bytes",
|
||||
);
|
||||
await expect(transport.request("after-close", {})).rejects.toThrow(
|
||||
"codex app-server transport is closed",
|
||||
);
|
||||
await transport.close();
|
||||
});
|
||||
|
||||
it("routes an oversized server response through deterministic closure", async () => {
|
||||
let transport: ProcessCodexAppServerTransport | undefined;
|
||||
const diagnostic = new Promise<string>((resolve) => {
|
||||
transport = nodeTransport(
|
||||
`setTimeout(() => process.stdout.write(JSON.stringify({ id: "server-1", method: "tool/call", params: {} }) + "\\n"), 20); setInterval(() => {}, 1000)`,
|
||||
{
|
||||
maxLineBytes: 128,
|
||||
onDiagnostic: (message) => resolve(message),
|
||||
},
|
||||
);
|
||||
transport.setServerRequestHandler(async () => ({ value: "x".repeat(256) }));
|
||||
});
|
||||
|
||||
await expect(diagnostic).resolves.toBe(
|
||||
"outbound codex JSON-RPC line exceeded 128 bytes",
|
||||
);
|
||||
await transport?.close();
|
||||
});
|
||||
|
||||
it("turns a synchronous server handler throw into a JSON-RPC error", async () => {
|
||||
const transport = nodeTransport(`
|
||||
process.stdin.setEncoding("utf8");
|
||||
process.stdin.once("data", (chunk) => {
|
||||
const response = JSON.parse(chunk.trim());
|
||||
process.stdout.write(JSON.stringify({ method: "observed", params: response }) + "\\n");
|
||||
});
|
||||
setTimeout(() => process.stdout.write(JSON.stringify({ id: "server-1", method: "tool/call", params: {} }) + "\\n"), 20);
|
||||
setInterval(() => {}, 1000);
|
||||
`);
|
||||
transport.setServerRequestHandler((): Promise<Record<string, unknown>> => {
|
||||
throw new Error("synchronous handler failure");
|
||||
});
|
||||
|
||||
await expect(
|
||||
transport.notifications()[Symbol.asyncIterator]().next(),
|
||||
).resolves.toMatchObject({
|
||||
done: false,
|
||||
value: {
|
||||
method: "observed",
|
||||
params: {
|
||||
id: "server-1",
|
||||
error: { code: -32_000, message: "Error: synchronous handler failure" },
|
||||
},
|
||||
},
|
||||
});
|
||||
await transport.close();
|
||||
});
|
||||
|
||||
it("routes an asynchronous child-stdin failure through deterministic closure", async () => {
|
||||
const diagnostics: string[] = [];
|
||||
const transport = nodeTransport(
|
||||
`require("node:fs").closeSync(0); process.stdout.write(JSON.stringify({ method: "ready", params: {} }) + "\\n"); setInterval(() => {}, 1000)`,
|
||||
{ onDiagnostic: (message) => diagnostics.push(message) },
|
||||
);
|
||||
const notifications = transport.notifications()[Symbol.asyncIterator]();
|
||||
await expect(notifications.next()).resolves.toMatchObject({
|
||||
done: false,
|
||||
value: { method: "ready" },
|
||||
});
|
||||
|
||||
await expect(transport.request("after-stdin-close", {})).rejects.toThrow();
|
||||
expect(diagnostics.some((message) => /EPIPE|closed|write/u.test(message))).toBe(true);
|
||||
await transport.close();
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,781 @@
|
|||
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import type { HarnessRuntimeRequestResolution } from "../../contracts/harness-driver.js";
|
||||
|
||||
export interface CodexRpcNotification {
|
||||
method: string;
|
||||
params: Record<string, unknown>;
|
||||
/** Internal-only correlation retained outside provider params and canonical PRP payloads. */
|
||||
paperclipTrace?: {
|
||||
sourceEventId: string;
|
||||
sourceEventType: string;
|
||||
};
|
||||
}
|
||||
|
||||
export interface CodexTraceInterpretation {
|
||||
sourceEventId: string;
|
||||
sourceEventType: string;
|
||||
providerMethod: string;
|
||||
disposition: "mapped" | "ignored" | "rejected";
|
||||
emittedEventIds: string[];
|
||||
reason: string;
|
||||
}
|
||||
|
||||
export interface CodexRpcServerRequest extends CodexRpcNotification {
|
||||
id: string | number;
|
||||
}
|
||||
|
||||
export type CodexServerRequestHandler = (
|
||||
request: CodexRpcServerRequest,
|
||||
) => Promise<Record<string, unknown>>;
|
||||
|
||||
export interface CodexAppServerTransport {
|
||||
request(
|
||||
method: string,
|
||||
params: Record<string, unknown>,
|
||||
): Promise<Record<string, unknown>>;
|
||||
notify(method: string, params?: Record<string, unknown>): void;
|
||||
notifications(): AsyncIterable<CodexRpcNotification>;
|
||||
setServerRequestHandler(handler: CodexServerRequestHandler): void;
|
||||
/**
|
||||
* Optional provider-neutral resolution path used when a transport has
|
||||
* already normalized a native server request behind another PRP boundary.
|
||||
*/
|
||||
resolveRuntimeRequest?(input: {
|
||||
requestId: string;
|
||||
turnId: string;
|
||||
resolution: HarnessRuntimeRequestResolution;
|
||||
}): Promise<void>;
|
||||
close(): Promise<void>;
|
||||
processInfo?(): CodexTransportProcessInfo;
|
||||
attachRun?(input: {
|
||||
runId: string;
|
||||
turnId: string;
|
||||
itemId: string;
|
||||
}): Promise<void>;
|
||||
/** Records post-rehydration driver mapping without affecting run authority. */
|
||||
recordTraceInterpretation?(input: CodexTraceInterpretation): void;
|
||||
}
|
||||
|
||||
export interface CodexTransportProcessInfo {
|
||||
pid: number | null;
|
||||
processGroupId: number | null;
|
||||
startedAt: string;
|
||||
exited: boolean;
|
||||
exitCode: number | null;
|
||||
signal: NodeJS.Signals | null;
|
||||
}
|
||||
|
||||
class BoundedAsyncQueue<T> implements AsyncIterable<T> {
|
||||
#values: Array<{ value: T; bytes: number }> = [];
|
||||
#waiters: Array<{
|
||||
resolve: (value: IteratorResult<T>) => void;
|
||||
reject: (error: Error) => void;
|
||||
}> = [];
|
||||
#queuedBytes = 0;
|
||||
#closed = false;
|
||||
#error: Error | null = null;
|
||||
|
||||
constructor(
|
||||
private readonly maxValues: number,
|
||||
private readonly maxBytes: number,
|
||||
) {}
|
||||
|
||||
push(value: T, bytes: number): boolean {
|
||||
if (this.#closed) return false;
|
||||
const waiter = this.#waiters.shift();
|
||||
if (waiter !== undefined) {
|
||||
waiter.resolve({ value, done: false });
|
||||
return true;
|
||||
}
|
||||
if (
|
||||
this.#values.length >= this.maxValues ||
|
||||
this.#queuedBytes + bytes > this.maxBytes
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
this.#values.push({ value, bytes });
|
||||
this.#queuedBytes += bytes;
|
||||
return true;
|
||||
}
|
||||
|
||||
close(error?: Error): void {
|
||||
if (this.#closed) return;
|
||||
this.#closed = true;
|
||||
this.#values = [];
|
||||
this.#queuedBytes = 0;
|
||||
this.#error = error ?? null;
|
||||
for (const waiter of this.#waiters.splice(0)) {
|
||||
if (error === undefined) waiter.resolve({ value: undefined, done: true });
|
||||
else waiter.reject(error);
|
||||
}
|
||||
}
|
||||
|
||||
[Symbol.asyncIterator](): AsyncIterator<T> {
|
||||
return {
|
||||
next: async () => {
|
||||
const queued = this.#values.shift();
|
||||
if (queued !== undefined) {
|
||||
this.#queuedBytes -= queued.bytes;
|
||||
return { value: queued.value, done: false };
|
||||
}
|
||||
if (this.#error !== null) return Promise.reject(this.#error);
|
||||
if (this.#closed) return { value: undefined, done: true };
|
||||
return new Promise<IteratorResult<T>>((resolve, reject) => {
|
||||
this.#waiters.push({ resolve, reject });
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
class BoundedLineDecoder {
|
||||
#buffer = Buffer.alloc(0);
|
||||
#closed = false;
|
||||
|
||||
constructor(
|
||||
private readonly maxLineBytes: number,
|
||||
private readonly onLine: (line: string, bytes: number) => void,
|
||||
private readonly onError: (error: Error) => void,
|
||||
) {}
|
||||
|
||||
write(chunk: Buffer | string): void {
|
||||
if (this.#closed) return;
|
||||
const incoming = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
|
||||
let offset = 0;
|
||||
while (offset < incoming.length) {
|
||||
const newline = incoming.indexOf(0x0a, offset);
|
||||
const end = newline < 0 ? incoming.length : newline;
|
||||
const segment = incoming.subarray(offset, end);
|
||||
if (this.#buffer.length + segment.length > this.maxLineBytes) {
|
||||
this.#fail(
|
||||
new Error(
|
||||
`codex app-server line exceeded ${this.maxLineBytes} bytes`,
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (segment.length > 0)
|
||||
this.#buffer = Buffer.concat([this.#buffer, segment]);
|
||||
if (newline < 0) return;
|
||||
const line = this.#buffer;
|
||||
this.#buffer = Buffer.alloc(0);
|
||||
this.onLine(line.toString("utf8"), line.length);
|
||||
if (this.#closed) return;
|
||||
offset = newline + 1;
|
||||
}
|
||||
}
|
||||
|
||||
end(): void {
|
||||
if (this.#closed) return;
|
||||
if (this.#buffer.length > 0) {
|
||||
this.#fail(
|
||||
new Error("codex app-server ended with an unterminated JSON-RPC line"),
|
||||
);
|
||||
}
|
||||
this.#closed = true;
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.#closed = true;
|
||||
this.#buffer = Buffer.alloc(0);
|
||||
}
|
||||
|
||||
#fail(error: Error): void {
|
||||
this.close();
|
||||
this.onError(error);
|
||||
}
|
||||
}
|
||||
|
||||
const SAFE_ENVIRONMENT_KEYS = [
|
||||
"ALL_PROXY",
|
||||
"CODEX_HOME",
|
||||
"HOME",
|
||||
"HTTP_PROXY",
|
||||
"HTTPS_PROXY",
|
||||
"LANG",
|
||||
"LC_ALL",
|
||||
"NO_PROXY",
|
||||
"NODE_EXTRA_CA_CERTS",
|
||||
"PATH",
|
||||
"PATHEXT",
|
||||
"SSL_CERT_FILE",
|
||||
"SystemRoot",
|
||||
"TEMP",
|
||||
"TMP",
|
||||
"TMPDIR",
|
||||
"WINDIR",
|
||||
] as const;
|
||||
|
||||
/**
|
||||
* Environment for the trusted app-server process itself. HOME and CODEX_HOME
|
||||
* are retained so Codex can authenticate; model-issued commands receive a
|
||||
* separate empty-by-default environment and filesystem permission profile.
|
||||
*/
|
||||
export function createSanitizedCodexEnvironment(
|
||||
source: NodeJS.ProcessEnv = process.env,
|
||||
): NodeJS.ProcessEnv {
|
||||
const environment: NodeJS.ProcessEnv = {};
|
||||
for (const key of SAFE_ENVIRONMENT_KEYS) {
|
||||
const value = source[key];
|
||||
if (value === undefined) continue;
|
||||
if (key.includes("PROXY") && proxyContainsCredentials(value)) continue;
|
||||
environment[key] = value;
|
||||
}
|
||||
return environment;
|
||||
}
|
||||
|
||||
export function sanitizedEnvironmentKeys(
|
||||
source: NodeJS.ProcessEnv = process.env,
|
||||
): string[] {
|
||||
return Object.keys(createSanitizedCodexEnvironment(source)).sort();
|
||||
}
|
||||
|
||||
export function redactCodexDiagnostic(message: string): string {
|
||||
return message
|
||||
.replaceAll(/\u001b\[[0-?]*[ -/]*[@-~]/g, "")
|
||||
.replace(/Bearer\s+[A-Za-z0-9._~+\/-]+/gi, "Bearer [REDACTED]")
|
||||
.replace(/Basic\s+([A-Za-z0-9+/=]+)/gi, (match, encoded: string) => {
|
||||
try {
|
||||
// Only redact an actual RFC 7617 credential. Treating every word after
|
||||
// “Basic” as base64 corrupted ordinary question copy such as
|
||||
// “Basic API” before it entered the Paperclip protocol.
|
||||
const decoded = Buffer.from(encoded, "base64").toString("utf8");
|
||||
return decoded.includes(":") ? "Basic [REDACTED]" : match;
|
||||
} catch {
|
||||
return match;
|
||||
}
|
||||
})
|
||||
.replace(/([a-z][a-z0-9+.-]*:\/\/)[^\s/@:]+:[^\s/@]+@/gi, "$1[REDACTED]@")
|
||||
.replace(
|
||||
/([?&](?:api[_-]?key|token|secret|password)=)[^&#\s]+/gi,
|
||||
"$1[REDACTED]",
|
||||
)
|
||||
.replace(
|
||||
/(["'](?:api[_-]?key|token|secret|password|authorization)["']\s*:\s*["'])[^"']+/gi,
|
||||
"$1[REDACTED]",
|
||||
)
|
||||
.replace(
|
||||
/(api[_-]?key|token|secret|password)\s*[=:]\s*[^\s,;]+/gi,
|
||||
"$1=[REDACTED]",
|
||||
)
|
||||
.replace(
|
||||
/(PAPERCLIP_API_KEY|OPENAI_API_KEY|OPENROUTER_API_KEY)=[^\s]+/g,
|
||||
"$1=[REDACTED]",
|
||||
);
|
||||
}
|
||||
|
||||
function proxyContainsCredentials(value: string): boolean {
|
||||
try {
|
||||
const url = new URL(value);
|
||||
return url.username.length > 0 || url.password.length > 0;
|
||||
} catch {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
export const CODEX_METHOD_NOT_FOUND = -32_601;
|
||||
export const CODEX_INVALID_REQUEST = -32_600;
|
||||
|
||||
/**
|
||||
* Codes where the provider refused the call itself rather than failing while
|
||||
* serving it: the method is absent (-32601), or the request was rejected
|
||||
* outright, which is how the Codex app-server reports a feature it has
|
||||
* switched off (-32600 `goals feature is disabled`). Server-error codes
|
||||
* (-32603 and the -32000..-32099 implementation range) and transport failures
|
||||
* describe one attempt, not what the build supports.
|
||||
*/
|
||||
const METHOD_UNAVAILABLE_CODES: readonly number[] = [
|
||||
CODEX_METHOD_NOT_FOUND,
|
||||
CODEX_INVALID_REQUEST,
|
||||
];
|
||||
|
||||
/**
|
||||
* A provider error response, carrying its JSON-RPC code. Callers need the code
|
||||
* to tell "this provider build will not serve the method" apart from "the call
|
||||
* failed this time", which are different facts about a capability.
|
||||
*/
|
||||
export class CodexRpcError extends Error {
|
||||
readonly code: number | null;
|
||||
|
||||
constructor(message: string, code: number | null) {
|
||||
super(message);
|
||||
this.name = "CodexRpcError";
|
||||
this.code = code;
|
||||
}
|
||||
|
||||
get methodUnavailable(): boolean {
|
||||
return this.code !== null && METHOD_UNAVAILABLE_CODES.includes(this.code);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* True only when the provider answered and its answer denies the method. A
|
||||
* plain `Error` here means the call never got an answer — a transport or
|
||||
* protocol failure — which must not be read as a missing capability.
|
||||
*/
|
||||
export function isCodexMethodUnavailable(error: unknown): boolean {
|
||||
return error instanceof CodexRpcError && error.methodUnavailable;
|
||||
}
|
||||
|
||||
function rpcErrorCode(value: unknown): number | null {
|
||||
if (typeof value !== "object" || value === null) return null;
|
||||
const code = (value as { code?: unknown }).code;
|
||||
return typeof code === "number" && Number.isFinite(code) ? code : null;
|
||||
}
|
||||
|
||||
interface PendingRequest {
|
||||
resolve: (value: Record<string, unknown>) => void;
|
||||
reject: (error: Error) => void;
|
||||
}
|
||||
|
||||
export interface ProcessCodexTransportOptions {
|
||||
command?: string;
|
||||
args?: string[];
|
||||
environment?: NodeJS.ProcessEnv;
|
||||
onDiagnostic?: (message: string) => void;
|
||||
maxLineBytes?: number;
|
||||
maxPendingRequests?: number;
|
||||
maxQueuedNotifications?: number;
|
||||
maxQueuedNotificationBytes?: number;
|
||||
maxBufferedOutputBytes?: number;
|
||||
maxInflightServerRequests?: number;
|
||||
/** Starts a dedicated process group so runnerd and its Codex child are reaped together. */
|
||||
processGroup?: boolean;
|
||||
closeGraceMs?: number;
|
||||
onProcess?: (info: CodexTransportProcessInfo) => void;
|
||||
}
|
||||
|
||||
const DEFAULT_MAX_LINE_BYTES = 1024 * 1024;
|
||||
const DEFAULT_MAX_PENDING_REQUESTS = 64;
|
||||
const DEFAULT_MAX_QUEUED_NOTIFICATIONS = 256;
|
||||
const DEFAULT_MAX_QUEUED_NOTIFICATION_BYTES = 4 * 1024 * 1024;
|
||||
const DEFAULT_MAX_BUFFERED_OUTPUT_BYTES = 4 * 1024 * 1024;
|
||||
const DEFAULT_MAX_INFLIGHT_SERVER_REQUESTS = 32;
|
||||
const MAX_DIAGNOSTIC_LINE_BYTES = 16 * 1024;
|
||||
|
||||
/** Local stdio JSON-RPC transport. App-server never becomes a WAN endpoint. */
|
||||
export class ProcessCodexAppServerTransport implements CodexAppServerTransport {
|
||||
#process: ChildProcessWithoutNullStreams;
|
||||
#notifications: BoundedAsyncQueue<CodexRpcNotification>;
|
||||
#pending = new Map<number, PendingRequest>();
|
||||
#nextId = 1;
|
||||
#serverRequestHandler: CodexServerRequestHandler = async () => ({
|
||||
success: false,
|
||||
contentItems: [{ type: "input_text", text: "Unsupported client request." }],
|
||||
});
|
||||
#inflightServerRequests = 0;
|
||||
#closed = false;
|
||||
#exited = false;
|
||||
#exitCode: number | null = null;
|
||||
#exitSignal: NodeJS.Signals | null = null;
|
||||
#resolveExit!: () => void;
|
||||
readonly #exitPromise: Promise<void>;
|
||||
readonly #processGroup: boolean;
|
||||
readonly #startedAt: string;
|
||||
readonly #closeGraceMs: number;
|
||||
readonly #onProcess?: (info: CodexTransportProcessInfo) => void;
|
||||
#onDiagnostic?: (message: string) => void;
|
||||
readonly #maxLineBytes: number;
|
||||
readonly #maxPendingRequests: number;
|
||||
readonly #maxInflightServerRequests: number;
|
||||
readonly #maxBufferedOutputBytes: number;
|
||||
readonly #stdoutDecoder: BoundedLineDecoder;
|
||||
readonly #stderrDecoder: BoundedLineDecoder;
|
||||
|
||||
constructor(options: ProcessCodexTransportOptions = {}) {
|
||||
this.#onDiagnostic = options.onDiagnostic;
|
||||
this.#maxLineBytes = positiveLimit(
|
||||
options.maxLineBytes,
|
||||
DEFAULT_MAX_LINE_BYTES,
|
||||
);
|
||||
this.#maxPendingRequests = positiveLimit(
|
||||
options.maxPendingRequests,
|
||||
DEFAULT_MAX_PENDING_REQUESTS,
|
||||
);
|
||||
this.#maxInflightServerRequests = positiveLimit(
|
||||
options.maxInflightServerRequests,
|
||||
DEFAULT_MAX_INFLIGHT_SERVER_REQUESTS,
|
||||
);
|
||||
this.#maxBufferedOutputBytes = positiveLimit(
|
||||
options.maxBufferedOutputBytes,
|
||||
DEFAULT_MAX_BUFFERED_OUTPUT_BYTES,
|
||||
);
|
||||
this.#processGroup =
|
||||
options.processGroup === true && process.platform !== "win32";
|
||||
this.#startedAt = new Date().toISOString();
|
||||
this.#closeGraceMs = positiveLimit(options.closeGraceMs, 1_000);
|
||||
this.#onProcess = options.onProcess;
|
||||
this.#exitPromise = new Promise<void>((resolve) => {
|
||||
this.#resolveExit = resolve;
|
||||
});
|
||||
this.#notifications = new BoundedAsyncQueue(
|
||||
positiveLimit(
|
||||
options.maxQueuedNotifications,
|
||||
DEFAULT_MAX_QUEUED_NOTIFICATIONS,
|
||||
),
|
||||
positiveLimit(
|
||||
options.maxQueuedNotificationBytes,
|
||||
DEFAULT_MAX_QUEUED_NOTIFICATION_BYTES,
|
||||
),
|
||||
);
|
||||
this.#process = spawn(
|
||||
options.command ?? "codex",
|
||||
options.args ?? ["app-server"],
|
||||
{
|
||||
env: options.environment ?? createSanitizedCodexEnvironment(),
|
||||
stdio: "pipe",
|
||||
detached: this.#processGroup,
|
||||
},
|
||||
);
|
||||
this.#stdoutDecoder = new BoundedLineDecoder(
|
||||
this.#maxLineBytes,
|
||||
(line, bytes) => this.#onLine(line, bytes),
|
||||
(error) => this.#fatal(error),
|
||||
);
|
||||
this.#stderrDecoder = new BoundedLineDecoder(
|
||||
MAX_DIAGNOSTIC_LINE_BYTES,
|
||||
(line) => this.#onDiagnostic?.(redactCodexDiagnostic(line)),
|
||||
() =>
|
||||
this.#onDiagnostic?.(
|
||||
"codex app-server diagnostic line exceeded retention limit",
|
||||
),
|
||||
);
|
||||
this.#process.stdout.on("data", (chunk: Buffer) =>
|
||||
this.#stdoutDecoder.write(chunk),
|
||||
);
|
||||
this.#process.stdout.on("end", () => {
|
||||
this.#stdoutDecoder.end();
|
||||
this.#fatal(new Error("codex app-server stdout ended before transport closure"));
|
||||
});
|
||||
this.#process.stdout.on("error", (error) => this.#fatal(error));
|
||||
this.#process.stdout.on("close", () => {
|
||||
this.#fatal(
|
||||
new Error("codex app-server stdout closed before transport closure"),
|
||||
);
|
||||
});
|
||||
this.#process.stderr.on("data", (chunk: Buffer) =>
|
||||
this.#stderrDecoder.write(chunk),
|
||||
);
|
||||
this.#process.stderr.on("end", () => this.#stderrDecoder.end());
|
||||
this.#process.stdin.on("error", (error) => this.#fatal(error));
|
||||
this.#process.on("error", (error) => this.#fatal(error));
|
||||
this.#process.on("exit", (code, signal) => {
|
||||
this.#exited = true;
|
||||
this.#exitCode = code;
|
||||
this.#exitSignal = signal;
|
||||
this.#resolveExit();
|
||||
this.#onProcess?.(this.processInfo());
|
||||
if (!this.#closed) {
|
||||
this.#fatal(
|
||||
new Error(
|
||||
`codex app-server exited (code=${code ?? "null"}, signal=${signal ?? "none"})`,
|
||||
),
|
||||
);
|
||||
}
|
||||
});
|
||||
this.#onProcess?.(this.processInfo());
|
||||
}
|
||||
|
||||
request(
|
||||
method: string,
|
||||
params: Record<string, unknown>,
|
||||
): Promise<Record<string, unknown>> {
|
||||
if (this.#closed)
|
||||
return Promise.reject(new Error("codex app-server transport is closed"));
|
||||
if (this.#pending.size >= this.#maxPendingRequests) {
|
||||
return Promise.reject(
|
||||
new Error(
|
||||
`codex app-server pending request limit ${this.#maxPendingRequests} exceeded`,
|
||||
),
|
||||
);
|
||||
}
|
||||
const id = this.#nextId++;
|
||||
return new Promise((resolve, reject) => {
|
||||
this.#pending.set(id, { resolve, reject });
|
||||
try {
|
||||
this.#write({ method, id, params });
|
||||
} catch (error) {
|
||||
this.#pending.delete(id);
|
||||
reject(error);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
notify(method: string, params?: Record<string, unknown>): void {
|
||||
this.#write(params === undefined ? { method } : { method, params });
|
||||
}
|
||||
|
||||
notifications(): AsyncIterable<CodexRpcNotification> {
|
||||
return this.#notifications;
|
||||
}
|
||||
|
||||
setServerRequestHandler(handler: CodexServerRequestHandler): void {
|
||||
this.#serverRequestHandler = handler;
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
const firstClose = !this.#closed;
|
||||
if (firstClose) {
|
||||
this.#closed = true;
|
||||
this.#stdoutDecoder.close();
|
||||
this.#stderrDecoder.close();
|
||||
this.#notifications.close();
|
||||
this.#failAll(new Error("codex app-server transport closed"));
|
||||
}
|
||||
if (!this.#exited) {
|
||||
if (firstClose) this.#signal("SIGTERM");
|
||||
const exited = await Promise.race([
|
||||
this.#exitPromise.then(() => true),
|
||||
new Promise<false>((resolve) =>
|
||||
setTimeout(() => resolve(false), this.#closeGraceMs),
|
||||
),
|
||||
]);
|
||||
if (!exited && !this.#exited) {
|
||||
this.#signal("SIGKILL");
|
||||
await Promise.race([
|
||||
this.#exitPromise,
|
||||
new Promise<void>((resolve) =>
|
||||
setTimeout(resolve, this.#closeGraceMs),
|
||||
),
|
||||
]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
processInfo(): CodexTransportProcessInfo {
|
||||
return {
|
||||
pid: this.#process.pid ?? null,
|
||||
processGroupId: this.#processGroup ? (this.#process.pid ?? null) : null,
|
||||
startedAt: this.#startedAt,
|
||||
exited: this.#exited,
|
||||
exitCode: this.#exitCode,
|
||||
signal: this.#exitSignal,
|
||||
};
|
||||
}
|
||||
|
||||
#signal(signal: NodeJS.Signals): void {
|
||||
const pid = this.#process.pid;
|
||||
if (this.#processGroup && pid !== undefined) {
|
||||
try {
|
||||
process.kill(-pid, signal);
|
||||
return;
|
||||
} catch {
|
||||
// The group may already be gone; the direct child kill below is idempotent.
|
||||
}
|
||||
}
|
||||
this.#process.kill(signal);
|
||||
}
|
||||
|
||||
#write(message: unknown): void {
|
||||
if (this.#closed) throw new Error("codex app-server transport is closed");
|
||||
let serialized: string;
|
||||
try {
|
||||
serialized = JSON.stringify(message);
|
||||
} catch (error) {
|
||||
const failure = error instanceof Error ? error : new Error(String(error));
|
||||
this.#fatal(failure);
|
||||
throw failure;
|
||||
}
|
||||
const line = `${serialized}\n`;
|
||||
const lineBytes = Buffer.byteLength(line);
|
||||
if (lineBytes - 1 > this.#maxLineBytes) {
|
||||
const error = new Error(
|
||||
`outbound codex JSON-RPC line exceeded ${this.#maxLineBytes} bytes`,
|
||||
);
|
||||
this.#fatal(error);
|
||||
throw error;
|
||||
}
|
||||
if (
|
||||
this.#process.stdin.writableLength + lineBytes >
|
||||
this.#maxBufferedOutputBytes
|
||||
) {
|
||||
const error = new Error(
|
||||
`outbound codex JSON-RPC buffer exceeded ${this.#maxBufferedOutputBytes} bytes`,
|
||||
);
|
||||
this.#fatal(error);
|
||||
throw error;
|
||||
}
|
||||
try {
|
||||
this.#process.stdin.write(line);
|
||||
} catch (error) {
|
||||
const failure = error instanceof Error ? error : new Error(String(error));
|
||||
this.#fatal(failure);
|
||||
throw failure;
|
||||
}
|
||||
}
|
||||
|
||||
#onLine(line: string, bytes: number): void {
|
||||
let value: unknown;
|
||||
try {
|
||||
value = JSON.parse(line);
|
||||
} catch {
|
||||
this.#fatal(new Error("codex app-server emitted malformed JSON"));
|
||||
return;
|
||||
}
|
||||
if (!isRecord(value)) {
|
||||
this.#fatal(
|
||||
new Error("codex app-server emitted a non-object JSON-RPC message"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
const message = value;
|
||||
const hasResult = Object.hasOwn(message, "result");
|
||||
const hasError = Object.hasOwn(message, "error");
|
||||
if (typeof message.id === "number" && (hasResult || hasError)) {
|
||||
if (
|
||||
!Number.isSafeInteger(message.id) ||
|
||||
message.id <= 0 ||
|
||||
hasResult === hasError
|
||||
) {
|
||||
this.#fatal(
|
||||
new Error("codex app-server emitted a malformed JSON-RPC response"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
const pending = this.#pending.get(message.id);
|
||||
if (pending === undefined) {
|
||||
this.#fatal(
|
||||
new Error("codex app-server responded to an unknown request id"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
this.#pending.delete(message.id);
|
||||
if (hasError) {
|
||||
pending.reject(
|
||||
new CodexRpcError(
|
||||
redactCodexDiagnostic(boundedJson(message.error)),
|
||||
rpcErrorCode(message.error),
|
||||
),
|
||||
);
|
||||
} else if (isRecord(message.result)) {
|
||||
pending.resolve(message.result);
|
||||
} else {
|
||||
const error = new Error(
|
||||
"codex app-server response result was not an object",
|
||||
);
|
||||
pending.reject(error);
|
||||
this.#fatal(error);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (
|
||||
typeof message.method !== "string" ||
|
||||
message.method.length === 0 ||
|
||||
hasResult ||
|
||||
hasError
|
||||
) {
|
||||
this.#fatal(
|
||||
new Error("codex app-server emitted a malformed JSON-RPC message"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (message.params !== undefined && !isRecord(message.params)) {
|
||||
this.#fatal(
|
||||
new Error("codex app-server message params were not an object"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
const params = message.params ?? {};
|
||||
if (message.id !== undefined) {
|
||||
if (!isValidServerRequestId(message.id)) {
|
||||
this.#fatal(
|
||||
new Error("codex app-server request id had an invalid type"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
this.#handleServerRequest({
|
||||
id: message.id,
|
||||
method: message.method,
|
||||
params,
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (!this.#notifications.push({ method: message.method, params }, bytes)) {
|
||||
this.#fatal(
|
||||
new Error("codex app-server notification queue limit exceeded"),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#handleServerRequest(request: CodexRpcServerRequest): void {
|
||||
if (this.#inflightServerRequests >= this.#maxInflightServerRequests) {
|
||||
this.#writeServerResponse({
|
||||
id: request.id,
|
||||
error: {
|
||||
code: -32_001,
|
||||
message: "Client request queue limit exceeded.",
|
||||
},
|
||||
});
|
||||
return;
|
||||
}
|
||||
this.#inflightServerRequests += 1;
|
||||
void Promise.resolve()
|
||||
.then(() => this.#serverRequestHandler(request))
|
||||
.then(
|
||||
(result) => {
|
||||
this.#writeServerResponse({ id: request.id, result });
|
||||
},
|
||||
(error: unknown) => {
|
||||
this.#writeServerResponse({
|
||||
id: request.id,
|
||||
error: {
|
||||
code: -32_000,
|
||||
message: redactCodexDiagnostic(String(error)),
|
||||
},
|
||||
});
|
||||
},
|
||||
)
|
||||
.finally(() => {
|
||||
this.#inflightServerRequests -= 1;
|
||||
});
|
||||
}
|
||||
|
||||
#writeServerResponse(message: unknown): void {
|
||||
if (this.#closed) return;
|
||||
try {
|
||||
this.#write(message);
|
||||
} catch (error) {
|
||||
this.#fatal(error instanceof Error ? error : new Error(String(error)));
|
||||
}
|
||||
}
|
||||
|
||||
#fatal(error: Error): void {
|
||||
if (this.#closed) return;
|
||||
this.#closed = true;
|
||||
this.#stdoutDecoder.close();
|
||||
this.#stderrDecoder.close();
|
||||
this.#signal("SIGKILL");
|
||||
this.#notifications.close(error);
|
||||
this.#failAll(error);
|
||||
this.#onDiagnostic?.(redactCodexDiagnostic(error.message));
|
||||
}
|
||||
|
||||
#failAll(error: Error): void {
|
||||
for (const pending of this.#pending.values()) pending.reject(error);
|
||||
this.#pending.clear();
|
||||
}
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function positiveLimit(value: number | undefined, fallback: number): number {
|
||||
return Number.isSafeInteger(value) && (value ?? 0) > 0 ? value! : fallback;
|
||||
}
|
||||
|
||||
function isValidServerRequestId(value: unknown): value is string | number {
|
||||
return (
|
||||
(typeof value === "number" && Number.isSafeInteger(value)) ||
|
||||
(typeof value === "string" &&
|
||||
value.length > 0 &&
|
||||
Buffer.byteLength(value) <= 256)
|
||||
);
|
||||
}
|
||||
|
||||
function boundedJson(value: unknown): string {
|
||||
const serialized = JSON.stringify(value) ?? String(value);
|
||||
return serialized.length <= MAX_DIAGNOSTIC_LINE_BYTES
|
||||
? serialized
|
||||
: `${serialized.slice(0, MAX_DIAGNOSTIC_LINE_BYTES)}...[truncated]`;
|
||||
}
|
||||
Loading…
Reference in New Issue