/** * 应用服务:会话与任务编排、能力查询、事件推送(架构设计第 2 节)。 * * 它持有会话与业务记录,但不做业务规则、不替引擎判断。 * A1 阶段用内存实现;A3.7 换成持久层后接口保持不变。 */ import { randomBytes } from 'node:crypto'; import { PROTOCOL_VERSION, type CapabilitiesResponse, type CancelResponse, type CreateSessionRequest, type CreateSessionResponse, type HealthResponse, type MeResponse, type PlatformEvent, type ProjectListResponse, type SendMessageRequest, type SendMessageResponse, type SessionDetailResponse, type SessionListResponse, type ApprovalDecisionRequest, type ApprovalDecisionResponse, type ExecutionTraceResponse } from '@aps/contracts'; import { CapabilityRegistry } from '../core/registry/registry.js'; import { createCore, type Core, type InvokeRequest } from '../core/execution/invoke.js'; import { createControlPlane, type ControlPlane } from '../core/control-plane/control-plane.js'; import { IdentityContext } from '../core/identity/context.js'; import { CredentialBroker } from '../core/credentials/broker.js'; import { PendingActionStore } from '../core/approval/store.js'; import { AuditLog } from '../core/audit/log.js'; import { createFakeRuntime } from '../agent/fake-runtime.js'; import type { AgentRuntime, RuntimeSession } from '../agent/runtime.js'; import { createRulePlanner, plannerToFakeTurn } from '../agent/rule-planner.js'; import { toToolDescriptors } from '../agent/tools.js'; import { RuntimeSupervisor, type SupervisorOptions } from '../agent/process-manager.js'; import type { SessionRepository } from '../core/storage/sessions.js'; import { capabilityFingerprint, isCapabilityVisible, visibleCapabilities, type CapabilityVisibility, } from '../core/registry/visibility.js'; import { runTurn } from './turn-runner.js'; import { buildContext, type ContextBudget } from './context-builder.js'; import type { ContextRepository } from '../core/storage/context.js'; import { PLATFORM_SYSTEM_PROMPT } from './system-prompt.js'; import type { Provider } from '@aps/provider-sdk'; import { AppError, badRequest, conflict } from './errors.js'; import { ProjectStore } from './projects.js'; import { newId, SessionStore, type SessionRecord } from './sessions.js'; import type { Subject } from './subject.js'; export interface ApplicationServiceOptions { version?: string; /** 引擎实现:默认是假引擎(A3.3 起可换成 Pi 子进程) */ runtime?: AgentRuntime; /** 会话持久化(A3.7):不传则纯内存,测试与临时运行用 */ sessionRepository?: SessionRepository; /** 能力可见性规则(A3.5):不传则开发期默认全部可见 */ capabilityVisibility?: CapabilityVisibility; /** 上下文检查点(A3.8):不传则不落检查点 */ contextRepository?: ContextRepository; /** 上下文预算覆盖(测试用) */ contextBudget?: Partial; /** 引擎进程管理(A3.10):不传则用默认值自建 */ supervisor?: RuntimeSupervisor; /** 进程管理参数覆盖(测试用) */ supervisorOptions?: Partial; /** A4 控制面:五块(身份、策略、凭据、确认、审计)绑在一起的入口;不传则用内存实现自建 */ controlPlane?: ControlPlane; /** 单独覆盖控制面的某一块(落库版由 server/index.ts 注入) */ identity?: IdentityContext; credentials?: CredentialBroker; approvals?: PendingActionStore; audit?: AuditLog; /** 确认卡有效期(测试用) */ approvalTtlMs?: number; } export class ApplicationService { private readonly projects = new ProjectStore(); private readonly sessions: SessionStore; private readonly version: string; /** 能力目录与执行入口:能力清单来自这里,执行只有一个出口 */ readonly registry = new CapabilityRegistry(); readonly core: Core; private readonly runtime: AgentRuntime; private readonly visibility: CapabilityVisibility; private readonly contextRepository: ContextRepository | undefined; private readonly contextBudget: Partial | undefined; /** 已经注入过上下文的运行时会话:同一运行时会话只注入一次(它自己有原生上下文) */ private readonly injectedRuntimeSessions = new Set(); /** 每个平台会话一个运行时会话(ADR-0006:首版一进程一会话) */ /** 引擎进程管理:空闲回收、并发上限、崩溃退避重启(A3.10) */ readonly supervisor: RuntimeSupervisor; /** A4 控制面:策略、身份、凭据、确认、审计的入口 */ readonly controlPlane: ControlPlane; constructor(options: ApplicationServiceOptions = {}) { this.version = options.version ?? '0.1.0'; this.sessions = new SessionStore(options.sessionRepository); this.visibility = options.capabilityVisibility ?? { defaultAllow: true, denyWrite: false, grants: [] }; this.contextRepository = options.contextRepository; this.contextBudget = options.contextBudget; this.supervisor = options.supervisor ?? new RuntimeSupervisor(options.supervisorOptions ?? {}); this.controlPlane = options.controlPlane ?? createControlPlane({ identity: options.identity ?? new IdentityContext(), credentials: options.credentials ?? new CredentialBroker({ masterKey: randomBytes(32) }), approvals: options.approvals ?? new PendingActionStore(), audit: options.audit ?? new AuditLog(), isVisible: (capability, subject) => isCapabilityVisible(capability, this.visibility, subject), ...(options.approvalTtlMs === undefined ? {} : { approvalTtlMs: options.approvalTtlMs }), }); this.core = createCore({ registry: this.registry, controlPlane: this.controlPlane }); this.runtime = options.runtime ?? createFakeRuntime({ planner: plannerToFakeTurn(createRulePlanner()) }); } /** 注册 Provider(A5 接入 APS、A8 接入第二个系统时调用) */ async registerProvider(provider: Provider): Promise { await this.registry.register(provider); } /** 上下文检查点(A3.8):平台保存的摘要与来源 */ checkpoint(sessionId: string) { return this.contextRepository?.get(sessionId); } /** 执行记录查询:GET /v1/executions/{id};待确认中的执行也能查到(回执 status = pending-confirmation) */ execution(subject: Subject, executionId: string) { const record = this.core.executions.get(executionId); if (record) { if (record.subjectUserId !== subject.userId) throw new AppError('forbidden', '这是别人的执行记录', { status: 403 }); return record.receipt; } const action = this.core.approvals?.byExecution(executionId); if (!action) return undefined; if (action.ownerUserId !== subject.userId) throw new AppError('forbidden', '这是别人的执行记录', { status: 403 }); return action.receipt; } /** 执行追溯(A4.5):按 executionId 追到人、参数摘要、审批与回执;只有本人能看 */ executionTrace(subject: Subject, executionId: string): ExecutionTraceResponse { const record = this.core.executions.get(executionId); const action = this.core.approvals?.byExecution(executionId); if (!record && !action) throw new AppError('not_found', '没有这条执行记录', { status: 404 }); const ownerUserId = record?.subjectUserId ?? action?.ownerUserId ?? ''; if (ownerUserId !== subject.userId) throw new AppError('forbidden', '这是别人的执行记录', { status: 403 }); const entries = this.controlPlane.audit.trace(executionId).map((entry) => ({ seq: entry.seq, ts: entry.ts, event: entry.event, actorUserId: entry.actorUserId, ...(entry.capabilityId === undefined ? {} : { capabilityId: entry.capabilityId }), ...(entry.providerId === undefined ? {} : { providerId: entry.providerId }), ...(entry.paramsDigest === undefined ? {} : { paramsDigest: entry.paramsDigest }), ...(entry.decision === undefined ? {} : { decision: entry.decision }), ...(entry.detail === undefined ? {} : { detail: entry.detail }), })); const receipt = record?.receipt ?? action?.receipt; return { executionId, approval: action === undefined ? null : { approvalId: action.approvalId, status: action.status, ...(action.decision === undefined ? {} : { decision: action.decision }), ...(action.decidedBy === undefined ? {} : { decidedBy: action.decidedBy }), ...(action.decidedAt === undefined ? {} : { decidedAt: action.decidedAt }), revision: action.revision, expiresAt: action.expiresAt, }, entries, chainOk: this.controlPlane.audit.verify().ok, ...(receipt === undefined ? {} : { receipt }), }; } /** 确认卡响应(协议第 7 节):校验归属与版本 → 签发一次性授权 → 执行 → 推事件 */ async decideApproval(subject: Subject, approvalId: string, body: ApprovalDecisionRequest): Promise { if (body.decision !== 'approve' && body.decision !== 'reject') throw badRequest('decision 只能是 approve 或 reject'); if (typeof body.idempotencyKey !== 'string' || body.idempotencyKey.length < 8) throw badRequest('idempotencyKey 至少 8 个字符'); if (!Number.isInteger(body.revision) || body.revision < 0) throw badRequest('revision 必须是非负整数'); const pending = this.core.approvals?.get(approvalId); if (pending) this.assertSessionAccess(subject, pending.sessionId); // 别人的会话连看都不给看 const result = await this.core.approve(subject, approvalId, { decision: body.decision, revision: body.revision, idempotencyKey: body.idempotencyKey, ...(body.note === undefined ? {} : { note: body.note }), }); if (result.kind === 'not-found') throw new AppError('not_found', '确认卡不存在', { status: 404 }); if (result.kind === 'forbidden') throw new AppError('forbidden', '只有会话发起人可以确认这次动作', { status: 403 }); const action = this.core.approvals?.get(approvalId); const record = action ? this.sessions.get(action.sessionId) : undefined; if (record) { record.bus.publish({ sessionId: record.summary.sessionId, type: 'approval.resolved', data: { approvalId, decision: result.decision } }); if (result.receipt && result.receipt.status !== 'pending-confirmation') { record.bus.publish({ sessionId: record.summary.sessionId, type: 'execution.updated', data: { executionId: result.receipt.executionId, status: result.receipt.status, ...(result.receipt.errorKind === undefined ? {} : { errorKind: result.receipt.errorKind }), }, }); } } return { approvalId, decision: result.decision, ...(result.receipt === undefined ? {} : { receipt: result.receipt }) }; } /** Core.invoke:工具调用的唯一出口(引擎侧经工具桥调用,见 A3.4) */ async invoke(request: InvokeRequest) { return this.core.invoke(request); } health(): HealthResponse { return { status: 'ok', version: this.version, protocolVersion: PROTOCOL_VERSION }; } me(subject: Subject): MeResponse { return { userId: subject.userId, tenantId: subject.tenantId, displayName: subject.displayName, clientKind: subject.clientKind, }; } listProjects(): ProjectListResponse { return { projects: this.projects.list() }; } listSessions(projectId?: string): SessionListResponse { return { sessions: this.sessions.list(projectId) }; } createSession(subject: Subject, request: CreateSessionRequest): CreateSessionResponse { const scope = request.scope; if (scope !== 'personal' && scope !== 'project') throw badRequest('scope 只能是 personal 或 project'); if (scope === 'project') { const projectId = request.projectId; if (!projectId) throw badRequest('项目会话必须带 projectId(ADR-0003)'); if (!this.projects.get(projectId)) throw badRequest(`项目 ${projectId} 不存在`); } const record = this.sessions.create({ scope, projectId: request.projectId, title: request.title }, subject); return { session: record.summary }; } getSession(subject: Subject, sessionId: string): SessionDetailResponse { const record = this.assertSessionAccess(subject, sessionId); return { session: record.summary, messages: record.messages }; } /** 这个身份能看到的能力:能力清单与引擎工具列表都走这里,保证两者一致 */ private visibleCapabilitiesFor(subject: Subject) { return visibleCapabilities(this.registry.list(), this.visibility, { userId: subject.userId, tenantId: subject.tenantId }); } /** 会话归属校验:不是自己的会话一律拒绝(协作归业务系统,见 D9) */ assertSessionAccess(subject: Subject, sessionId: string): SessionRecord { const record = this.sessions.require(sessionId); if (record.owner.userId !== subject.userId) { throw new AppError('forbidden', '这是别人的会话', { status: 403 }); } return record; } capabilities(subject: Subject, sessionId?: string): CapabilitiesResponse { if (sessionId) this.assertSessionAccess(subject, sessionId); return { capabilities: this.visibleCapabilitiesFor(subject).map((capability) => { const providerId = this.registry.resolve(capability.id)?.providerId ?? ''; return { id: capability.id, providerId, title: capability.title ?? capability.description, purpose: capability.description, effect: capability.effect, inputSchema: capability.inputSchema, ...(capability.synonyms === undefined ? {} : { synonyms: capability.synonyms }), }; }), }; } sendMessage(subject: Subject, sessionId: string, request: SendMessageRequest): SendMessageResponse { const record = this.assertSessionAccess(subject, sessionId); if (typeof request.text !== 'string' || request.text.trim() === '') throw badRequest('text 不能为空'); if (typeof request.idempotencyKey !== 'string' || request.idempotencyKey.length < 8) { throw badRequest('idempotencyKey 至少 8 个字符'); } // 幂等重试优先于「会话忙」判定:同一 key 的重试必须返回原结果,不能报冲突 const cached = this.sessions.findIdempotencyKey(sessionId, request.idempotencyKey); if (cached) return { accepted: true, messageId: cached.messageId, runId: cached.runId }; if (record.state !== 'idle') throw conflict('上一个运行还没结束,先取消或等它完成'); const messageId = newId('m'); const runId = newId('r'); this.sessions.rememberIdempotencyKey(sessionId, request.idempotencyKey, messageId, runId); this.sessions.appendMessage(sessionId, { messageId, role: 'user', ts: Date.now(), text: request.text }); record.bus.publish({ sessionId, runId, type: 'message.user', data: { messageId, text: request.text } }); const abort = new AbortController(); record.runs.set(runId, { runId, abort }); record.state = 'running'; void this.runTurn(record, runId, messageId, request.text, abort.signal); return { accepted: true, messageId, runId }; } /** 引擎回调入口(A3.11 工具桥的平台侧):按会话归属者发起调用,身份不由调用方自带 */ async invokeFromSession( sessionId: string, input: { capabilityId: string; params: unknown; idempotencyKey?: string; timeoutMs?: number }, ) { const record = this.sessions.require(sessionId); const receipt = await this.core.invoke({ subject: { userId: record.owner.userId, tenantId: record.owner.tenantId }, sessionId, projectId: record.summary.projectId, capabilityId: input.capabilityId, params: input.params, ...(input.idempotencyKey === undefined ? {} : { idempotencyKey: input.idempotencyKey }), ...(input.timeoutMs === undefined ? {} : { timeoutMs: input.timeoutMs }), }); // 写操作被策略挡在确认卡上:把确认卡推给界面(协议第 7 节第 2 步) if (receipt.status === 'pending-confirmation' && receipt.approvalId !== undefined) { const action = this.core.approvals?.get(receipt.approvalId); if (action) { record.bus.publish({ sessionId, type: 'approval.requested', data: { approvalId: action.approvalId, capabilityId: action.capabilityId, paramsDigest: action.paramsDigest, impactSummary: action.impactSummary, revision: action.revision, expiresAt: action.expiresAt, }, }); } } return receipt; } cancel(subject: Subject, sessionId: string): CancelResponse { const record = this.assertSessionAccess(subject, sessionId); if (record.state !== 'running') return { state: 'idle' }; record.state = 'cancelling'; // 先让界面看到「正在取消」,再中止引擎与执行 const activeRun = [...record.runs.values()][0]; record.bus.publish({ sessionId, ...(activeRun === undefined ? {} : { runId: activeRun.runId }), type: 'run.state', data: { state: 'cancelling' }, }); // 取消:让引擎停在这一轮,回合自然结束(资源等空闲回收或失败回收) this.supervisor.get(sessionId)?.abort(); for (const run of record.runs.values()) run.abort.abort(); return { state: 'cancelling' }; } /** 事件流保活:无事件时由入口层定时调用(协议文档第 5 节) */ heartbeat(sessionId: string): void { this.sessions.require(sessionId).bus.publish({ sessionId, type: 'heartbeat', data: {} }); } subscribe(subject: Subject, sessionId: string, since: number, listener: (event: PlatformEvent) => void): () => void { return this.assertSessionAccess(subject, sessionId).bus.subscribe(since, listener); } lastSeq(sessionId: string): number { return this.sessions.require(sessionId).bus.lastSeq; } private async runTurn(record: SessionRecord, runId: string, messageId: string, text: string, signal: AbortSignal): Promise { const sessionId = record.summary.sessionId; const assistantMessageId = `${messageId}-reply`; try { const runtimeSession = await this.runtimeSessionFor(record); // 上下文由平台构建:新起的运行时会话要带上「摘要 + 最近若干轮」,已有原生上下文的会话不重复注入 const context = this.buildTurnContext(record, messageId); const shouldInject = !this.injectedRuntimeSessions.has(runtimeSession.id); this.injectedRuntimeSessions.add(runtimeSession.id); const promptText = shouldInject && context.prefix !== '' ? `${context.prefix}${text}` : text; const result = await runTurn({ runtimeSession, sessionId, runId, assistantMessageId, text: promptText, signal, publish: (event) => { record.bus.publish(event); }, invoke: (capabilityId, args, idempotencyKey) => this.invokeFromSession(sessionId, { capabilityId, params: args, idempotencyKey }), }); if (result.text !== '') { this.sessions.appendMessage(sessionId, { messageId: assistantMessageId, role: 'assistant', ts: Date.now(), text: result.text }); } // 正常回合:清零连续失败计数;错误回合:回收后按退避重建(A3.10) if (result.finishReason === 'error') { await this.supervisor.evict(sessionId, 'failed', '回合以错误收尾'); } else { this.supervisor.markHealthy(sessionId); this.supervisor.touch(sessionId); } // 检查点:把这次构建出的摘要与来源落库,摘要与原始消息都保留(实施设计 1.3) this.contextRepository?.save({ sessionId, summary: context.summary, summarySourceIds: context.summarySourceIds, includedMessageIds: context.includedMessageIds, lastMessageId: messageId, pinned: [], estimatedChars: context.estimatedChars, }); } catch (error) { await this.supervisor.evict(sessionId, 'failed', '引擎调用抛错').catch(() => undefined); record.bus.publish({ sessionId, runId, type: 'error', data: { code: 'engine.failed', message: error instanceof Error ? error.message : '引擎执行失败', retryable: true }, }); } finally { record.runs.delete(runId); if (record.state !== 'idle') record.state = 'idle'; } } /** 构建这一轮要注入的上下文(排除刚写入的当前用户消息) */ private buildTurnContext(record: SessionRecord, currentMessageId: string) { const history = record.messages.filter((message) => message.messageId !== currentMessageId); const executions = this.core.executions.listBySession(record.summary.sessionId, 6).map((item) => ({ capabilityId: item.capabilityId, status: item.status, executionId: item.executionId, paramsDigest: item.paramsDigest, createdAt: item.createdAt, })); return buildContext({ messages: history, executions, ...(this.contextBudget === undefined ? {} : { budget: this.contextBudget }), }); } /** 取(或创建)平台会话对应的运行时会话:工具来自能力目录,历史来自平台记录 */ private async runtimeSessionFor(record: SessionRecord): Promise { const sessionId = record.summary.sessionId; const capabilities = this.visibleCapabilitiesFor(record.owner); // 身份或工具集变了就重建:工具集不能跨身份复用(架构设计 5.1「按身份给工具」) const fingerprint = `${record.owner.userId}|${capabilityFingerprint(capabilities)}`; return this.supervisor.acquire(sessionId, fingerprint, () => this.runtime.createSession({ sessionId, subject: { userId: record.owner.userId, tenantId: record.owner.tenantId }, projectId: record.summary.projectId, system: PLATFORM_SYSTEM_PROMPT, tools: toToolDescriptors(capabilities), history: record.messages, }), ); } /** 进程管理统计(观测与用例用) */ processStats() { return this.supervisor.stats(); } }