BPOS-TMP-REPO/apps/api/src/app/mastra-agent-runner.ts

/**
 * loop 的 Agent 回路(**开环**)—— `AgentPort` 的 Mastra 实现(M5-2;app 层装配 §5.2)。
 *
 * 与单拍版(`agent-runner.ts`)的分工:
 *   · 单拍版:问一次模型、要一个行动,引擎替它执行 —— 保守、每步可拦;
 *   · **开环版(此处)**:给它一张**工具总线**,让它在一拍内自主连环调工具
 *     (读类真调、写类登记意向、LLM 五件、调人),拍末只交回**这一拍的结论**。
 *
 * 放开的是工具循环,**不是规矩**:
 *   · 能调什么由活动声明定死(工具总线在编译期出清单,白名单在缝里再把一次关);
 *   · 写类工具**过程中一律不落地**,只登记动作意向 —— 放行那一刻才由系统执行;
 *   · 调人由总线执行(发卡 + 立监听),引擎收到 `asked` 就把这一拍挂起,不重复发卡;
 *   · 终局唯放行:Agent 只能 `submit`(提交放行),不能自己宣布完成;
 *   · 用量按 Mastra 回报的 usage 计入任务预算 —— 不计,预算熔断永远不会响。
 *
 * 代价(如实记 debt M5-2):开环下模型的**逐轮**调用不经模型网关的录制回放
 * (tool-calling 往返 `generate` 的 text-in/text-out 抽象承载不了)——
 * 四元绑定与凭证隔离照旧走网关 `resolve`;**工具调用一缝完整**,工具侧录制回放不破。
 */
import { createOpenAICompatible } from '@ai-sdk/openai-compatible';
import { Agent } from '@mastra/core/agent';
import { z } from 'zod';
import type { TaskId } from '@contracts';
import type { AgentAction, AgentTurn, EnabledSet } from '@modules/loop-engine';
import type { ModelGateway } from '@modules/llm';
import type { BusTool } from '@modules/toolbox';
import { ledgerDictionary } from './agent-runner.ts';
import type { WritableLedger } from './agent-runner.ts';
import { buildMastraTools } from './tool-bus.ts';
import type { ToolBusContext, ToolBusPorts } from './tool-bus.ts';

/** 拍末结论 —— Agent 干完活要说清「这一拍算什么」,不能只是聊完就走 */
const TURN_RESULT = z.object({
  kind: z.enum(['start', 'plan', 'disable', 'judge', 'submit', 'idle']).describe('这一拍的结论'),
  itemId: z.string().optional().describe('start / plan / disable 针对哪个计划项'),
  reason: z.string().optional(),
  criterion: z.string().optional().describe('judge:评估的是哪条语义性前提'),
  satisfied: z.boolean().optional().describe('judge:满足与否'),
  rationale: z.string().optional().describe('结论理由 —— 留痕要看,写实'),
  /** 对负责人说的一句人话 —— 每一拍都要带(他正看着这一单) */
  say: z.string().optional().describe('对负责人说的一句人话:这一拍干了什么、拿到什么结论'),
  tag: z.enum(['guidance', 'criterion', 'conclusion']).optional(),
  view: z.string().optional().describe('可选:换成哪张局面模版'),
});

export type MastraAgentRunnerDeps = {
  readonly gateway: ModelGateway;
  /** 活动声明的自定义模型名 */
  readonly modelOf: (taskId: TaskId) => Promise<string>;
  /** 这一单能调什么:工具总线清单(声明 × 目录编译而来) */
  readonly busOf: (taskId: TaskId) => Promise<readonly BusTool[]>;
  /** 三路缝 */
  readonly ports: ToolBusPorts;
  /** 拍内上下文:负责人与人工调用上限 */
  readonly contextOf: (taskId: TaskId) => Promise<Omit<ToolBusContext, 'model'>>;
  /** token 回报:计入任务预算 */
  readonly onUsage: (taskId: TaskId, tokens: number) => Promise<void>;
  /** 一拍内最多几轮工具调用 —— 环外还有 Harness 的预算硬闸,这只是拍内的礼貌上限 */
  readonly maxToolRounds?: number;
};

