aps-agent/server/agent/conformance.ts

/**
 * 引擎一致性测试套件（任务 A3.2）。
 *
 * 同一套用例跑任何运行时实现：多轮对话、工具调用与回执、事件顺序、取消、错误分类、会话恢复
 * （架构设计 5.2 的验收方式）。FakeRuntime 现在跑，PiRuntime 在 A3.3 接上后跑同一套；
 * A3.9「引擎可替换验证」也是拿它做判据。
 */
import type { ExecutionReceipt } from '@aps/contracts';

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

export type RuntimeScenario = 'text' | 'tool_call' | 'cancel' | 'error';

export interface ConformanceCheck {
  name: string;
  ok: boolean;
  detail?: string;
}

export interface ConformanceReport {
  runtimeName: string;
  ok: boolean;
  checks: ConformanceCheck[];
}

export interface ConformanceOptions {
  /** 为某个场景创建一个运行时；同一场景可能创建多次 */
  createRuntime: (scenario: RuntimeScenario) => AgentRuntime | Promise<AgentRuntime>;
  /** 能力目录投影 */
  tools: ToolDescriptor[];
  /** 各场景的输入文本；真引擎用真实问法，假引擎用任意文本 */
  prompts?: Partial<Record<RuntimeScenario, string>>;
  /** 单次 send 的超时（毫秒），防止实现挂住测试 */
  sendTimeoutMs?: number;
  /** 跳过的场景（附原因），报告里记为跳过而不是通过 */
  skipScenarios?: Partial<Record<RuntimeScenario, string>>;
  /**
   * 会话 id 覆盖：真引擎（Pi 走扩展桥）要求平台侧真的存在这个会话，
   * 所以调用方可以先在平台建会话再把 id 传进来。
   */
  sessionIds?: Partial<Record<RuntimeScenario | 'restore', string>>;
  /** 工具场景期望的能力 id（默认 test.echo；真引擎用平台能力目录里的 id） */
  toolCapabilityId?: string;
}

const DEFAULT_PROMPTS: Record<RuntimeScenario, string> = {
  text: '你好，介绍一下你自己',
  tool_call: '查一下下周的排产结果',
  cancel: '跑一个很长的任务',
  error: '触发一次错误',
};

const CAPABILITY_FOR_SCENARIO: Partial<Record<RuntimeScenario, string>> = {
  tool_call: 'test.echo',
};

export function createTestTools(): ToolDescriptor[] {
  return [
    { capabilityId: 'test.echo', toolName: 'test_echo', description: '回显', parameters: { type: 'object' } },
    { capabilityId: 'test.write', toolName: 'test_write', description: '假写入', parameters: { type: 'object' } },
  ];
}

/** 单次迭代也要有超时：真引擎挂住时不能让测试跟着挂住 */
function withTimeout<T>(promise: Promise<T>, timeoutMs: number, label: string): Promise<T> {
  return Promise.race([
    promise,
    new Promise<T>((_resolve, reject) => {
      setTimeout(() => reject(new Error(`${label} 超过 ${timeoutMs}ms 未返回`)), timeoutMs).unref();
    }),
  ]);
}

async function collect(
  events: AsyncIterable<RuntimeEvent>,
  timeoutMs: number,
  hooks?: { onToolCall?: (callId: string) => void },
): Promise<RuntimeEvent[]> {
  const collected: RuntimeEvent[] = [];
  const iterator = events[Symbol.asyncIterator]();
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    const next = await withTimeout(iterator.next(), Math.max(50, deadline - Date.now()), '等待运行时事件');
    if (next.done) return collected;
    const event = next.value as RuntimeEvent;
    collected.push(event);
    if (event.type === 'tool_call') hooks?.onToolCall?.(event.callId);
    if (event.type === 'turn_end') return collected;
  }
  throw new Error(`send 未在 ${timeoutMs}ms 内结束；已收到：${collected.map((e) => e.type).join(',') || '（无）'}`);
}

