/** * Pi 引擎适配(任务 A3.3,ADR-0006):把 Pi 作为长驻子进程驱动。 * * - 用官方 RpcClient 拉起 `pi --mode rpc`,协议是标准输入输出的 JSONL * - 工具面:启动参数带 --no-builtin-tools / --no-extensions,Pi 自带的 read/bash/edit/write 全部关掉; * 平台的工具由 Pi 侧扩展桥提供(A3.11),扩展还没接之前 Pi 手里没有工具(评测里对应场景跳过) * - 事件面:把 Pi 的会话事件映射成平台的 RuntimeEvent,上层只认平台类型 * - 状态:平台保存会话历史;Pi 自己的 session 文件放在平台管理目录里,只作同引擎恢复用(ADR-0003) */ import { existsSync, mkdirSync, writeFileSync } from 'node:fs'; import { homedir } from 'node:os'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { RpcClient, type JsonAgentSessionEvent } from '@earendil-works/pi-coding-agent'; import type { ExecutionReceipt } from '@aps/contracts'; import type { AgentRuntime, CreateRuntimeSessionInput, RuntimeEvent, RuntimeSession, SendOptions } from '../runtime.js'; export interface PiRuntimeOptions { provider?: string; model?: string; apiKey?: string; cwd?: string; cliPath?: string; /** Pi 自己的会话文件目录(默认 ~/.aps/pi-sessions) */ sessionDir?: string; /** 额外的 CLI 参数,慎用:会覆盖不了默认的关闭项,只是追加 */ extraArgs?: string[]; /** 单回合最长等待(毫秒),超时按引擎错误处理 */ turnTimeoutMs?: number; /** * 能力桥:给定平台内部回调地址与令牌。 * 传了它就会加载平台的能力桥扩展,Pi 才真正拥有平台的能力目录(A3.11)。 * 用函数是为了延后取值——端口要等监听之后才知道。 */ bridge?: () => { baseUrl: string; internalToken: string } | undefined; /** 能力桥扩展文件路径(默认取仓库里的 server/agent/pi/extension/aps-bridge.ts) */ extensionPath?: string; /** 桥配置文件目录(默认 ~/.aps/pi-bridge,内含内部令牌,不入库) */ bridgeDir?: string; } /** provider → 环境变量名:密钥走环境,不进命令行(命令行会被进程列表看到) */ const API_KEY_ENV: Record = { deepseek: 'DEEPSEEK_API_KEY', openai: 'OPENAI_API_KEY', anthropic: 'ANTHROPIC_API_KEY', google: 'GEMINI_API_KEY', }; const DEFAULT_TURN_TIMEOUT_MS = 120_000; const POLL_MS = 25; /** * RpcClient 默认找的是包里的 dist/cli.js,实际可执行入口在 dist/bundle/cli.js,必须显式给。 * * 三种取法依次尝试:ESM 解析(这个包的 exports 只声明了 import 条件,CJS 的 require.resolve 会失败)、 * 仓库根下的包目录、server 包下的包目录。 */ function defaultCliPath(): string { const candidates: string[] = []; try { const resolver = (import.meta as unknown as { resolve?: (specifier: string) => string }).resolve; const resolved = resolver?.('@earendil-works/pi-coding-agent'); if (resolved) candidates.push(join(dirname(fileURLToPath(resolved)), 'bundle', 'cli.js')); } catch { /* 落到下面的候选路径 */ } for (const root of [process.cwd(), join(process.cwd(), 'server')]) { candidates.push(join(root, 'node_modules', '@earendil-works', 'pi-coding-agent', 'dist', 'bundle', 'cli.js')); } return candidates.find((candidate) => existsSync(candidate)) ?? ''; } /** 把 Pi 的失败文本归到平台的六分类(实施设计 5) */ export function classifyPiFailure(detail: string): 'forbidden' | 'rate_limited' | 'unavailable' | 'unknown_result' { if (/401|403|auth|api key|unauthor/i.test(detail)) return 'forbidden'; if (/429|rate.?limit|too many/i.test(detail)) return 'rate_limited'; if (/timeout|ETIMEDOUT|ECONNRESET|ENOTFOUND|socket hang up/i.test(detail)) return 'unavailable'; return 'unavailable'; } /** Pi 的事件联合很宽,这里只映射平台需要的部分,其余忽略 */ export function mapPiEvent(event: JsonAgentSessionEvent): RuntimeEvent[] { const raw = event as unknown as Record; const type = String(raw.type ?? ''); switch (type) { case 'message_update': { const inner = raw.assistantMessageEvent as Record | undefined; if (inner && inner.type === 'text_delta' && typeof inner.delta === 'string') { return [{ type: 'text', delta: inner.delta }]; } return []; } case 'tool_execution_start': return [ { type: 'tool_call', callId: String(raw.toolCallId ?? ''), capabilityId: String(raw.toolName ?? ''), args: raw.args, }, ]; case 'tool_execution_end': { const callId = String(raw.toolCallId ?? ''); // 扩展把平台回执放在 result.details(实测形状):平台侧的 executionId 才是有意义的那个, // Pi 自己的 toolCallId 只用于配对 const result = raw.result as { details?: { executionId?: string; status?: string; result?: unknown; errorKind?: string } } | undefined; const details = result?.details; const failed = raw.isError === true || details?.status === 'failed'; const receipt: ExecutionReceipt = { executionId: details?.executionId ?? callId, status: failed ? 'failed' : ((details?.status as ExecutionReceipt['status']) ?? 'succeeded'), ...(details?.result === undefined ? {} : { result: details.result }), ...(failed ? { errorKind: (details?.errorKind as ExecutionReceipt['errorKind']) ?? 'unknown_result' } : {}), }; return [{ type: 'tool_result', callId, receipt }]; } case 'message_end': { // 模型调用失败不是独立事件:助手消息带 stopReason=error 与 errorMessage(实测形状) const message = raw.message as { role?: string; stopReason?: string; errorMessage?: string } | undefined; if (!message || message.role !== 'assistant') return []; if (message.stopReason !== 'error' && message.errorMessage === undefined) return []; const detail = typeof message.errorMessage === 'string' ? message.errorMessage : ''; return [{ type: 'error', kind: classifyPiFailure(detail), message: detail || '模型调用失败(未提供详情)' }]; } case 'agent_settled': // Pi 明确表示不会再自动继续(重试、压缩、排队都结束了) return [{ type: 'turn_end', reason: 'stop' }]; case 'error': { const reason = raw.reason ?? raw.error; const detail = typeof reason === 'string' ? reason : 'Pi 报错(未提供详情)'; return [{ type: 'error', kind: classifyPiFailure(detail), message: detail }]; } default: return []; } } /** 默认扩展路径:从编译产物反推仓库里的源文件(Pi 用 jiti 直接读 TS) */ function defaultExtensionPath(): string { return fileURLToPath(new URL('../../../agent/pi/extension/aps-bridge.ts', import.meta.url)); } export function createPiRuntime(options: PiRuntimeOptions = {}): AgentRuntime { const provider = options.provider ?? 'deepseek'; const model = options.model ?? 'deepseek-chat'; // Pi 原生会话目录:引擎原生状态只用于同引擎恢复,可用 APS_PI_SESSION_DIR 换到空目录验证「平台补上下文」 const sessionDir = options.sessionDir ?? process.env.APS_PI_SESSION_DIR ?? join(homedir(), '.aps', 'pi-sessions'); const bridgeDir = options.bridgeDir ?? join(homedir(), '.aps', 'pi-bridge'); const turnTimeoutMs = options.turnTimeoutMs ?? DEFAULT_TURN_TIMEOUT_MS; return { name: `pi:${provider}/${model}`, // 工具在 Pi 子进程里执行(经扩展桥回调平台的内部端点),平台只观察事件 toolExecution: 'runtime', async createSession(input: CreateRuntimeSessionInput): Promise { const env: Record = {}; for (const [key, value] of Object.entries(process.env)) { if (value !== undefined) env[key] = value; } if (options.apiKey !== undefined && options.apiKey !== '') { env[API_KEY_ENV[provider] ?? 'DEEPSEEK_API_KEY'] = options.apiKey; } // 能力桥:写一份会话专属配置给扩展读(含内部令牌,放用户目录、不入库) const bridge = options.bridge?.(); if (bridge) { mkdirSync(bridgeDir, { recursive: true }); const bridgeConfigPath = join(bridgeDir, `${input.sessionId}.json`); const bridgeConfig = { baseUrl: bridge.baseUrl, internalToken: bridge.internalToken, sessionId: input.sessionId, capabilities: input.tools.map((tool) => ({ toolName: tool.toolName, capabilityId: tool.capabilityId, description: tool.description, parameters: tool.parameters, })), }; writeFileSync(bridgeConfigPath, JSON.stringify(bridgeConfig, null, 2), { mode: 0o600 }); env.APS_BRIDGE_CONFIG = bridgeConfigPath; } const cliPath = options.cliPath ?? defaultCliPath(); const client = new RpcClient({ ...(cliPath === '' ? {} : { cliPath }), ...(options.cwd === undefined ? {} : { cwd: options.cwd }), provider, model, env, args: [ // 工具只来自能力目录:关掉内置工具与自动发现的扩展/技能(A3.11 再接扩展桥) '--no-builtin-tools', '--no-extensions', '--no-skills', '--no-prompt-templates', '--no-context-files', // 只加载平台自己的能力桥扩展(--no-extensions 关掉自动发现,-e 显式加载仍然有效) '-e', options.extensionPath ?? defaultExtensionPath(), '--session-dir', sessionDir, '--session-id', input.sessionId, ...(input.system === undefined ? [] : ['--system-prompt', input.system]), ...(options.extraArgs ?? []), ], }); if ((options.apiKey ?? '') === '' && (env[API_KEY_ENV[provider] ?? 'DEEPSEEK_API_KEY'] ?? '') === '') { throw new Error(`Pi 引擎未配置模型密钥:请在 ~/.aps/config.json 的 llm.apiKey 里填写,或设置 ${API_KEY_ENV[provider] ?? 'DEEPSEEK_API_KEY'}(ADR-0011)`); } await client.start(); return { id: `pi-${input.sessionId}`, toolExecution: 'runtime', async *send(text: string, sendOptions?: SendOptions): AsyncIterable { const queue: RuntimeEvent[] = []; let wake: (() => void) | undefined; let aborted = false; const debugEvents = process.env.APS_PI_DEBUG_EVENTS === '1'; const unsubscribe = client.onEvent((event) => { // 排查 Pi 集成用:APS_PI_DEBUG_EVENTS=1 时把原始事件打到 stderr if (debugEvents) process.stderr.write(`[pi-raw] ${JSON.stringify(event).slice(0, 1500)} `); queue.push(...mapPiEvent(event)); wake?.(); wake = undefined; }); try { await client.prompt(text); const deadline = Date.now() + turnTimeoutMs; while (Date.now() < deadline) { if (sendOptions?.signal?.aborted === true && !aborted) { aborted = true; await client.abort(); } while (queue.length > 0) { const next = queue.shift() as RuntimeEvent; if (next.type === 'turn_end') { yield { type: 'turn_end', reason: aborted ? 'cancelled' : next.reason }; return; } yield next; } await new Promise((resolve) => { wake = resolve; setTimeout(resolve, POLL_MS); }); } yield { type: 'error', kind: 'unknown_result', message: `Pi 回合超过 ${turnTimeoutMs}ms 未结束` }; yield { type: 'turn_end', reason: 'error' }; } catch (error) { yield { type: 'error', kind: 'unknown_result', message: error instanceof Error ? error.message : 'Pi 子进程异常' }; yield { type: 'turn_end', reason: 'error' }; } finally { unsubscribe(); } }, resolveToolResult(_callId: string, _receipt: ExecutionReceipt): void { // 工具由 Pi 侧扩展执行(A3.11):回执要经扩展桥送回去,这里暂时无处可送 }, abort(): void { void client.abort(); }, async dispose(): Promise { await client.stop(); }, }; }, }; }