export function createMastraAgentRunner(deps: MastraAgentRunnerDeps): {
  decide(input: {
    taskId: TaskId;
    narrative: string;
    enabled: EnabledSet;
    scope: Record<string, unknown>;
    writable: WritableLedger;
    /** 开环路径的工具清单来自 `busOf`(它自己会取),这里只为满足同一个端口 */
    tools?: unknown;
  }): Promise<AgentTurn>;
} {
  return {
    async decide({ taskId, narrative, enabled, scope, writable }) {
      const modelName = await deps.modelOf(taskId);
      const base = await deps.contextOf(taskId);
      const bus = await deps.busOf(taskId);
      const trace: { askedCardId?: string } = {};

      const instructions = [
        '你是这一单任务的案例工作者。你有一组工具可用,可以连续调用它们把这一拍的活干完。',
        '',
        '规矩(硬约束,违反即被系统拦下):',
        '① **写类工具不会当场生效** —— 调了只是登记动作意向,放行那一刻系统才真执行。',
        '   所以不要假设世界已经改变,更不要因为「已经调过」就跳过必要的确认。',
        '② **调人(human.ask)会让这一拍就此挂起等回传** —— 只在真需要人裁断/补料时调,',
        '   而且必须把卡填实:title 一句话说清要什么、ask 写明具体问题、basis 给出判断依据。',
        '   一张只写「请处理」的卡等于白发。调完人就结束这一拍,别再调别的。',
        '③ **终局唯放行**:成果齐了就 submit(提交放行),由人验收 —— 你不能自己宣布完成。',
        '④ 干完活必须给出这一拍的结论(见输出格式);拿不准就 idle 并说清卡在哪。',
        '',
        '⑥ **每一拍都要带一句 say**:对负责人说人话(他正看着这一单;只干活不说话,他的对话区是空的)。',
        '⑤ **成果就是台账里的行**:干出来的结论用 ledger.write 逐行写进台账(草稿层),',
        '   不写就等于没做 —— 验收判据读的是台账,出口载荷也从台账推。只准写字典里的地址与列名。',
        '',
        `【案例说明书与记忆】\n${narrative}`,
        '',
        ledgerDictionary(writable),
      ]
        .filter((line) => line !== '')
        .join('\n');

      const endpoint = await deps.gateway.resolve(modelName);
      const provider = createOpenAICompatible({
        name: endpoint.provider,
        apiKey: endpoint.apiKey,
        baseURL: endpoint.baseUrl,
      });

      const agent = new Agent({
        id: `loop-${taskId}`,
        name: '案例工作者',
        instructions,
        model: provider.chatModel(endpoint.model),
        tools: buildMastraTools(bus, deps.ports, { ...base, model: modelName }, trace),
      });

      const prompt = [
        `【当下可为(Enabled 集)】\n${
          enabled.entries
            .map((e) => `- ${e.id} ${e.name}(${e.type}${e.requiresManualStart ? ' · 须你启动' : ''}${e.discretionary ? ' · 裁量项' : ''})`)
            .join('\n') || '(空)'
        }`,
        enabled.inProgress.length === 0
          ? ''
          : `【进行中(已开工,未交成果)】\n${enabled.inProgress.map((e) => `- ${e.id} ${e.name}`).join('\n')}`,
        `【目标缺口】\n${enabled.gaps.join('、') || '(无)'}`,
        enabled.judgements.length === 0
          ? ''
          : `【待你评估的语义性前提】\n${enabled.judgements.map((j) => `- ${j.criterion}:${j.statement}`).join('\n')}`,
        enabled.entries.length === 0 && enabled.inProgress.length === 0
          ? '【注意】当下可为与进行中都空 —— 这是依赖断点,调人或说清卡在哪。'
          : '',
        `【任务数据】\n${JSON.stringify(scope).slice(0, 4000)}`,
        '',
        '干这一拍该干的活(可以连续调工具),最后给出结论。',
      ]
        .filter((line) => line !== '')
        .join('\n');

      try {
        const result = await agent.generate(prompt, {
          maxSteps: deps.maxToolRounds ?? 8,
          structuredOutput: { schema: TURN_RESULT },
        } as never);

        const usage = (result as { usage?: { totalTokens?: number; inputTokens?: number; outputTokens?: number } }).usage;
        const tokens = usage?.totalTokens ?? (usage?.inputTokens ?? 0) + (usage?.outputTokens ?? 0);
        if (tokens > 0) await deps.onUsage(taskId, tokens);

        // 调过人就必须收在 asked:卡已发、监听已立,引擎只收尾挂起(再发一张就是重复调人)
        const object = (result as { object?: unknown }).object;
        const said = sayOf(object);
        if (trace.askedCardId !== undefined) {
          return {
            actions: [{ kind: 'asked', cardId: trace.askedCardId, rationale: '本回合已调人,等回传' }],
            ...(said === undefined ? {} : { say: said }),
          };
        }
        // 开环一拍 = Mastra 在拍内自主连环调工具,拍末只交一个收束行动 —— 包成单元素串
        return { actions: [normalizeTurn(object, enabled)], ...(said === undefined ? {} : { say: said }) };
      } catch (error) {
        /*
         * 这一拍问不出结果(端点不可达、鉴权失败、超时、工具连环炸)——
         * **不是异常路径,是断点的一种**:机器当下决断不了就该调人。
         * 抛出去只会让任务空转:推进器每一拍重试、预算照烧,负责人却什么都看不到。
         *
         * 但**已经调过人**的那一拍例外:卡在世界上了,老老实实收成 asked,
         * 不要再叠一张「模型出错」的卡上去。
         */
        if (trace.askedCardId !== undefined) {
          return { actions: [{ kind: 'asked', cardId: trace.askedCardId, rationale: '本回合已调人(其后模型出错)' }] };
        }
        const detail = error instanceof Error ? error.message : String(error);
        return {
          say: { text: `这一轮我没能决断:${detail}`, tag: 'conclusion' as const },
          actions: [{
          kind: 'human',
          itemId: enabled.entries[0]?.id ?? 'breakpoint',
          workOrder: 'execution',
          card: {
            title: '模型这一轮未能决断,转人裁断',
            reason: detail,
            basis: [
              { k: '当下可为', v: enabled.entries.map((e) => e.name).join('、') || '(空)' },
              { k: '目标缺口', v: enabled.gaps.join('、') || '(无)' },
            ],
            ask: '请判断下一步:接着走哪一项,或补齐缺口后让环继续。',
            fields: ['decision'],
          },
          }],
        };
      }
    },
  };
}

