BPOS-MVP/apps/api/src/modules/flow-engine/internal/usecase/flow-runner.ts

/**
 * flow 运行器 —— 照谱执行,零临场调度(编码规范 §5.3;语义出处 AM 05)。
 */
import type { FlowStep, TaskId } from '@contracts';
import type { AdvanceOutcome, FlowEngineService } from '../../contract.ts';
import { FlowDeclarationInvalidError } from '../../contract.ts';
import type {
  CardGateway,
  Clock,
  FlowStateStore,
  ListenerRegistrar,
  LlmStepGateway,
  MemoryRecorder,
  PredicateEvaluator,
  TaskContext,
  TaskDataGateway,
  ToolStepInvoker,
} from '../../port.ts';
import type { FlowPosition } from '../domain/step-sequence.ts';
import {
  checkDeclaration,
  fallbackFields,
  lastSelectedPorts,
  lastStepIsAcceptance,
  lastVerdictPassed,
  failureAction,
  initialPosition,
  markStep,
  nextAction,
} from '../domain/step-sequence.ts';

export type FlowRunnerDeps = {
  readonly state: FlowStateStore;
  readonly evaluate: PredicateEvaluator;
  readonly data: TaskDataGateway;
  readonly tools: ToolStepInvoker;
  readonly llm: LlmStepGateway;
  readonly cards: CardGateway;
  readonly listeners: ListenerRegistrar;
  readonly memory: MemoryRecorder;
  readonly tasks: TaskContext;
  readonly clock: Clock;
};

