Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion devlog
100 changes: 70 additions & 30 deletions src/agent/pi-runtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@ import { probeOpenCodexEndpointModels } from '../cli/opencodex-models.js';
import { launchSpec } from '../core/exec-name.js';
import { mergeEnvWindowsSafe } from './spawn-env.js';
import { createTextStreamReader } from './stream-text.js';
import type { RuntimeTurnOutcome } from '../shared/runtime-contract.js';
import { PiTurnAccumulator, PiRuntimeError, piSupportsSettled } from './runtime/pi-turn.js';

export type PiProfileMode = 'basic' | 'openai' | 'anthropic' | 'vertex';
export type PiApiKind = 'openai-completions' | 'openai-responses' | 'anthropic-messages' | 'google-vertex';
Expand Down Expand Up @@ -60,12 +62,18 @@ export interface PiRpcSession {
effort?: string;
onEvent?: (event: PiRuntimeEvent) => void;
onRawRecord?: (record: unknown) => void;
}): Promise<{ text: string; stderr: string }>;
}): Promise<PiPromptResult>;
abort(): Promise<void>;
close(): void;
kill(): void;
}

export interface PiPromptResult {
text: string;
stderr: string;
runtimeOutcome?: RuntimeTurnOutcome;
}

function notifyPiRawRecord(observer: ((record: unknown) => void) | undefined, record: unknown): void {
try { observer?.(record); }
catch { console.warn('[jaw:pi] raw activity observer failed'); }
Expand Down Expand Up @@ -275,17 +283,26 @@ export function resolvePiCommand(env: NodeJS.ProcessEnv = process.env): PiComman
};
}

