/** * HTTP / SSE 入口(架构设计第 2 节)。 * * 只做协议:路由、请求解析、身份解析、响应与错误包装、事件流推送。 * 业务编排在 app/,控制面在 core/(A4 起)。路由表来自协议里的 ENDPOINT_SPECS, * 保证「文档写了什么端点,代码就实现什么端点」。 */ import { createServer, type IncomingMessage, type Server, type ServerResponse } from 'node:http'; import type { Subject } from '../app/subject.js'; import { ENDPOINT_SPECS, type ApprovalDecisionRequest, type BindingCompleteRequest, 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 { 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(subject, sessionId)); case 'GET /v1/sessions/{id}/events': return openEventStream(req, res, service, sessionId, subject); 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(subject, 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(subject, executionId); if (!receipt) throw new AppError('not_found', `执行记录 不存在`, { status: 404 }); return sendJson(res, 200, receipt); } case 'GET /v1/executions/{executionId}/trace': { // 执行追溯(A4.5):审计链 + 确认记录 + 回执;只有本人能看 const executionId = params.get('executionId') ?? ''; return sendJson(res, 200, service.executionTrace(subject, executionId)); } case 'GET /v1/bindings': return sendJson(res, 200, service.bindings(subject)); case 'POST /v1/bindings/{providerId}/start': // 发起绑定:按 Provider 声明的登录方式给出下一步(oauth 返回跳转地址) return sendJson(res, 200, service.bindingStart(params.get('providerId') ?? '')); case 'POST /v1/bindings/{providerId}/complete': { // 完成绑定:密码只在这一条请求里出现,服务端换完凭据即丢弃(ADR-0012) const providerId = params.get('providerId') ?? ''; const body = (await readJson(req)) as BindingCompleteRequest; return sendJson(res, 200, await service.bindingComplete(subject, providerId, body)); } case 'DELETE /v1/bindings/{providerId}': return sendJson(res, 200, service.unbind(subject, params.get('providerId') ?? '')); case 'POST /v1/approvals/{approvalId}': { // 确认卡响应(协议第 7 节):只有会话发起人能确认;版本或有效期不符会返回 expired const approvalId = params.get('approvalId') ?? ''; const body = (await readJson(req)) as ApprovalDecisionRequest; return sendJson(res, 200, await service.decideApproval(subject, approvalId, body)); } default: throw new AppError('not_found', `未实现的端点:${route.spec.method} ${route.spec.path}`, { status: 404 }); } } function matchRoute(method: string, pathname: string): { route: Route; params: Map } | 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(); 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, subject: Subject, ): 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.assertSessionAccess(subject, sessionId); // 会话不存在或不属于当前身份时先拒 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(subject, sessionId, since, (event: PlatformEvent) => 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 { 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,DELETE,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 { 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); }