/** 拍末那句交代 —— 每一拍都该有;没给就没有,不替它编 */
function sayOf(object: unknown): { text: string; tag?: 'guidance' | 'criterion' | 'conclusion'; view?: string } | undefined {
  const raw = (object ?? {}) as Record<string, unknown>;
  const text = String(raw['say'] ?? '').trim();
  if (text === '') return undefined;
  const tag = raw['tag'];
  const view = String(raw['view'] ?? '').trim();
  return {
    text,
    ...(tag === 'guidance' || tag === 'criterion' || tag === 'conclusion' ? { tag } : {}),
    ...(view === '' ? {} : { view }),
  };
}

/** 拍末结论归一为 AgentAction;给不出结论就当空拍(idle → judge 留痕),绝不乱动世界 */
function normalizeTurn(object: unknown, enabled: EnabledSet): AgentAction {
  const raw = (object ?? {}) as Record<string, unknown>;
  const kind = String(raw['kind'] ?? 'idle');
  const itemId = String(raw['itemId'] ?? enabled.entries[0]?.id ?? '');
  const rationale = String(raw['rationale'] ?? '');

  switch (kind) {
    case 'start':
    case 'plan':
      return { kind: kind as 'start' | 'plan', itemId };
    case 'disable':
      return { kind: 'disable', itemId, reason: String(raw['reason'] ?? '此单不走这条') };
    case 'judge':
      return {
        kind: 'judge',
        criterion: String(raw['criterion'] ?? ''),
        satisfied: raw['satisfied'] === true,
        rationale,
      };
    case 'submit':
      return { kind: 'submit', rationale: rationale === '' ? '判定成果已齐' : rationale };
    default:
      /*
       * idle:这一拍干了活(工具调用已经落在总线上)但推不动局面。
       * 记成一条 judge 留痕 —— 空拍也要说清卡在哪,不然人看不见环在原地打转。
       */
      return {
        kind: 'judge',
        criterion: '本回合是否推动了局面',
        satisfied: false,
        rationale: rationale === '' ? '模型未给出结论' : rationale,
      };
  }
}