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, 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<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 的事件联合很宽，这里只映射平台需要的部分，其余忽略 */
/**
 * Pi 的事件联合很宽，这里只映射平台需要的部分，其余忽略。
 * resolveCapabilityId：把 Pi 的工具名（fake_echo）还原成能力 id（fake.echo）——平台事件一律用能力 id。
 */
export function mapPiEvent(event: JsonAgentSessionEvent, resolveCapabilityId?: (toolName: string) => string): 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': {
      const toolName = String(raw.toolName ?? '');
      return [
        {
          type: 'tool_call',
          callId: String(raw.toolCallId ?? ''),
          capabilityId: resolveCapabilityId?.(toolName) ?? 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<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 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）`);
      }

      // 工具名 → 能力 id 的反向映射（扩展注册时用的是 toolName）
      const nameToCapability = new Map(input.tools.map((tool) => [tool.toolName, tool.capabilityId]));
      const resolveCapabilityId = (toolName: string): string => nameToCapability.get(toolName) ?? toolName;

      await client.start();

      return {
        id: `pi-${input.sessionId}`,
        toolExecution: 'runtime',
        async *send(text: string, sendOptions?: SendOptions): AsyncIterable<RuntimeEvent> {
          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, resolveCapabilityId));
            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();
        },
        /** 日志轮转用：把引擎 stderr 交给进程管理层（它只保留尾部） */
        stderrTail(): string {
          return client.getStderr();
        },
        async dispose(): Promise<void> {
          await client.stop();
        },
      };
    },
  };
}