aps-agent/server/agent/pi/runtime.ts

/**
 * 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 } 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;
}

/** provider → 环境变量名：密钥走环境，不进命令行（命令行会被进程列表看到） */
const API_KEY_ENV: Record<string, string> = {
  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<string, unknown>;
  const type = String(raw.type ?? '');

  switch (type) {
    case 'message_update': {
      const inner = raw.assistantMessageEvent as Record<string, unknown> | 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 ?? '');
      const receipt: ExecutionReceipt = {
        executionId: callId,
        status: raw.isError === true ? 'failed' : 'succeeded',
        result: raw.result,
        ...(raw.isError === true ? { errorKind: 'unknown_result' as const } : {}),
      };
      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 [];
  }
}

export function createPiRuntime(options: PiRuntimeOptions = {}): AgentRuntime {
  const provider = options.provider ?? 'deepseek';
  const model = options.model ?? 'deepseek-chat';
  const sessionDir = options.sessionDir ?? join(homedir(), '.aps', 'pi-sessions');
  const turnTimeoutMs = options.turnTimeoutMs ?? DEFAULT_TURN_TIMEOUT_MS;

  return {
    name: `pi:${provider}/${model}`,
    async createSession(input: CreateRuntimeSessionInput): Promise<RuntimeSession> {
      const env: Record<string, string> = {};
      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 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',
          '--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}`,
        async *send(text: string, sendOptions?: SendOptions): AsyncIterable<RuntimeEvent> {
          const queue: RuntimeEvent[] = [];
          let wake: (() => void) | undefined;
          let aborted = false;

          const unsubscribe = client.onEvent((event) => {
            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<void>((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<void> {
          await client.stop();
        },
      };
    },
  };
}