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

/**
 * 假引擎（任务 A3.2）：确定性的运行时，用于一致性测试与问法评测（A2.7）。
 *
 * 它不做任何「理解」：按脚本或计划器产出事件，形状与真引擎一致。
 * 这样一致性测试可以断言事件顺序，评测装置也能在无模型的情况下跑通。
 */
import { randomUUID } from 'node:crypto';

import type { ExecutionReceipt } from '@aps/contracts';
import type { ErrorKind } from '@aps/provider-sdk';

import type { AgentRuntime, CreateRuntimeSessionInput, RuntimeEvent, RuntimeSession, SendOptions, ToolDescriptor } from './runtime.js';

/** 一个回合的脚本：先说话、再调工具、拿到回执后继续说话，或直接报错 */
export interface FakeTurn {
  text?: string[];
  toolCalls?: Array<{ capabilityId: string; args: unknown }>;
  afterToolsText?: string[];
  error?: { kind: ErrorKind; message: string };
  /** 开场延迟（毫秒）：用来制造一个「还在跑」的窗口，便于测取消，默认 0 */
  delayMs?: number;
}

export interface FakeRuntimeOptions {
  name?: string;
  /** 按回合顺序播放；用完后回到默认行为 */
  turns?: FakeTurn[];
  /** 或者由计划器决定每个回合怎么走 */
  planner?: (text: string, tools: ToolDescriptor[]) => FakeTurn;
}

export interface FakeRuntime extends AgentRuntime {
  stats: { sessions: number; turns: number; aborts: number };
}

const sleep = (ms: number) => (ms > 0 ? new Promise((resolve) => setTimeout(resolve, ms)) : Promise.resolve());

export function createFakeRuntime(options: FakeRuntimeOptions = {}): FakeRuntime {
  const stats = { sessions: 0, turns: 0, aborts: 0 };
  const scriptedTurns = [...(options.turns ?? [])];

  return {
    name: options.name ?? 'fake',
    toolExecution: 'platform',
    stats,
    async createSession(input: CreateRuntimeSessionInput): Promise<RuntimeSession> {
      stats.sessions += 1;
      const id = `fake-${input.sessionId}-${randomUUID().slice(0, 8)}`;
      const pending = new Map<string, (receipt: ExecutionReceipt) => void>();
      let aborted = false;
      let disposed = false;

      const nextTurn = (text: string): FakeTurn => {
        const scripted = scriptedTurns.shift();
        if (scripted) return scripted;
        if (options.planner) return options.planner(text, input.tools);
        return { text: [`（假引擎）收到：${text}`] };
      };

      async function* runTurn(text: string, sendOptions?: SendOptions): AsyncGenerator<RuntimeEvent> {
        if (disposed) throw new Error('运行时已释放');
        aborted = false;
        stats.turns += 1;
        const turn = nextTurn(text);
        const delay = turn.delayMs ?? 0;

        const checkAbort = () => aborted || sendOptions?.signal?.aborted === true;
        if (checkAbort()) {
          yield { type: 'turn_end', reason: 'cancelled' };
          return;
        }

        if (delay > 0) {
          await sleep(delay);
          if (checkAbort()) {
            yield { type: 'turn_end', reason: 'cancelled' };
            return;
          }
        }

        for (const delta of turn.text ?? []) {
          yield { type: 'text', delta };
        }

        if (turn.error) {
          yield { type: 'error', kind: turn.error.kind, message: turn.error.message };
          yield { type: 'turn_end', reason: 'error' };
          return;
        }

        for (const call of turn.toolCalls ?? []) {
          const callId = `call-${randomUUID().slice(0, 8)}`;
          // 先登记回调再 yield：消费者可能在收到事件的同一刻就把回执送回来
          const receiptPromise = new Promise<ExecutionReceipt>((resolve) => {
            pending.set(callId, resolve);
          });
          yield { type: 'tool_call', callId, capabilityId: call.capabilityId, args: call.args };

          // 等平台把回执送回来（Pi 走扩展桥，这里走内存回调）
          const receipt = await receiptPromise;
          pending.delete(callId);
          yield { type: 'tool_result', callId, receipt };

          if (checkAbort()) {
            yield { type: 'turn_end', reason: 'cancelled' };
            return;
          }
        }

        for (const delta of turn.afterToolsText ?? []) {
          yield { type: 'text', delta };
        }

        yield { type: 'turn_end', reason: checkAbort() ? 'cancelled' : 'stop', usage: { turns: 1 } };
      }

      return {
        id,
        toolExecution: 'platform',
        send(text: string, sendOptions?: SendOptions): AsyncIterable<RuntimeEvent> {
          return runTurn(text, sendOptions);
        },
        resolveToolResult(callId: string, receipt: ExecutionReceipt): void {
          const resolve = pending.get(callId);
          if (resolve) resolve(receipt);
        },
        abort(): void {
          stats.aborts += 1;
          aborted = true;
          for (const resolve of pending.values()) resolve({ executionId: 'aborted', status: 'failed', errorKind: 'conflict' });
          pending.clear();
        },
        async dispose(): Promise<void> {
          disposed = true;
          pending.clear();
        },
      };
    },
  };
}