aps-agent/server/entry/http.ts

/**
 * HTTP / SSE 入口（架构设计第 2 节）。
 *
 * 只做协议：路由、请求解析、身份解析、响应与错误包装、事件流推送。
 * 业务编排在 app/，控制面在 core/（A4 起）。路由表来自协议里的 ENDPOINT_SPECS，
 * 保证「文档写了什么端点，代码就实现什么端点」。
 */
import { createServer, type IncomingMessage, type Server, type ServerResponse } from 'node:http';

import { ENDPOINT_SPECS, type CreateSessionRequest, type EndpointSpec, type PlatformEvent, type SendMessageRequest } from '@aps/contracts';

import { AppError } from '../app/errors.js';
import type { ApplicationService } from '../app/service.js';
import { resolveSubject } from './dev-identity.js';
import { clientDistDir, createStaticHandler } from './static.js';

const MAX_BODY_BYTES = 1_000_000;
const HEARTBEAT_MS = 15_000;

interface Route {
  spec: EndpointSpec;
  regex: RegExp;
  names: string[];
}

const serveStatic = createStaticHandler(clientDistDir());

const ROUTES: Route[] = ENDPOINT_SPECS.map((spec) => {
  const names: string[] = [];
  const pattern = spec.path.replace(/\{(\w+)\}/g, (_match, name: string) => {
    names.push(name);
    return '([^/]+)';
  });
  return { spec, regex: new RegExp(`^${pattern}$`), names };
});

export function createRequestHandler(service: ApplicationService, internalToken: string): (req: IncomingMessage, res: ServerResponse) => void {
  return (req, res) => {
    void handleRequest(req, res, service, internalToken).catch((error: unknown) => respondError(res, error));
  };
}

export function createHttpServer(service: ApplicationService, internalToken: string): Server {
  return createServer(createRequestHandler(service, internalToken));
}

async function handleRequest(req: IncomingMessage, res: ServerResponse, service: ApplicationService, internalToken: string): Promise<void> {
  setCorsHeaders(req, res);
  if (req.method === 'OPTIONS') {
    res.writeHead(204).end();
    return;
  }

  const url = new URL(req.url ?? '/', 'http://localhost');

  // 引擎工具回调（架构设计 5.2 / ADR-0006）：只允许本机 loopback + 启动时生成的内部令牌。
  // 这是引擎与平台之间唯一的调用入口，平台侧仍收口到 Core.invoke。
  if (url.pathname === '/internal/tool-call') {
    return handleInternalToolCall(req, res, service, internalToken);
  }

  // 非 API 路径先尝试托管客户端产物（A1.5；未构建时返回 false，继续走协议错误处理）
  if ((req.method === 'GET' || req.method === 'HEAD') && !url.pathname.startsWith('/v1')) {
    if (await serveStatic(url.pathname, res)) return;
  }

  const match = matchRoute(req.method ?? 'GET', url.pathname);
  if (!match) {
    throw new AppError('not_found', `没有这个端点：${req.method ?? 'GET'} ${url.pathname}`, { status: 404 });
  }

  const { route, params } = match;
  const subject = resolveSubject(req.headers);
  const sessionId = params.get('id') ?? '';
  const projectId = url.searchParams.get('projectId') ?? undefined;
  const sessionFilter = url.searchParams.get('sessionId') ?? undefined;

  switch (`${route.spec.method} ${route.spec.path}`) {
    case 'GET /v1/health':
      return sendJson(res, 200, service.health());
    case 'GET /v1/me':
      return sendJson(res, 200, service.me(subject));
    case 'GET /v1/projects':
      return sendJson(res, 200, service.listProjects());
    case 'GET /v1/sessions':
      return sendJson(res, 200, service.listSessions(projectId));
    case 'POST /v1/sessions':
      return sendJson(res, 200, service.createSession(subject, (await readJson(req)) as CreateSessionRequest));
    case 'GET /v1/sessions/{id}':
      return sendJson(res, 200, service.getSession(sessionId));
    case 'GET /v1/sessions/{id}/events':
      return openEventStream(req, res, service, sessionId);
    case 'POST /v1/sessions/{id}/messages':
      return sendJson(res, 202, service.sendMessage(subject, sessionId, (await readJson(req)) as SendMessageRequest));
    case 'POST /v1/sessions/{id}/cancel':
      return sendJson(res, 200, service.cancel(sessionId));
    case 'GET /v1/capabilities':
      return sendJson(res, 200, service.capabilities(subject, sessionFilter));
    case 'GET /v1/executions/{executionId}': {
      const executionId = params.get('executionId') ?? '';
      const receipt = service.execution(executionId);
      if (!receipt) throw new AppError('not_found', `执行记录  不存在`, { status: 404 });
      return sendJson(res, 200, receipt);
    }
    case 'POST /v1/approvals/{approvalId}':
      // 确认卡在 A4（控制面）之后才有
      throw new AppError('not_implemented', '确认卡将在 A4 阶段实现', { status: 501 });
    default:
      throw new AppError('not_found', `未实现的端点：${route.spec.method} ${route.spec.path}`, { status: 404 });
  }
}

function matchRoute(method: string, pathname: string): { route: Route; params: Map<string, string> } | undefined {
  for (const route of ROUTES) {
    if (route.spec.method !== method) continue;
    const matched = route.regex.exec(pathname);
    if (!matched) continue;
    const params = new Map<string, string>();
    route.names.forEach((name, index) => params.set(name, matched[index + 1] ?? ''));
    return { route, params };
  }
  return undefined;
}

