Repository navigation
refactor(pi): use native RPC CLI runtime #384
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -53,7 +53,9 @@ function checkLocalAgentProviderAvailability( | |
| case "opencode": | ||
| return commandAvailability(providerInstanceId, "opencode", providerEnv); | ||
| case "pi": | ||
| return packageAvailability(providerInstanceId, "@earendil-works/pi-coding-agent"); | ||
| return providerEnv.PI_COMMAND | ||
| ? commandAvailability(providerInstanceId, providerEnv.PI_COMMAND, providerEnv) | ||
| : packageAvailability(providerInstanceId, "@earendil-works/pi-coding-agent"); | ||
|
Comment on lines
55
to
+58
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If Knowledge Base Used: Agent discovery, availability, and profiles ArtifactsSystem Pi discovery and runtime reproduction script
Run without a Pi command override
Run with an explicit Pi command override
|
||
| case "acp": { | ||
| const configured = config ? subagentProviderConfig(config, providerInstanceId) : undefined; | ||
| const command = configured?.command | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| import assert from "node:assert/strict"; | ||
| import { PiRpcConnection } from "./local-agent-pi-rpc.js"; | ||
|
|
||
| const script = String.raw` | ||
| let buffer = ""; | ||
| process.stdin.setEncoding("utf8"); | ||
| process.stdin.on("data", (chunk) => { | ||
| buffer += chunk; | ||
| for (;;) { | ||
| const newline = buffer.indexOf("\n"); | ||
| if (newline < 0) break; | ||
| const line = buffer.slice(0, newline); | ||
| buffer = buffer.slice(newline + 1); | ||
| if (!line) continue; | ||
| const request = JSON.parse(line); | ||
| process.stdout.write("not-json\n"); | ||
| process.stdout.write(JSON.stringify({ type: "agent_start" }) + "\n"); | ||
| process.stdout.write(JSON.stringify({ | ||
| type: "response", | ||
| id: request.id, | ||
| command: request.type, | ||
| success: true, | ||
| data: { echoed: request.value }, | ||
| }) + "\n"); | ||
| } | ||
| }); | ||
| `; | ||
|
|
||
| const connection = PiRpcConnection.spawn({ | ||
| command: process.execPath, | ||
| args: ["-e", script], | ||
| cwd: process.cwd(), | ||
| env: process.env, | ||
| }); | ||
| const response = await connection.request({ type: "echo", value: "hello" }); | ||
| assert.deepEqual(response, { echoed: "hello" }); | ||
| assert.deepEqual(await connection.nextEvent(1_000), { type: "agent_start" }); | ||
| connection.close(); |
| Original file line number | Diff line number | Diff line change | ||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| @@ -0,0 +1,197 @@ | ||||||||||||||||||||||||||||||||||||
| import { createRequire } from "node:module"; | ||||||||||||||||||||||||||||||||||||
| import type { ChildProcessWithoutNullStreams } from "node:child_process"; | ||||||||||||||||||||||||||||||||||||
| import { terminateProcessTree } from "./process-platform.js"; | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| const require = createRequire(import.meta.url); | ||||||||||||||||||||||||||||||||||||
| const spawn = require("cross-spawn") as typeof import("node:child_process").spawn; | ||||||||||||||||||||||||||||||||||||
| const DEFAULT_REQUEST_TIMEOUT_MS = 30_000; | ||||||||||||||||||||||||||||||||||||
| const MAX_RECORD_CHARS = 8 * 1024 * 1024; | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| export type PiRpcRecord = Record<string, unknown>; | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| interface PendingRequest { | ||||||||||||||||||||||||||||||||||||
| resolve(value: unknown): void; | ||||||||||||||||||||||||||||||||||||
| reject(error: Error): void; | ||||||||||||||||||||||||||||||||||||
| timer: NodeJS.Timeout; | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| export class PiRpcConnection { | ||||||||||||||||||||||||||||||||||||
| private readonly pending = new Map<string, PendingRequest>(); | ||||||||||||||||||||||||||||||||||||
| private readonly queuedEvents: PiRpcRecord[] = []; | ||||||||||||||||||||||||||||||||||||
| private readonly eventWaiters: Array<{ | ||||||||||||||||||||||||||||||||||||
| resolve(value: PiRpcRecord): void; | ||||||||||||||||||||||||||||||||||||
| reject(error: Error): void; | ||||||||||||||||||||||||||||||||||||
| timer?: NodeJS.Timeout; | ||||||||||||||||||||||||||||||||||||
| }> = []; | ||||||||||||||||||||||||||||||||||||
| private nextRequestId = 0; | ||||||||||||||||||||||||||||||||||||
| private buffer = ""; | ||||||||||||||||||||||||||||||||||||
| private closed = false; | ||||||||||||||||||||||||||||||||||||
| private failure?: Error; | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| private constructor(private readonly child: ChildProcessWithoutNullStreams) { | ||||||||||||||||||||||||||||||||||||
| child.stdout.setEncoding("utf8"); | ||||||||||||||||||||||||||||||||||||
| child.stdout.on("data", (chunk: string) => this.consume(chunk)); | ||||||||||||||||||||||||||||||||||||
| child.once("error", (error) => this.fail(error)); | ||||||||||||||||||||||||||||||||||||
| child.once("exit", (code) => { | ||||||||||||||||||||||||||||||||||||
| if (!this.closed) this.fail(new Error(`Pi RPC process exited with code ${code}.`)); | ||||||||||||||||||||||||||||||||||||
| }); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
Comment on lines
+31
to
+38
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win Add an error listener to The constructor listens for 🐛 Proposed fix private constructor(private readonly child: ChildProcessWithoutNullStreams) {
child.stdout.setEncoding("utf8");
child.stdout.on("data", (chunk: string) => this.consume(chunk));
+ child.stdin.on("error", (error) => this.fail(error));
child.once("error", (error) => this.fail(error));📝 Committable suggestion
Suggested change
🤖 Prompt for AI AgentsSource: Learnings |
||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| static spawn(input: { | ||||||||||||||||||||||||||||||||||||
| command: string; | ||||||||||||||||||||||||||||||||||||
| args: string[]; | ||||||||||||||||||||||||||||||||||||
| cwd: string; | ||||||||||||||||||||||||||||||||||||
| env: NodeJS.ProcessEnv; | ||||||||||||||||||||||||||||||||||||
| }): PiRpcConnection { | ||||||||||||||||||||||||||||||||||||
| const detached = process.platform !== "win32"; | ||||||||||||||||||||||||||||||||||||
| const child = spawn(input.command, input.args, { | ||||||||||||||||||||||||||||||||||||
| cwd: input.cwd, | ||||||||||||||||||||||||||||||||||||
| env: input.env, | ||||||||||||||||||||||||||||||||||||
| stdio: ["pipe", "pipe", "pipe"], | ||||||||||||||||||||||||||||||||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
The child’s stderr is piped but never consumed. If Pi or an extension writes enough diagnostics, the pipe fills and blocks the child before it can send its RPC response, causing the agent turn to time out. The pipe must be drained before merging. Knowledge Base Used: Agent runtime and provider adapters ArtifactsExecutable child-diagnostics RPC contract check
RPC timeout with the shipped undrained stderr pipe
Successful RPC response with stderr drained
|
||||||||||||||||||||||||||||||||||||
| detached, | ||||||||||||||||||||||||||||||||||||
| windowsHide: true, | ||||||||||||||||||||||||||||||||||||
| }); | ||||||||||||||||||||||||||||||||||||
|
Comment on lines
+47
to
+53
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win Drain or redirect the child's stderr.
Drain stderr into a small bounded buffer. Include the buffered text in the exit error from 🐛 Proposed fix private constructor(private readonly child: ChildProcessWithoutNullStreams) {
child.stdout.setEncoding("utf8");
child.stdout.on("data", (chunk: string) => this.consume(chunk));
+ child.stderr.setEncoding("utf8");
+ child.stderr.on("data", (chunk: string) => {
+ this.stderrTail = (this.stderrTail + chunk).slice(-8_192);
+ });
child.once("error", (error) => this.fail(error));
child.once("exit", (code) => {
- if (!this.closed) this.fail(new Error(`Pi RPC process exited with code ${code}.`));
+ if (!this.closed) {
+ const detail = this.stderrTail.trim();
+ this.fail(new Error(`Pi RPC process exited with code ${code}.${detail ? ` ${detail}` : ""}`));
+ }
});
}Also add the field 🤖 Prompt for AI Agents |
||||||||||||||||||||||||||||||||||||
| if (!child.stdin || !child.stdout || !child.stderr) { | ||||||||||||||||||||||||||||||||||||
| throw new Error("Pi RPC process did not expose stdio pipes."); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
| return new PiRpcConnection(child as ChildProcessWithoutNullStreams); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| async request(record: PiRpcRecord, timeoutMs = DEFAULT_REQUEST_TIMEOUT_MS): Promise<unknown> { | ||||||||||||||||||||||||||||||||||||
| this.assertOpen(); | ||||||||||||||||||||||||||||||||||||
| const id = `devspace-${++this.nextRequestId}`; | ||||||||||||||||||||||||||||||||||||
| return new Promise((resolve, reject) => { | ||||||||||||||||||||||||||||||||||||
| const timer = setTimeout(() => { | ||||||||||||||||||||||||||||||||||||
| this.pending.delete(id); | ||||||||||||||||||||||||||||||||||||
| reject(new Error(`Pi RPC ${String(record.type ?? "request")} timed out after ${timeoutMs}ms.`)); | ||||||||||||||||||||||||||||||||||||
| }, timeoutMs); | ||||||||||||||||||||||||||||||||||||
| timer.unref(); | ||||||||||||||||||||||||||||||||||||
| this.pending.set(id, { resolve, reject, timer }); | ||||||||||||||||||||||||||||||||||||
| this.write({ ...record, id }).catch((error) => { | ||||||||||||||||||||||||||||||||||||
| const pending = this.pending.get(id); | ||||||||||||||||||||||||||||||||||||
| if (!pending) return; | ||||||||||||||||||||||||||||||||||||
| this.pending.delete(id); | ||||||||||||||||||||||||||||||||||||
| clearTimeout(pending.timer); | ||||||||||||||||||||||||||||||||||||
| pending.reject(error); | ||||||||||||||||||||||||||||||||||||
| }); | ||||||||||||||||||||||||||||||||||||
| }); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| async send(record: PiRpcRecord): Promise<void> { | ||||||||||||||||||||||||||||||||||||
| this.assertOpen(); | ||||||||||||||||||||||||||||||||||||
| await this.write(record); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| async nextEvent(timeoutMs?: number): Promise<PiRpcRecord | undefined> { | ||||||||||||||||||||||||||||||||||||
| this.assertOpen(); | ||||||||||||||||||||||||||||||||||||
| const queued = this.queuedEvents.shift(); | ||||||||||||||||||||||||||||||||||||
| if (queued) return queued; | ||||||||||||||||||||||||||||||||||||
| return new Promise((resolve, reject) => { | ||||||||||||||||||||||||||||||||||||
| const waiter: (typeof this.eventWaiters)[number] = { | ||||||||||||||||||||||||||||||||||||
| resolve: (value) => { | ||||||||||||||||||||||||||||||||||||
| if (waiter.timer) clearTimeout(waiter.timer); | ||||||||||||||||||||||||||||||||||||
| resolve(value); | ||||||||||||||||||||||||||||||||||||
| }, | ||||||||||||||||||||||||||||||||||||
| reject: (error) => { | ||||||||||||||||||||||||||||||||||||
| if (waiter.timer) clearTimeout(waiter.timer); | ||||||||||||||||||||||||||||||||||||
| reject(error); | ||||||||||||||||||||||||||||||||||||
| }, | ||||||||||||||||||||||||||||||||||||
| }; | ||||||||||||||||||||||||||||||||||||
| if (timeoutMs !== undefined) { | ||||||||||||||||||||||||||||||||||||
| waiter.timer = setTimeout(() => { | ||||||||||||||||||||||||||||||||||||
| const index = this.eventWaiters.indexOf(waiter); | ||||||||||||||||||||||||||||||||||||
| if (index >= 0) this.eventWaiters.splice(index, 1); | ||||||||||||||||||||||||||||||||||||
| resolve(undefined); | ||||||||||||||||||||||||||||||||||||
| }, timeoutMs); | ||||||||||||||||||||||||||||||||||||
| waiter.timer.unref(); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
| this.eventWaiters.push(waiter); | ||||||||||||||||||||||||||||||||||||
| }); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| isAlive(): boolean { | ||||||||||||||||||||||||||||||||||||
| return !this.closed && !this.failure && this.child.exitCode === null; | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| close(): void { | ||||||||||||||||||||||||||||||||||||
| if (this.closed) return; | ||||||||||||||||||||||||||||||||||||
| this.closed = true; | ||||||||||||||||||||||||||||||||||||
| const detached = process.platform !== "win32"; | ||||||||||||||||||||||||||||||||||||
| terminateProcessTree(this.child, "SIGTERM", detached); | ||||||||||||||||||||||||||||||||||||
| this.fail(new Error("Pi RPC process closed.")); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| private async write(record: PiRpcRecord): Promise<void> { | ||||||||||||||||||||||||||||||||||||
| const line = `${JSON.stringify(record)}\n`; | ||||||||||||||||||||||||||||||||||||
| await new Promise<void>((resolve, reject) => { | ||||||||||||||||||||||||||||||||||||
| this.child.stdin.write(line, (error) => error ? reject(error) : resolve()); | ||||||||||||||||||||||||||||||||||||
| }); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| private consume(chunk: string): void { | ||||||||||||||||||||||||||||||||||||
| this.buffer += chunk; | ||||||||||||||||||||||||||||||||||||
| if (this.buffer.length > MAX_RECORD_CHARS && !this.buffer.includes("\n")) { | ||||||||||||||||||||||||||||||||||||
| this.buffer = ""; | ||||||||||||||||||||||||||||||||||||
| return; | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
| for (;;) { | ||||||||||||||||||||||||||||||||||||
| const newline = this.buffer.indexOf("\n"); | ||||||||||||||||||||||||||||||||||||
| if (newline < 0) return; | ||||||||||||||||||||||||||||||||||||
| let line = this.buffer.slice(0, newline); | ||||||||||||||||||||||||||||||||||||
| this.buffer = this.buffer.slice(newline + 1); | ||||||||||||||||||||||||||||||||||||
| if (line.endsWith("\r")) line = line.slice(0, -1); | ||||||||||||||||||||||||||||||||||||
| if (!line.trim()) continue; | ||||||||||||||||||||||||||||||||||||
| let parsed: unknown; | ||||||||||||||||||||||||||||||||||||
| try { | ||||||||||||||||||||||||||||||||||||
| parsed = JSON.parse(line); | ||||||||||||||||||||||||||||||||||||
| } catch { | ||||||||||||||||||||||||||||||||||||
| continue; | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
| if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) continue; | ||||||||||||||||||||||||||||||||||||
| this.route(parsed as PiRpcRecord); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| private route(record: PiRpcRecord): void { | ||||||||||||||||||||||||||||||||||||
| if (record.type === "response" && typeof record.id === "string") { | ||||||||||||||||||||||||||||||||||||
| const pending = this.pending.get(record.id); | ||||||||||||||||||||||||||||||||||||
| if (pending) { | ||||||||||||||||||||||||||||||||||||
| this.pending.delete(record.id); | ||||||||||||||||||||||||||||||||||||
| clearTimeout(pending.timer); | ||||||||||||||||||||||||||||||||||||
| if (record.success === true) pending.resolve(record.data); | ||||||||||||||||||||||||||||||||||||
| else pending.reject(new Error(String(record.error ?? `Pi RPC ${String(record.command)} failed.`))); | ||||||||||||||||||||||||||||||||||||
| return; | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
| const waiter = this.eventWaiters.shift(); | ||||||||||||||||||||||||||||||||||||
| if (waiter) waiter.resolve(record); | ||||||||||||||||||||||||||||||||||||
| else this.queuedEvents.push(record); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| private assertOpen(): void { | ||||||||||||||||||||||||||||||||||||
| if (this.failure) throw this.failure; | ||||||||||||||||||||||||||||||||||||
| if (this.closed) throw new Error("Pi RPC process is closed."); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| private fail(error: Error): void { | ||||||||||||||||||||||||||||||||||||
| if (!this.failure) this.failure = error; | ||||||||||||||||||||||||||||||||||||
| for (const [id, pending] of this.pending) { | ||||||||||||||||||||||||||||||||||||
| this.pending.delete(id); | ||||||||||||||||||||||||||||||||||||
| clearTimeout(pending.timer); | ||||||||||||||||||||||||||||||||||||
| pending.reject(error); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
| for (const waiter of this.eventWaiters.splice(0)) waiter.reject(error); | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| export function piRecordString(value: unknown, key: string): string | undefined { | ||||||||||||||||||||||||||||||||||||
| if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; | ||||||||||||||||||||||||||||||||||||
| const field = (value as Record<string, unknown>)[key]; | ||||||||||||||||||||||||||||||||||||
| return typeof field === "string" ? field : undefined; | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||
| export function parsePiModelSlug(model: string): { provider: string; modelId: string } | undefined { | ||||||||||||||||||||||||||||||||||||
| const separator = model.indexOf("/"); | ||||||||||||||||||||||||||||||||||||
| if (separator <= 0 || separator === model.length - 1) return undefined; | ||||||||||||||||||||||||||||||||||||
| return { provider: model.slice(0, separator), modelId: model.slice(separator + 1) }; | ||||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Update the stale Pi effort description.
Lines 98-100 now say that Pi runs in RPC mode. Line 127 still says that
effortmaps to "the AgentSession thinking-level control". That in-process API is no longer used. The runtime now sends RPCset_thinking_level. In the same way,docs/configuration.mdlines 226-229 list process-level overrides but leave outPI_COMMAND.📝 Proposed doc fix
🤖 Prompt for AI Agents