395 lines
10 KiB
JavaScript
395 lines
10 KiB
JavaScript
import { randomUUID } from "node:crypto";
|
|
import { spawn } from "node:child_process";
|
|
import path from "node:path";
|
|
import { StringDecoder } from "node:string_decoder";
|
|
import { PI_RPC_COMMANDS } from "./pi-rpc-command-spec.js";
|
|
|
|
export const DEFAULT_COMMAND_TIMEOUT_MS = 30_000;
|
|
export const COMPACT_COMMAND_TIMEOUT_MS = 5 * 60_000;
|
|
export const MAX_PI_RPC_FRAME_BYTES = 1024 * 1024;
|
|
|
|
const supportedCommands = new Set(PI_RPC_COMMANDS);
|
|
|
|
export class PiRpcError extends Error {
|
|
constructor(code, message) {
|
|
super(message);
|
|
this.name = "PiRpcError";
|
|
this.code = code;
|
|
}
|
|
}
|
|
|
|
function isRecord(value) {
|
|
return value !== null && typeof value === "object" && !Array.isArray(value);
|
|
}
|
|
|
|
function assertAbsolutePath(value, field) {
|
|
if (typeof value !== "string" || !path.isAbsolute(value)) {
|
|
throw new TypeError(`${field} must be an absolute path`);
|
|
}
|
|
}
|
|
|
|
function emitSafely(callback, ...values) {
|
|
try {
|
|
callback(...values);
|
|
} catch {
|
|
// Adapter observers must not interrupt the RPC reader.
|
|
}
|
|
}
|
|
|
|
function normalizePiEvent(event) {
|
|
switch (event.type) {
|
|
case "agent_start":
|
|
return { type: "agent_state", data: { state: "streaming", event } };
|
|
case "agent_end":
|
|
// A low-level run can be followed by compaction, retry, or queued work.
|
|
// Only agent_settled is an authoritative idle transition.
|
|
return { type: "transcript", data: { event } };
|
|
case "agent_settled":
|
|
return { type: "agent_state", data: { state: "idle", event } };
|
|
case "turn_start":
|
|
case "turn_end":
|
|
case "message_start":
|
|
case "message_end":
|
|
return { type: "transcript", data: { event } };
|
|
case "message_update":
|
|
return { type: "stream", data: { event } };
|
|
case "tool_execution_start":
|
|
case "tool_execution_update":
|
|
case "tool_execution_end":
|
|
return { type: "tool", data: { event } };
|
|
case "queue_update":
|
|
return { type: "queue", data: { event } };
|
|
case "compaction_start":
|
|
case "compaction_end":
|
|
case "auto_retry_start":
|
|
case "auto_retry_end":
|
|
return { type: "recovery", data: { event } };
|
|
case "extension_ui_request":
|
|
return { type: "extension_ui_request", data: { event } };
|
|
case "extension_error":
|
|
// Extension errors are diagnostics; the Pi process can still be healthy.
|
|
return { type: "transcript", data: { event } };
|
|
default:
|
|
return undefined;
|
|
}
|
|
}
|
|
|
|
function parseFrame(line, onError) {
|
|
try {
|
|
const value = JSON.parse(line);
|
|
if (!isRecord(value) || typeof value.type !== "string") {
|
|
throw new PiRpcError(
|
|
"invalid_message",
|
|
"Pi RPC frame must be an object with a type",
|
|
);
|
|
}
|
|
return value;
|
|
} catch (error) {
|
|
onError(
|
|
error instanceof PiRpcError
|
|
? error
|
|
: new PiRpcError("invalid_json", "Pi RPC emitted invalid JSON"),
|
|
);
|
|
return undefined;
|
|
}
|
|
}
|
|
|
|
function serializeFrame(value) {
|
|
try {
|
|
return `${JSON.stringify(value)}\n`;
|
|
} catch {
|
|
throw new PiRpcError(
|
|
"invalid_command",
|
|
"Pi RPC command must be JSON-serializable",
|
|
);
|
|
}
|
|
}
|
|
|
|
export function startPiRpcAdapter({
|
|
cwd,
|
|
sessionDir,
|
|
sessionPath,
|
|
command = "pi",
|
|
spawnProcess = spawn,
|
|
onEvent = () => {},
|
|
onError = () => {},
|
|
commandTimeoutMs = DEFAULT_COMMAND_TIMEOUT_MS,
|
|
compactCommandTimeoutMs = COMPACT_COMMAND_TIMEOUT_MS,
|
|
maxFrameBytes = MAX_PI_RPC_FRAME_BYTES,
|
|
}) {
|
|
assertAbsolutePath(cwd, "cwd");
|
|
assertAbsolutePath(sessionDir, "sessionDir");
|
|
if (sessionPath !== undefined) assertAbsolutePath(sessionPath, "sessionPath");
|
|
if (typeof spawnProcess !== "function")
|
|
throw new TypeError("spawnProcess must be a function");
|
|
|
|
const child = spawnProcess(
|
|
command,
|
|
[
|
|
"--mode",
|
|
"rpc",
|
|
"--session-dir",
|
|
sessionDir,
|
|
...(sessionPath ? ["--session", sessionPath] : []),
|
|
],
|
|
{
|
|
cwd,
|
|
stdio: ["pipe", "pipe", "pipe"],
|
|
},
|
|
);
|
|
if (!child?.stdin || !child.stdout || !child.stderr) {
|
|
throw new PiRpcError(
|
|
"spawn_failed",
|
|
"Pi RPC child must expose stdin, stdout, and stderr streams",
|
|
);
|
|
}
|
|
|
|
let sequence = 0;
|
|
let stdoutBuffer = "";
|
|
let discardingOversizedFrame = false;
|
|
let closed = false;
|
|
let intentionalStop = false;
|
|
const decoder = new StringDecoder("utf8");
|
|
const pending = new Map();
|
|
let resolveExit;
|
|
const exited = new Promise((resolve) => {
|
|
resolveExit = resolve;
|
|
});
|
|
|
|
const reportError = (error, metadata) => emitSafely(onError, error, metadata);
|
|
const rejectPending = (error) => {
|
|
for (const entry of pending.values()) {
|
|
clearTimeout(entry.timeout);
|
|
entry.reject(error);
|
|
}
|
|
pending.clear();
|
|
};
|
|
const finish = ({ code, signal, error }) => {
|
|
if (closed) return;
|
|
closed = true;
|
|
const terminalError =
|
|
error ??
|
|
new PiRpcError(
|
|
"child_exited",
|
|
`Pi RPC child exited before responding (code ${code ?? "null"}, signal ${signal ?? "none"})`,
|
|
);
|
|
rejectPending(terminalError);
|
|
if (!intentionalStop) reportError(terminalError);
|
|
resolveExit({ code, signal });
|
|
};
|
|
const handleResponse = (response) => {
|
|
if (typeof response.id !== "string") {
|
|
reportError(
|
|
new PiRpcError(
|
|
"uncorrelated_response",
|
|
"Pi RPC response did not include an id",
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
const entry = pending.get(response.id);
|
|
if (!entry) {
|
|
reportError(
|
|
new PiRpcError(
|
|
"uncorrelated_response",
|
|
`Pi RPC response has no pending command: ${response.id}`,
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
clearTimeout(entry.timeout);
|
|
pending.delete(response.id);
|
|
entry.resolve(response);
|
|
};
|
|
const handleEvent = (event) => {
|
|
const normalized = normalizePiEvent(event);
|
|
if (!normalized) {
|
|
reportError(
|
|
new PiRpcError(
|
|
"unsupported_event",
|
|
`Pi RPC emitted unsupported event: ${event.type}`,
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
emitSafely(onEvent, { seq: ++sequence, ...normalized });
|
|
};
|
|
const handleLine = (line) => {
|
|
const frame = parseFrame(
|
|
line.endsWith("\r") ? line.slice(0, -1) : line,
|
|
reportError,
|
|
);
|
|
if (!frame) return;
|
|
if (frame.type === "response") handleResponse(frame);
|
|
else handleEvent(frame);
|
|
};
|
|
const reportOversizedFrame = () =>
|
|
reportError(
|
|
new PiRpcError(
|
|
"frame_too_large",
|
|
`Pi RPC frame exceeds ${maxFrameBytes} bytes`,
|
|
),
|
|
{ terminal: false },
|
|
);
|
|
const consumeStdout = (text) => {
|
|
let remaining = text;
|
|
if (discardingOversizedFrame) {
|
|
const newlineIndex = remaining.indexOf("\n");
|
|
if (newlineIndex === -1) return;
|
|
discardingOversizedFrame = false;
|
|
remaining = remaining.slice(newlineIndex + 1);
|
|
}
|
|
stdoutBuffer += remaining;
|
|
let newlineIndex;
|
|
while ((newlineIndex = stdoutBuffer.indexOf("\n")) !== -1) {
|
|
const line = stdoutBuffer.slice(0, newlineIndex);
|
|
stdoutBuffer = stdoutBuffer.slice(newlineIndex + 1);
|
|
if (Buffer.byteLength(line, "utf8") > maxFrameBytes) {
|
|
reportOversizedFrame();
|
|
continue;
|
|
}
|
|
handleLine(line);
|
|
}
|
|
if (Buffer.byteLength(stdoutBuffer, "utf8") > maxFrameBytes) {
|
|
reportOversizedFrame();
|
|
stdoutBuffer = "";
|
|
discardingOversizedFrame = true;
|
|
}
|
|
};
|
|
|
|
child.stdout.on("data", (chunk) => {
|
|
if (!closed) consumeStdout(decoder.write(chunk));
|
|
});
|
|
child.stdout.on("end", () => {
|
|
if (closed) return;
|
|
consumeStdout(decoder.end());
|
|
if (discardingOversizedFrame) {
|
|
discardingOversizedFrame = false;
|
|
return;
|
|
}
|
|
if (stdoutBuffer.length > 0)
|
|
reportError(
|
|
new PiRpcError(
|
|
"unterminated_frame",
|
|
"Pi RPC stdout ended without an LF-terminated frame",
|
|
),
|
|
);
|
|
stdoutBuffer = "";
|
|
});
|
|
child.stderr.on("data", () => {});
|
|
child.stderr.on("error", (error) =>
|
|
reportError(new PiRpcError("stderr_error", error.message)),
|
|
);
|
|
child.on("error", (error) =>
|
|
finish({
|
|
code: null,
|
|
signal: null,
|
|
error: new PiRpcError("child_error", error.message),
|
|
}),
|
|
);
|
|
child.on("exit", (code, signal) => finish({ code, signal }));
|
|
|
|
function write(value) {
|
|
if (closed || child.stdin.destroyed)
|
|
throw new PiRpcError("child_exited", "Pi RPC child is not available");
|
|
child.stdin.write(serializeFrame(value));
|
|
}
|
|
|
|
return {
|
|
child,
|
|
get sequence() {
|
|
return sequence;
|
|
},
|
|
send(commandInput, { timeoutMs: requestedTimeoutMs } = {}) {
|
|
if (!isRecord(commandInput) || typeof commandInput.type !== "string") {
|
|
return Promise.reject(
|
|
new PiRpcError(
|
|
"invalid_command",
|
|
"Pi RPC command must include a type",
|
|
),
|
|
);
|
|
}
|
|
if (Object.hasOwn(commandInput, "id")) {
|
|
return Promise.reject(
|
|
new PiRpcError(
|
|
"invalid_command",
|
|
"command IDs are assigned by the adapter",
|
|
),
|
|
);
|
|
}
|
|
if (!supportedCommands.has(commandInput.type)) {
|
|
return Promise.reject(
|
|
new PiRpcError(
|
|
"unsupported_command",
|
|
`Pi RPC command is not supported: ${commandInput.type}`,
|
|
),
|
|
);
|
|
}
|
|
if (
|
|
requestedTimeoutMs !== undefined &&
|
|
(!Number.isSafeInteger(requestedTimeoutMs) || requestedTimeoutMs <= 0)
|
|
) {
|
|
return Promise.reject(
|
|
new PiRpcError(
|
|
"invalid_timeout",
|
|
"Pi RPC command timeout must be a positive safe integer",
|
|
),
|
|
);
|
|
}
|
|
|
|
const id = `bridge-${randomUUID()}`;
|
|
const timeoutMs =
|
|
requestedTimeoutMs ??
|
|
(commandInput.type === "compact"
|
|
? compactCommandTimeoutMs
|
|
: commandTimeoutMs);
|
|
return new Promise((resolve, reject) => {
|
|
const timeout = setTimeout(() => {
|
|
pending.delete(id);
|
|
reject(
|
|
new PiRpcError(
|
|
"command_timeout",
|
|
`Pi RPC command timed out: ${commandInput.type}`,
|
|
),
|
|
);
|
|
}, timeoutMs);
|
|
timeout.unref?.();
|
|
pending.set(id, { resolve, reject, timeout });
|
|
try {
|
|
write({ ...commandInput, id });
|
|
} catch (error) {
|
|
clearTimeout(timeout);
|
|
pending.delete(id);
|
|
reject(error);
|
|
}
|
|
});
|
|
},
|
|
respondToExtension(requestId, response) {
|
|
if (
|
|
typeof requestId !== "string" ||
|
|
requestId.length === 0 ||
|
|
!isRecord(response)
|
|
) {
|
|
throw new PiRpcError(
|
|
"invalid_extension_response",
|
|
"extension response requires a request id and response object",
|
|
);
|
|
}
|
|
// Pi uses this id to resume the blocked extension dialog; it cannot be replaced by a command ID.
|
|
write({ type: "extension_ui_response", id: requestId, ...response });
|
|
},
|
|
async stop() {
|
|
if (closed) return exited;
|
|
intentionalStop = true;
|
|
rejectPending(
|
|
new PiRpcError(
|
|
"child_stopped",
|
|
"Pi RPC child was stopped by the bridge",
|
|
),
|
|
);
|
|
child.kill("SIGTERM");
|
|
return exited;
|
|
},
|
|
};
|
|
}
|