export async function runRuntimeConformance(options: ConformanceOptions): Promise<ConformanceReport> {
  const timeoutMs = options.sendTimeoutMs ?? 5_000;
  const skips = options.skipScenarios ?? {};
  const ids = {
    text: options.sessionIds?.text ?? 'conformance-1',
    tool_call: options.sessionIds?.tool_call ?? 'conformance-tools',
    cancel: options.sessionIds?.cancel ?? 'conformance-cancel',
    error: options.sessionIds?.error ?? 'conformance-error',
    restore: options.sessionIds?.restore ?? 'conformance-restore',
  };
  const prompts = { ...DEFAULT_PROMPTS, ...options.prompts };
  const checks: ConformanceCheck[] = [];
  const record = (name: string, ok: boolean, detail?: string) => {
    checks.push(detail === undefined ? { name, ok } : { name, ok, detail });
  };

  let runtimeName = '(未知)';
  let session: RuntimeSession | undefined;

  try {
    const runtime = await options.createRuntime('text');
    runtimeName = runtime.name;
    session = await runtime.createSession({ sessionId: ids.text, subject: { userId: 'u', tenantId: 't' }, projectId: null, tools: options.tools });
    record('createSession 返回会话', typeof session.id === 'string' && session.id.length > 0);

    // 1) 单轮文本：事件顺序与结束原因
    const textEvents = await collect(session.send(prompts.text), timeoutMs);
    record('单轮产出文本并正常结束', textEvents.some((e) => e.type === 'text') && textEvents.at(-1)?.type === 'turn_end');
    const last = textEvents.at(-1);
    record('结束原因为 stop', last?.type === 'turn_end' && last.reason === 'stop', last?.type === 'turn_end' ? `reason=${last.reason}` : '缺少 turn_end');
    const endIndex = textEvents.findIndex((e) => e.type === 'turn_end');
    record('turn_end 是最后一个事件', endIndex === textEvents.length - 1, `turn_end 在第 ${endIndex + 1}/${textEvents.length} 个`);

    // 2) 多轮：同一会话连续两轮
    const secondTurn = await collect(session.send(prompts.text), timeoutMs);
    record('同一会话可多轮', secondTurn.at(-1)?.type === 'turn_end' && session.id === session.id);
  } catch (error) {
    record('基础回合（文本与多轮）', false, error instanceof Error ? error.message : '未知错误');
  }

  // 3) 工具调用与回执
  if (skips.tool_call !== undefined) {
    record('工具调用与回执：跳过', true, skips.tool_call);
  } else {
    try {
      const runtime = await options.createRuntime('tool_call');
      const toolSession = await runtime.createSession({ sessionId: ids.tool_call, subject: { userId: 'u', tenantId: 't' }, projectId: null, tools: options.tools });
      let toolCallSeen = false;
      const capabilityId = options.toolCapabilityId ?? CAPABILITY_FOR_SCENARIO.tool_call ?? options.tools[0]?.capabilityId ?? '';
      const events = await collect(toolSession.send(prompts.tool_call), timeoutMs, {
        onToolCall: (callId) => {
          toolCallSeen = true;
          // platform 模式：平台执行后把回执送回；runtime 模式：工具在引擎侧执行（扩展桥），平台只观察
          if (toolSession.toolExecution === 'platform') {
            const receipt: ExecutionReceipt = { executionId: `exec-${callId}`, status: 'succeeded', result: { ok: true } };
            toolSession.resolveToolResult(callId, receipt);
          }
        },
      });
      const receiptSeen = events.some((e) => e.type === 'tool_result');
      const callEvent = events.find((e) => e.type === 'tool_call');
      record('发出 tool_call 并带能力 id', toolCallSeen && callEvent?.type === 'tool_call' && callEvent.capabilityId === capabilityId, `期望 ${capabilityId}`);
      record('回执回流后产出 tool_result', receiptSeen, `执行模式：${toolSession.toolExecution}`);
      record('工具回合以 turn_end 收尾', events.at(-1)?.type === 'turn_end');
      await toolSession.dispose();
    } catch (error) {
      record('工具调用与回执', false, error instanceof Error ? error.message : '未知错误');
    }
  }

  // 4) 取消
  if (skips.cancel !== undefined) {
    record('取消：跳过', true, skips.cancel);
  } else try {
    const runtime = await options.createRuntime('cancel');
    const cancelSession = await runtime.createSession({ sessionId: ids.cancel, subject: { userId: 'u', tenantId: 't' }, projectId: null, tools: options.tools });
    const controller = new AbortController();
    const events: RuntimeEvent[] = [];
    const iterator = cancelSession.send(prompts.cancel, { signal: controller.signal })[Symbol.asyncIterator]();
    cancelSession.abort();
    controller.abort();
    const deadline = Date.now() + timeoutMs;
    while (Date.now() < deadline) {
      const next = await withTimeout(iterator.next(), timeoutMs, '等待取消后的事件');
      if (next.done) break;
      events.push(next.value as RuntimeEvent);
      if ((next.value as RuntimeEvent).type === 'turn_end') break;
    }
    const end = events.at(-1);
    record('取消后以 turn_end(cancelled) 结束', end?.type === 'turn_end' && end.reason === 'cancelled', end?.type === 'turn_end' ? `reason=${end.reason}` : '缺少 turn_end');
    await cancelSession.dispose();
  } catch (error) {
    record('取消', false, error instanceof Error ? error.message : '未知错误');
  }

  // 5) 错误分类
  if (skips.error !== undefined) {
    record('错误分类：跳过', true, skips.error);
  } else try {
    const runtime = await options.createRuntime('error');
    const errorSession = await runtime.createSession({ sessionId: ids.error, subject: { userId: 'u', tenantId: 't' }, projectId: null, tools: options.tools });
    const events = await collect(errorSession.send(prompts.error), timeoutMs);
    const errorEvent = events.find((e) => e.type === 'error');
    const validKinds = ['invalid_argument', 'forbidden', 'conflict', 'rate_limited', 'unavailable', 'unknown_result'];
    record(
      '错误事件带合法分类',
      errorEvent?.type === 'error' && validKinds.includes(errorEvent.kind),
      errorEvent?.type === 'error' ? `kind=${errorEvent.kind}` : '未收到 error 事件',
    );
    await errorSession.dispose();
  } catch (error) {
    record('错误分类', false, error instanceof Error ? error.message : '未知错误');
  }

  // 6) 会话恢复：用平台历史创建新会话并继续
  if (skips.text !== undefined) {
    record('会话恢复：跳过', true, skips.text);
  } else try {
    const runtime = await options.createRuntime('text');
    const restored = await runtime.createSession({
      sessionId: ids.restore,
      subject: { userId: 'u', tenantId: 't' },
      projectId: null,
      tools: options.tools,
      history: [
        { messageId: 'm-1', role: 'user', ts: Date.now(), text: '之前的问题' },
        { messageId: 'm-2', role: 'assistant', ts: Date.now(), text: '之前的回答' },
      ],
    });
    const events = await collect(restored.send(prompts.text), timeoutMs);
    record('带历史恢复后仍可对话', events.at(-1)?.type === 'turn_end');
    await restored.dispose();
  } catch (error) {
    record('会话恢复', false, error instanceof Error ? error.message : '未知错误');
  }

  if (session) await session.dispose();

  return { runtimeName, ok: checks.every((check) => check.ok), checks };
}