JX-BPOS/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,
  SchemaValidator,
  ViewGateway,
  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,
  resolvePath,
} from '../domain/step-sequence.ts';

export type FlowRunnerDeps = {
  readonly state: FlowStateStore;
  readonly evaluate: PredicateEvaluator;
  readonly data: TaskDataGateway;
  readonly tools: ToolStepInvoker;
  readonly llm: LlmStepGateway;
  readonly schemas: SchemaValidator;
  readonly views: ViewGateway;
  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)) {
      // 路径优先(模型常按 schema 回嵌套结构),取不到再退回扁平键
      patch[field] = resolvePath(output, path) ?? 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 = resolvePath(output, path) ?? output[path] ?? resolvePath(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,
    });
  }

  /**
   * 按步序位置推视图 —— 声明了 `view` 才推,推的是这一步跑完后的**任务视图**。
   * 推送失败不该把步拖垮:呈现是投影,投影出问题不改变已经发生的事实。
   */
  async function pushStepView(taskId: TaskId, step: FlowStep, activityId: string): Promise<void> {
    if (step.view === undefined) return;
    try {
      await deps.views.push({ taskId, activityId, template: step.view, scope: await deps.data.view(taskId) });
    } catch (error) {
      /*
       * 呈现失败只丢这一帧画面,不影响步序推进 —— 但**必须留痕**:
       * 静默吞掉会让「视图区一直空着」变成一桩查不出原因的怪事。
       */
      await deps.memory.append({
        taskId,
        kind: 'engine-transition',
        title: `视图推送失败(不影响推进):${error instanceof Error ? error.message : String(error)}`,
        detail: { step: step.name, template: step.view },
      });
    }
  }

  /** 回传协议 → 出参:取 `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] = resolvePath(scope, path);
    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);
        await pushStepView(taskId, step, context.activityId);
        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 步失败');

        /*
         * 「出参必过 schema 校验」(AM 12.3)。
         * 解析不出来或不合 schema = **这一步失败了**,走声明的失败走向;
         * 从前是静默当成 `{}` 写下去 —— 一条空记录、一行不报错,
         * 真正的毛病要到下游缺字段时才炸,而且炸在别处。
         */
        const parsed = result.object;
        const check = deps.schemas.validate(step.outputSchema, parsed);
        if (parsed === undefined || !check.ok) {
          const detail = parsed === undefined ? '出参不是 JSON' : check.errors.join('; ');
          await deps.memory.append({
            taskId,
            kind: 'result',
            title: `步 ${index + 1} 出参不合 schema:${detail}`,
            detail: { step: step.name, text: (result.text ?? '').slice(0, 500) },
          });
          return handleFailure(taskId, step, index, position, attempt, `出参不合 schema:${detail}`);
        }

        const output = parsed as Record<string, unknown>;
        await writeOutput(taskId, step, output);
        await pushStepView(taskId, step, context.activityId);
        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': {
        // 异步步:完成条件 = 人事件;时限必配。
        // 发卡本身可能被 Harness 拦下(调人预算触顶等)—— 那是**这一步失败了**,
        // 走它声明的失败走向;不接住就会每一拍重发一次、永远发不出去也永远不报错。
        let card: { id: string };
        try {
          card = await deps.cards.issue({
            taskId,
            ownerId: context.ownerId,
            humanCallCap: context.humanCallCap,
            workOrder: step.workOrder,
            card: step.card,
          });
        } catch (error) {
          return handleFailure(
            taskId,
            step,
            index,
            position,
            attempt,
            error instanceof Error ? error.message : String(error),
          );
        }
        await deps.listeners.register({
          taskId,
          eventType: 'human',
          keyFilter: { cardId: card.id },
          deadline: step.deadline,
        });
        // 发卡的同时把全景推上去:人要在数据上作判断,数据就得摆在卡旁边
        await pushStepView(taskId, step, context.activityId);
        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' };
    },
  };
}