function resolvePiCommandIdentity(command: PiCommand, env: NodeJS.ProcessEnv = process.env): string {
function probePiCommandVersion(command: PiCommand, env: NodeJS.ProcessEnv) {
const spec = launchSpec(command.command, [...command.baseArgs, '--version'], process.platform, env);
const result = spawnSync(spec.file, spec.args, {
return spawnSync(spec.file, spec.args, {
encoding: 'utf8',
env,
timeout: 15_000,
});
}

function resolvePiCommandIdentity(command: PiCommand, env: NodeJS.ProcessEnv = process.env): string {
const result = probePiCommandVersion(command, env);
const version = `${result.stdout || ''}\n${result.stderr || ''}`.trim() || `exit:${String(result.status)}`;
return JSON.stringify({ source: command.source, command: command.command, baseArgs: command.baseArgs, version });
}

function usesPiSettled(command: PiCommand): boolean {
const result = probePiCommandVersion(command, process.env);

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Cache Pi version detection outside the request path

Whenever a Pi runtime is created, usesPiSettled() synchronously launches <pi command> --version with a 15-second timeout; resolvePiCommand() may already have run its own probe immediately beforehand. With the npm-exec fallback or a slow/broken wrapper, each pooled-session creation—and every direct worker turn—can block the server event loop for seconds, delaying unrelated HTTP, SSE, and agent work. Cache the detected capability or perform this probe asynchronously rather than running spawnSync during each spawn.

Useful? React with 👍 / 👎.

return result.status === 0 && piSupportsSettled(result.stdout || '');
}

function loadPiAbortEffective(profileId: string, command: PiCommand): boolean {
try {
const raw = JSON.parse(fs.readFileSync(join(JAW_HOME, 'pi', 'rpc-capabilities.json'), 'utf8')) as unknown;
Expand Down Expand Up @@ -493,11 +510,12 @@ function extractPiSessionId(obj: Record<string, unknown>): string {
}

type PersistentPrompt = {
text: string;
turn: PiTurnAccumulator;
requestId: number;
stderrStart: number;
onEvent: ((event: PiRuntimeEvent) => void) | undefined;
onRawRecord: ((record: unknown) => void) | undefined;
resolve: (result: { text: string; stderr: string }) => void;
resolve: (result: PiPromptResult) => void;
reject: (error: Error) => void;
};

Expand Down Expand Up @@ -527,6 +545,7 @@ export function spawnPersistentPiRpc(profile: PiProfile, pi: PiSettings, options
}): PiRpcSession {
const dir = ensurePiRuntimeConfig(pi, profile.id, options.effort || '', options.root);
const cmd = resolvePiCommand();
const settledProtocol = usesPiSettled(cmd);
const args = [
...cmd.baseArgs,
'--mode', 'rpc',
Expand Down Expand Up @@ -608,15 +627,23 @@ export function spawnPersistentPiRpc(profile: PiProfile, pi: PiSettings, options
}
if (event.tool) activePrompt?.onEvent?.({ kind: 'tool', ...event.tool });
if (event.thinking) activePrompt?.onEvent?.({ kind: 'thinking', text: event.thinking });
if (event.text && activePrompt && !(event.done && activePrompt.text)) {
activePrompt.text += event.text;
activePrompt.onEvent?.({ kind: 'text', text: event.text });
if (activePrompt && record['type'] === 'response' && record['command'] === 'prompt'
&& record['id'] === activePrompt.requestId && record['success'] === false) {
const message = typeof record['error'] === 'string' ? record['error']
: trimString((record['error'] as Record<string, unknown> | undefined)?.['message']);
rejectOutstanding(new Error(message || 'pi rpc prompt rejected'));
return;
}
if (event.done || (abortWait && nonRunning)) {
const accepted = activePrompt?.turn.observe(record);
if (accepted?.text) activePrompt?.onEvent?.({ kind: 'text', text: accepted.text });
const abortNonRunning = Boolean(isAbortResponse) && nonRunning && !abortWait?.terminal;
const terminal = accepted?.done || (!activePrompt && record['type'] === (settledProtocol ? 'agent_settled' : 'agent_end'));
if (terminal || abortNonRunning) {
if (activePrompt) {
const prompt = activePrompt;
activePrompt = null;
prompt.resolve({ text: prompt.text, stderr: stderr.slice(prompt.stderrStart) });
const runtimeOutcome = prompt.turn.snapshot(abortNonRunning ? 'stopped' : undefined);
prompt.resolve({ text: runtimeOutcome.partialText, stderr: stderr.slice(prompt.stderrStart), runtimeOutcome });
}
if (abortWait) abortWait.terminal = true;
}
Expand All @@ -626,7 +653,7 @@ export function spawnPersistentPiRpc(profile: PiProfile, pi: PiSettings, options
if (activePrompt) {
const prompt = activePrompt;
activePrompt = null;
prompt.reject(error);
prompt.reject(new PiRuntimeError(error, prompt.turn.snapshot(child.killed ? 'stopped' : 'error')));
}
if (abortWait) {
const wait = abortWait;
Expand All @@ -651,13 +678,13 @@ export function spawnPersistentPiRpc(profile: PiProfile, pi: PiSettings, options
// A head-only cap without this reset would freeze stderr.length
// and make every later slice() return ''.
stderr = '';
activePrompt = { text: '', stderrStart: stderr.length, onEvent: opts.onEvent, onRawRecord: opts.onRawRecord, resolve, reject };
activePrompt = { turn: new PiTurnAccumulator(settledProtocol), requestId: 0, stderrStart: stderr.length,
onEvent: opts.onEvent, onRawRecord: opts.onRawRecord, resolve, reject };
try {
if (opts.effort) write('set_thinking_level', { level: opts.effort });
write('prompt', { message });
activePrompt.requestId = write('prompt', { message });
} catch (error) {
activePrompt = null;
reject(error as Error);
rejectOutstanding(error as Error);
}
});
},
Expand Down Expand Up @@ -728,9 +755,10 @@ export function spawnPiRpc(profile: PiProfile, pi: PiSettings, options: {
onEvent?: (event: PiRuntimeEvent) => void;
onRawRecord?: (record: unknown) => void;
root?: string;
}): { child: ChildProcess; done: Promise<{ text: string; code: number; sessionId?: string | null; stderr: string }> } {
}): { child: ChildProcess; done: Promise<PiPromptResult & { code: number; sessionId?: string | null }> } {
const dir = ensurePiRuntimeConfig(pi, profile.id, options.effort || '', options.root);
const cmd = resolvePiCommand();
const turn = new PiTurnAccumulator(usesPiSettled(cmd));
const args = [
...cmd.baseArgs,
'--mode', 'rpc',
Expand All @@ -751,23 +779,25 @@ export function spawnPiRpc(profile: PiProfile, pi: PiSettings, options: {
const decoder = new StringDecoder('utf8');
let buffer = '';
let stderr = '';
let text = '';
const stderrReader = createTextStreamReader();
let sessionId: string | null = null;
let doneSettled = false;
let seq = 1;
const done = new Promise<{ text: string; code: number; sessionId?: string | null; stderr: string }>((resolve, reject) => {
const finish = (code = 0) => {
let promptId = 0;
const done = new Promise<PiPromptResult & { code: number; sessionId?: string | null }>((resolve, reject) => {
const finish = (code = 0, status?: RuntimeTurnOutcome['status']) => {
if (doneSettled) return;
doneSettled = true;
try { child.stdin.end(); } catch { /* already closed */ }
setTimeout(() => {
if (!child.killed && child.exitCode == null) child.kill('SIGTERM');
}, 750);
resolve({ text, code, sessionId, stderr });
const runtimeOutcome = turn.snapshot(status);
resolve({ text: runtimeOutcome.partialText, code, sessionId, stderr, runtimeOutcome });
};
let parseFailures = 0;
const dispatchLine = (line: string) => {
if (doneSettled) return;
let parsed: unknown;
try { parsed = JSON.parse(line); }
catch {
Expand All @@ -776,20 +806,28 @@ export function spawnPiRpc(profile: PiProfile, pi: PiSettings, options: {
return;
}
notifyPiRawRecord(options.onRawRecord, parsed);
const record = parsed && typeof parsed === 'object' ? parsed as Record<string, unknown> : {};
if (record['type'] === 'response' && record['command'] === 'prompt'
&& record['id'] === promptId && record['success'] === false) {
finish(1, 'error');
return;
}
const event = parsePiRpcRecord(parsed);
if (event.sessionId) {
sessionId = event.sessionId;
options.onEvent?.({ kind: 'session', sessionId });
}
if (event.tool) options.onEvent?.({ kind: 'tool', ...event.tool });
if (event.thinking) options.onEvent?.({ kind: 'thinking', text: event.thinking });
if (event.text && !(event.done && text)) {
text += event.text;
options.onEvent?.({ kind: 'text', text: event.text });
}
if (event.done) finish(0);
const accepted = turn.observe(parsed);
if (accepted.text) options.onEvent?.({ kind: 'text', text: accepted.text });
if (accepted.done) finish(0);
};
child.on('error', reject);
child.on('error', error => {
if (doneSettled) return;
doneSettled = true;
reject(new PiRuntimeError(error, turn.snapshot('error')));
});
child.stdout.on('data', (chunk) => {
buffer += decoder.write(chunk);
const lines = buffer.split('\n');
Expand All @@ -802,19 +840,21 @@ export function spawnPiRpc(profile: PiProfile, pi: PiSettings, options: {
for (const line of lines) if (line.trim()) dispatchLine(line.trim());
});
child.stderr.on('data', (chunk) => { if (stderr.length < 4000) stderr += stderrReader.write(chunk); });
child.on('close', (code) => {
child.on('close', (code, signal) => {
if (buffer.trim()) dispatchLine(buffer.trim());
finish(code ?? 0);
finish(code ?? 1, signal || child.killed ? 'stopped' : 'error');
});
});
const write = (type: string, fields: Record<string, unknown> = {}) => {
child.stdin.write(JSON.stringify({ id: seq++, type, ...fields }) + '\n');
const id = seq++;
child.stdin.write(JSON.stringify({ id, type, ...fields }) + '\n');
return id;
};
write('get_state');
if (options.effort) write('set_thinking_level', { level: options.effort });
const fullPrompt = options.sysPrompt ? `${options.sysPrompt}\n\n${options.prompt}` : options.prompt;
const hasHistory = fullPrompt.includes('[Recent Context]');
console.log(`[jaw:pi] prompt len=${fullPrompt.length}, hasHistory=${hasHistory}, effort=${options.effort || 'none'}, sessionId=${options.sessionId || 'new'}`);
write('prompt', { message: fullPrompt });
promptId = write('prompt', { message: fullPrompt });
return { child, done };
}
144 changes: 144 additions & 0 deletions src/agent/runtime/pi-turn.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,144 @@
import type { RuntimeTurnOutcome } from '../../shared/runtime-contract.js';
import { appendBoundedFullText } from '../events/fulltext-bound.js';

type RecordValue = Record<string, unknown>;
const record = (value: unknown): RecordValue => value !== null && typeof value === 'object' && !Array.isArray(value)
? value as RecordValue : {};

function assistant(value: unknown): Omit<RuntimeTurnOutcome, 'partialText'> & { text: string | null } | null {
const message = record(value);
if (message['role'] !== 'assistant') return null;
const content = message['content'];
let text: string | null = null, tool = false, oversized = false, invalid = !Array.isArray(content);
for (const part of Array.isArray(content) ? content : []) {
const block = record(part);
if (block['type'] === 'toolCall') { tool = true; continue; }
if (block['type'] === 'thinking') continue;
if (block['type'] !== 'text' || typeof block['text'] !== 'string') { invalid = true; continue; }
if (block['text'] === '') { text ??= ''; continue; }
const bounded = appendBoundedFullText(text ?? '', block['text']);
text = bounded.text; oversized ||= bounded.truncated;
}
const reason = message['stopReason'];
const status = reason === 'aborted' ? 'stopped'
: !invalid && !oversized && (reason === 'stop' || reason === 'toolUse') ? 'done' : 'error';
return { text, status, finalText: status === 'done' && reason === 'stop' && !tool ? text : null };
}

/** agent_settled entered upstream RPC in 0.80.4; willRetry existed before it. */
export function piSupportsSettled(version: string): boolean {
const match = /^(?:pi\s+)?v?(\d+)\.(\d+)\.(\d+)\s*$/.exec(version.trim());
if (!match) return false;
const [, major, minor, patch] = match;
return Number(major) > 0 || Number(minor) > 80 || (Number(minor) === 80 && Number(patch) >= 4);
}

/** Per admitted RPC prompt; no journal, session IDs, raw snapshots or callbacks retained. */
export class PiTurnAccumulator {
private partial = '';
private current = '';
private runText = '';
private completed = 0;
private candidate: Omit<RuntimeTurnOutcome, 'partialText'> = { status: 'done', finalText: null };
private ended = false;

constructor(private readonly settledProtocol: boolean) {}

private append(text: string): string {
const previous = this.partial.length;
this.partial = appendBoundedFullText(this.partial, text).text;
this.runText = appendBoundedFullText(this.runText, text).text;
this.current = appendBoundedFullText(this.current, text).text;
return text.slice(0, this.partial.length - previous);
}

private reconcile(text: string | null): string {
// Snapshots echo prior deltas. A changed snapshot still owns finalText,
// but must not rewrite or duplicate already accepted salvage bytes.
if (text === null || !text.startsWith(this.current)) return '';
return this.append(text.slice(this.current.length));
}

observe(value: unknown): { text: string; done: boolean } {
if (this.ended) return { text: '', done: false };
const row = record(value);
let text = '';
if (row['type'] === 'agent_start') {
this.current = ''; this.runText = ''; this.completed = 0;
this.candidate = { status: 'done', finalText: null };
} else if (row['type'] === 'message_start' && record(row['message'])['role'] === 'assistant') {
this.current = '';
this.candidate = { status: 'error', finalText: null };
} else if (row['type'] === 'message_update') {
const event = record(row['assistantMessageEvent']);
const role = record(row['message'])['role'];
if ((role === undefined || role === 'assistant') && event['type'] === 'text_delta' && typeof event['delta'] === 'string') {
text = this.append(event['delta']);
}
} else if (row['type'] === 'message_end') {
const message = assistant(row['message']);
if (message) {
text = this.reconcile(message.text);
this.candidate = { status: message.status, finalText: message.finalText };
this.completed++;
this.current = '';
}
} else if (row['type'] === 'agent_end') {
text = this.terminal(row['messages']);
this.ended = !this.settledProtocol && row['willRetry'] !== true;
} else if (row['type'] === 'agent_settled') {
this.ended = true;
}
return { text, done: this.ended };
}

private terminal(value: unknown): string {
if (!Array.isArray(value)) {
this.candidate = { status: this.candidate.status === 'stopped' ? 'stopped' : 'error', finalText: null };
return '';
}
let index = 0, added = '';
// Without message_end boundaries, the aggregate snapshot echoes the
// low-level run's stream. Consume that prefix by offset, not text identity.
let streamed = this.completed === 0 ? this.runText : '';
// An empty/tool-only terminal cannot erase an observed failure. Only
// an actual assistant snapshot can supply a newer completion status.
this.candidate = { status: this.candidate.status, finalText: null };
for (const entry of value) {
const message = assistant(entry);
if (!message) continue;
this.candidate = { status: message.status, finalText: message.finalText };
if (index++ < this.completed) continue;
if (this.completed === 0) {
const snapshot = message.text ?? '';
const consumed = Math.min(streamed.length, snapshot.length);
this.current = streamed.slice(0, consumed);
streamed = streamed.slice(consumed);
}
added = appendBoundedFullText(added, this.reconcile(message.text)).text;
this.current = '';
}
this.completed = index;
return added;
}

snapshot(status?: RuntimeTurnOutcome['status']): RuntimeTurnOutcome {
return { status: status ?? this.candidate.status,
finalText: status ? null : this.candidate.finalText, partialText: this.partial };
}
}

/** Only this local carrier can supply failure outcome; arbitrary Error fields cannot. */
export class PiRuntimeError extends Error {
readonly runtimeOutcome: RuntimeTurnOutcome;
constructor(cause: Error, outcome: RuntimeTurnOutcome) {
super(cause.message, { cause });
this.name = 'PiRuntimeError';
this.runtimeOutcome = { ...outcome };
}
}

export function piFailureOutcome(error: unknown): RuntimeTurnOutcome | undefined {
if (!(error instanceof PiRuntimeError) || !Object.hasOwn(error, 'runtimeOutcome')) return undefined;
return { ...error.runtimeOutcome };
}
Loading
Loading