aps-agent/server/app/service.ts

/**
 * 应用服务：会话与任务编排、能力查询、事件推送（架构设计第 2 节）。
 *
 * 它持有会话与业务记录，但不做业务规则、不替引擎判断。
 * A1 阶段用内存实现；A3.7 换成持久层后接口保持不变。
 */
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 } from '@aps/contracts';

import { CapabilityRegistry } from '../core/registry/registry.js';
import { createCore, type Core, type InvokeRequest } from '../core/execution/invoke.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 { runTurn } from './turn-runner.js';
import type { Provider } from '@aps/provider-sdk';
import { 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;
}

export class ApplicationService {
  private readonly projects = new ProjectStore();
  private readonly sessions = new SessionStore();
  private readonly version: string;
  /** 能力目录与执行入口：能力清单来自这里，执行只有一个出口 */
  readonly registry = new CapabilityRegistry();
  readonly core: Core;
  private readonly runtime: AgentRuntime;
  /** 每个平台会话一个运行时会话（ADR-0006：首版一进程一会话） */
  private readonly runtimeSessions = new Map<string, RuntimeSession>();

  constructor(options: ApplicationServiceOptions = {}) {
    this.version = options.version ?? '0.1.0';
    this.core = createCore({ registry: this.registry });
    this.runtime = options.runtime ?? createFakeRuntime({ planner: plannerToFakeTurn(createRulePlanner()) });
  }

  /** 注册 Provider（A5 接入 APS、A8 接入第二个系统时调用） */
  async registerProvider(provider: Provider): Promise<void> {
    await this.registry.register(provider);
  }

  /** 执行记录查询：GET /v1/executions/{id} */
  execution(executionId: string) {
    return this.core.executions.get(executionId)?.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(sessionId: string): SessionDetailResponse {
    const record = this.sessions.require(sessionId);
    return { session: record.summary, messages: record.messages };
  }

  capabilities(subject: Subject, sessionId?: string): CapabilitiesResponse {
    if (sessionId) this.sessions.require(sessionId); // 会话不存在直接 404
    // A4.1 起在这里按身份与策略过滤；当前返回能力目录里的全部能力
    void subject;
    return {
      capabilities: this.registry.list().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.sessions.require(sessionId);
    void subject; // A4.1 起用于策略判定，A1 阶段只做存在性校验

    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 = record.accepted.get(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');
    record.accepted.set(request.idempotencyKey, { messageId, runId });
    record.messages.push({ 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);
    return 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 }),
    });
  }
  cancel(sessionId: string): CancelResponse {
    const record = this.sessions.require(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.runtimeSessions.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(sessionId: string, since: number, listener: (event: PlatformEvent) => void): () => void {
    return this.sessions.require(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<void> {
    const sessionId = record.summary.sessionId;
    const assistantMessageId = `${messageId}-reply`;
    try {
      const runtimeSession = await this.runtimeSessionFor(record);
      const result = await runTurn({
        runtimeSession,
        sessionId,
        runId,
        assistantMessageId,
        text,
        signal,
        publish: (event) => {
          record.bus.publish(event);
        },
        invoke: (capabilityId, args, idempotencyKey) =>
          this.invokeFromSession(sessionId, { capabilityId, params: args, idempotencyKey }),
      });
      if (result.text !== '') {
        record.messages.push({ messageId: assistantMessageId, role: 'assistant', ts: Date.now(), text: result.text });
      }
    } catch (error) {
      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 async runtimeSessionFor(record: SessionRecord): Promise<RuntimeSession> {
    const sessionId = record.summary.sessionId;
    const existing = this.runtimeSessions.get(sessionId);
    if (existing) return existing;
    const created = await this.runtime.createSession({
      sessionId,
      subject: { userId: record.owner.userId, tenantId: record.owner.tenantId },
      projectId: record.summary.projectId,
      tools: toToolDescriptors(this.registry.list()),
      history: record.messages,
    });
    this.runtimeSessions.set(sessionId, created);
    return created;
  }
}