function openEventStream(req: IncomingMessage, res: ServerResponse, service: ApplicationService, sessionId: string): void {
  const url = new URL(req.url ?? '/', 'http://localhost');
  const headerValue = req.headers['last-event-id'];
  const raw = url.searchParams.get('since') ?? (Array.isArray(headerValue) ? headerValue[0] : headerValue) ?? '0';
  const parsed = Number.parseInt(String(raw), 10);
  const since = Number.isFinite(parsed) && parsed > 0 ? parsed : 0;

  service.lastSeq(sessionId); // 会话不存在时先 404

  res.writeHead(200, {
    'content-type': 'text/event-stream; charset=utf-8',
    'cache-control': 'no-cache, no-transform',
    connection: 'keep-alive',
    'x-accel-buffering': 'no',
  });
  res.write(': connected\n\n'); // 让客户端立刻知道连接已建立

  const unsubscribe = service.subscribe(sessionId, since, (event) => writeEvent(res, event));
  const heartbeat = setInterval(() => service.heartbeat(sessionId), HEARTBEAT_MS);
  heartbeat.unref();

  req.on('close', () => {
    clearInterval(heartbeat);
    unsubscribe();
    res.end();
  });
}

function writeEvent(res: ServerResponse, event: PlatformEvent): void {
  res.write(`id: ${event.seq}\nevent: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`);
}

async function readJson(req: IncomingMessage): Promise<unknown> {
  const chunks: Buffer[] = [];
  let size = 0;
  for await (const chunk of req) {
    const buffer = chunk as Buffer;
    size += buffer.length;
    if (size > MAX_BODY_BYTES) throw new AppError('payload_too_large', '请求体过大', { status: 413 });
    chunks.push(buffer);
  }
  if (size === 0) return {};
  const text = Buffer.concat(chunks).toString('utf8');
  try {
    return JSON.parse(text);
  } catch {
    throw new AppError('invalid_json', '请求体不是合法 JSON', { status: 400 });
  }
}

function sendJson(res: ServerResponse, status: number, body: unknown): void {
  const payload = JSON.stringify(body);
  res.writeHead(status, { 'content-type': 'application/json; charset=utf-8' });
  res.end(payload);
}

function respondError(res: ServerResponse, error: unknown): void {
  if (res.headersSent) {
    res.end();
    return;
  }
  const appError =
    error instanceof AppError
      ? error
      : new AppError('internal', error instanceof Error ? error.message : '内部错误', { status: 500, retryable: true });
  sendJson(res, appError.status, {
    error: { code: appError.code, message: appError.message, retryable: appError.retryable },
  });
}

function setCorsHeaders(req: IncomingMessage, res: ServerResponse): void {
  // 开发期回显来源，便于桌面端与浏览器本地联调；A4.1 身份落地后按白名单收紧。
  const origin = req.headers.origin;
  if (origin) {
    res.setHeader('access-control-allow-origin', origin);
    res.setHeader('vary', 'origin');
    res.setHeader('access-control-allow-credentials', 'true');
  }
  res.setHeader('access-control-allow-methods', 'GET,POST,OPTIONS');
  res.setHeader('access-control-allow-headers', 'content-type,last-event-id,x-aps-client,authorization');
}

/**
 * 引擎工具回调：POST /internal/tool-call
 * 入参 { sessionId, capabilityId, params, idempotencyKey? }
 * 鉴权：loopback 来源 + 启动时生成的内部令牌（A3.11 会在此基础上加 nonce 轮换与 origin 校验）
 */
async function handleInternalToolCall(
  req: IncomingMessage,
  res: ServerResponse,
  service: ApplicationService,
  internalToken: string,
): Promise<void> {
  if (req.method !== 'POST') {
    throw new AppError('method_not_allowed', '内部回调只接受 POST', { status: 405 });
  }
  const remote = req.socket.remoteAddress ?? '';
  const isLoopback = remote === '127.0.0.1' || remote === '::1' || remote === '::ffff:127.0.0.1';
  if (!isLoopback) {
    throw new AppError('forbidden', '内部回调只允许本机调用', { status: 403 });
  }
  const provided = req.headers['x-aps-internal-token'];
  if (String(Array.isArray(provided) ? provided[0] : (provided ?? '')) !== internalToken) {
    throw new AppError('unauthorized', '内部令牌不正确', { status: 401 });
  }

  const body = (await readJson(req)) as { sessionId?: unknown; capabilityId?: unknown; params?: unknown; idempotencyKey?: unknown; timeoutMs?: unknown };
  const sessionId = typeof body.sessionId === 'string' ? body.sessionId : '';
  const capabilityId = typeof body.capabilityId === 'string' ? body.capabilityId : '';
  if (!sessionId || !capabilityId) {
    throw new AppError('invalid_request', 'sessionId 与 capabilityId 必填', { status: 400 });
  }

  const receipt = await service.invokeFromSession(sessionId, {
    capabilityId,
    params: body.params,
    ...(typeof body.idempotencyKey === 'string' ? { idempotencyKey: body.idempotencyKey } : {}),
    ...(typeof body.timeoutMs === 'number' ? { timeoutMs: body.timeoutMs } : {}),
  });
  sendJson(res, 200, receipt);
}