export function createFlowRunner(deps: FlowRunnerDeps): FlowEngineService {
  async function loaded(taskId: TaskId) {
    const found = await deps.state.load(taskId);
    if (found === undefined) throw new Error(`flow 未装载:${taskId}`);
    return found;
  }

  /** 出参写入落**草稿层**(草稿态);放行前不落事实 */
  async function writeOutput(taskId: TaskId, step: FlowStep, output: Record<string, unknown>): Promise<void> {
    if (step.output === undefined) return;
    const patch: Record<string, unknown> = {};
    for (const [field, path] of Object.entries(step.output.fields)) {
      patch[field] = output[path] ?? output[field];
    }
    // 写哪一行:主键取自出参,取不到再回作用域(起单入参往往就带着主键)
    const scope = await deps.data.view(taskId);
    const key: Record<string, string | number> = {};
    for (const [column, path] of Object.entries(step.output.key)) {
      const value = output[path] ?? scope[path];
      // 主键值只能是标量;取不到就不填,让台账的非空约束在合并时如实报出来
      if (typeof value === 'string' || typeof value === 'number') key[column] = value;
    }
    await deps.data.draftWrite({
      taskId,
      domain: step.output.domain,
      object: step.output.object,
      key,
      patch,
    });
  }

  /** 回传协议 → 出参:取 `values`;非回传形状的人事件按原样当出参 */
  function replyValuesOf(payload: Record<string, unknown>): Record<string, unknown> {
    const values = payload['values'];
    return typeof values === 'object' && values !== null ? (values as Record<string, unknown>) : payload;
  }

  function inputOf(step: FlowStep, scope: Record<string, unknown>): Record<string, unknown> {
    const out: Record<string, unknown> = {};
    for (const [key, path] of Object.entries(step.input)) {
      out[key] = path.split('.').reduce<unknown>((acc, seg) => {
        if (acc === null || acc === undefined || typeof acc !== 'object') return undefined;
        return (acc as Record<string, unknown>)[seg];
      }, scope);
    }
    return out;
  }

  async function runOne(
    taskId: TaskId,
    step: FlowStep,
    index: number,
    position: FlowPosition,
  ): Promise<AdvanceOutcome> {
    const context = await deps.tasks.of(taskId);
    const scope = await deps.data.view(taskId);
    const payload = inputOf(step, scope);
    const attempt = (position.steps[index]?.attempt ?? 0) + 1;

    await deps.memory.append({
      taskId,
      kind: 'action',
      title: `步 ${index + 1} · ${step.label}(${step.kind})`,
      detail: { step: step.name, attempt, input: payload },
    });

    switch (step.kind) {
      case 'tool':
      case 'pure-tool': {
        // 写类工具在 flow 里同样登记为动作意向 —— 触达世界的动作集中在放行时落地
        if (step.kind === 'tool' && (await deps.tools.isWriteEffect(step.tool))) {
          await deps.tools.registerIntent({ taskId, tool: step.tool, input: payload, label: step.label });
          const next = markStep(position, index, { state: 'done', attempt });
          await deps.state.save(taskId, (await loaded(taskId)).model, next);
          return { position: next, status: 'running' };
        }
        const result = await deps.tools.invoke({
          tool: step.tool,
          input: payload,
          activityId: context.activityId,
          taskId,
          stepNo: index,
        });
        if (!result.ok) return handleFailure(taskId, step, index, position, attempt, result.error?.message ?? '工具失败');
        if (result.output !== undefined) await writeOutput(taskId, step, result.output);
        const next = markStep(position, index, { state: 'done', attempt, ...(result.output === undefined ? {} : { output: result.output }) });
        await deps.state.save(taskId, (await loaded(taskId)).model, next);
        return { position: next, status: 'running' };
      }

      case 'llm': {
        // LLM 只以「被调用的工具」出现,不是调度者;模板与出参 schema 随声明版本冻结
        const prompt = step.template.replace(/\{\{\s*([\w.]+)\s*\}\}/g, (_, path: string) => String(payload[path] ?? ''));
        const result = await deps.llm.generate({
          model: step.model,
          prompt,
          outputSchema: step.outputSchema,
          taskId,
          stepNo: index,
        });
        if (!result.ok) return handleFailure(taskId, step, index, position, attempt, result.error ?? 'LLM 步失败');
        const output = (result.object ?? {}) as Record<string, unknown>;
        await writeOutput(taskId, step, output);
        const next = markStep(position, index, { state: 'done', attempt, output });
        await deps.state.save(taskId, (await loaded(taskId)).model, next);
        return { position: next, status: 'running' };
      }

      case 'human': {
        // 异步步:完成条件 = 人事件;时限必配
        const card = await deps.cards.issue({
          taskId,
          ownerId: context.ownerId,
          humanCallCap: context.humanCallCap,
          workOrder: step.workOrder,
          card: step.card,
        });
        await deps.listeners.register({
          taskId,
          eventType: 'human',
          keyFilter: { cardId: card.id },
          deadline: step.deadline,
        });
        const next = markStep(position, index, { state: 'awaiting', attempt });
        await deps.state.save(taskId, (await loaded(taskId)).model, next);
        return {
          position: next,
          status: 'awaiting',
          waitingFor: { kind: 'human', ref: card.id, deadline: step.deadline },
        };
      }
    }
  }

  /** 失败路径**声明给定**,不临场发明 */
  async function handleFailure(
    taskId: TaskId,
    step: FlowStep,
    index: number,
    position: FlowPosition,
    attempt: number,
    error: string,
  ): Promise<AdvanceOutcome> {
    const action = failureAction(step, attempt);
    await deps.memory.append({
      taskId,
      kind: 'result',
      title: `步 ${index + 1} 失败:${error} → ${action.kind}`,
      detail: { step: step.name, attempt, action },
    });

    if (action.kind === 'retry') {
      const next = markStep(position, index, { state: 'pending', attempt, error });
      await deps.state.save(taskId, (await loaded(taskId)).model, next);
      return { position: next, status: 'running' };
    }

    if (action.kind === 'human') {
      const context = await deps.tasks.of(taskId);
      const card = await deps.cards.issue({
        taskId,
        ownerId: context.ownerId,
        humanCallCap: context.humanCallCap,
        workOrder: action.workOrder,
        card: {
          title: `步「${step.label}」重试仍败,转人兜底`,
          reason: error,
          // 兜底的人要交的东西,和这一步**本来该产出的东西**同形:
          // 出参契约是步的契约,不因换谁来做而变(统一执行契约)。
          fields: fallbackFields(step, await deps.data.view(taskId)),
        },
      });
      await deps.listeners.register({
        taskId,
        eventType: 'human',
        keyFilter: { cardId: card.id },
        deadline: 'P3D',
      });
      const next = markStep(position, index, { state: 'awaiting', attempt, error });
      await deps.state.save(taskId, (await loaded(taskId)).model, next);
      return { position: next, status: 'awaiting', waitingFor: { kind: 'human', ref: card.id, deadline: 'P3D' } };
    }

    const next = markStep(position, index, { state: 'failed', attempt, error });
    await deps.state.save(taskId, (await loaded(taskId)).model, next);
    return { position: next, status: 'blocked', blockedReason: error };
  }

  return {
    async load(taskId, model) {
      const issues = checkDeclaration(model.steps);
      if (issues.length > 0) throw new FlowDeclarationInvalidError(issues);
      const position = initialPosition(model.steps);
      await deps.state.save(taskId, model, position);
      await deps.memory.append({
        taskId,
        kind: 'engine-transition',
        title: `装载步序声明:${model.steps.length} 步`,
      });
      return position;
    },

    async position(taskId) {
      return (await deps.state.load(taskId))?.position;
    },

    async advance(taskId) {
      const { model, position } = await loaded(taskId);
      const scope = await deps.data.view(taskId);
      const action = nextAction(model.steps, position, scope, deps.evaluate);

      switch (action.kind) {
        case 'finished': {
          // 末步是验收工单时,终态由其**裁决**给出 —— 完成只出自放行通过。
          // 裁决记在末步的 runtime 上(resume 时落笔),谱走完在这里读出来。
          const released = lastStepIsAcceptance(model.steps) && lastVerdictPassed(position);
          // 出口选择随放行一并确认,落地收敛时照它发射
          return { position, status: 'finished', released, selectedPorts: lastSelectedPorts(position) };
        }
        case 'skip': {
          await deps.memory.append({
            taskId,
            kind: 'engine-transition',
            title: `跳步:${action.step.label} —— ${action.reason}`,
          });
          const next = markStep(position, action.index, { state: 'skipped' });
          await deps.state.save(taskId, model, next);
          return { position: next, status: 'running' };
        }
        case 'await':
          return {
            position,
            status: 'awaiting',
            waitingFor: { kind: 'event', ref: action.step.name, deadline: 'P3D' },
          };
        case 'run':
          return runOne(taskId, action.step, action.index, position);
      }
    },

    /** 事件即步的完成条件 —— 到达即续跑,挂起期间事件不丢 */
    async resume(taskId, event) {
      const { model, position } = await loaded(taskId);
      const index = position.steps.findIndex((s) => s.state === 'awaiting');
      if (index < 0) return { position, status: 'running' };

      const step = model.steps[index];
      await deps.memory.append({
        taskId,
        kind: 'engine-transition',
        title: `事件到达,步 ${index + 1} 完成条件满足(${event.kind})`,
        detail: { ref: event.ref },
      });
      // 放行裁决落笔在这一步的运行时上:谱走完时据此定终局
      const verdict = event.kind === 'human' ? String(event.payload['verdict'] ?? '') : '';
      const chosen = event.payload['selectedPorts'];
      const selectedPorts =
        typeof chosen === 'object' && chosen !== null ? (chosen as Record<string, string>) : undefined;
      if (step !== undefined && event.kind === 'human') {
        // 人事件的载荷是**回传协议**的形状(`{mode, workOrderId, values}`),
        // 出参在 `values` 里;直接把整个载荷当出参写,落进去的会是协议壳而不是答案。
        await writeOutput(taskId, step, replyValuesOf(event.payload));
      }
      const next = markStep(
        position,
        index,
        verdict === '' ? { state: 'done' } : { state: 'done', verdict, ...(selectedPorts === undefined ? {} : { selectedPorts }) },
      );
      await deps.state.save(taskId, model, next);
      return { position: next, status: 'running' };
    },
  };
}