BPOS-TMP-REPO/apps/api/src/modules/task/internal/usecase/task-service.ts

/**
 * task 用例 —— 任务运行时(编码规范 §5.3;语义出处 OS 04 / 4.3,AM 10)。
 *
 * 预算在**环外**强制:每一拍先扣先判,再交范式引擎 —— 活动不自查自守。
 */
import type { StaffId, TaskId } from '@contracts';
import { taskIdSchema } from '@contracts';
import type { StartTaskInput, Task, TaskService, TickOutcome } from '../../contract.ts';
import { TaskAlreadyTerminalError, TaskNotFoundError } from '../../contract.ts';
import type {
  ActivityLoader,
  Clock,
  IdGenerator,
  InteractionPhaseSetter,
  LandingGateway,
  ListenerProbe,
  MemoryRecorder,
  ParadigmEngine,
  PersistenceProbe,
  ReassignNotifier,
  SchemaValidator,
  TaskEventPublisher,
  TaskStore,
} from '../../port.ts';
import { durationMs, judge, terminalOf, ZERO_CONSUMPTION } from '../domain/harness.ts';
import { applyControl, decideTerminal, resumeReadiness } from '../domain/lifecycle.ts';

export type TaskServiceDeps = {
  readonly store: TaskStore;
  readonly activities: ActivityLoader;
  readonly schemas: SchemaValidator;
  /** 范式引擎按 paradigm 分派 —— 对上层不可区分 */
  readonly engines: Readonly<Record<'loop' | 'flow', ParadigmEngine>>;
  readonly memory: MemoryRecorder;
  /** 落地收敛 —— 放行通过的那一刻才碰事实 */
  readonly landing: LandingGateway;
  readonly listeners: ListenerProbe;
  readonly persistence: PersistenceProbe;
  readonly events: TaskEventPublisher;
  readonly notifier: ReassignNotifier;
  readonly interaction: InteractionPhaseSetter;
  readonly clock: Clock;
  readonly ids: IdGenerator;
};

export function createTaskService(deps: TaskServiceDeps): TaskService {
  async function require(taskId: TaskId): Promise<Task> {
    const task = await deps.store.get(taskId);
    if (task === undefined) throw new TaskNotFoundError(taskId);
    return task;
  }

  function engineOf(task: Task): ParadigmEngine {
    return deps.engines[task.paradigm];
  }

  /** Harness 在环外先判:越限即熔断并挂起,处置归治理台 */
  async function harnessCheck(task: Task, stallAfter: string) {
    const lastProgress = (await deps.store.lastProgressAt(task.id)) ?? new Date(task.startedAt);
    const stall = {
      sinceLastProgressMs: deps.clock.now().getTime() - lastProgress.getTime(),
      stallAfterMs: durationMs(stallAfter),
      hasArmedListener: await deps.listeners.hasArmedListener(task.id),
      awaitingHumanReply: task.runtime === 'awaiting-human',
    };
    return judge(task.used, task.caps, stall);
  }

  return {
    async start(input: StartTaskInput): Promise<Task> {
      const loaded = await deps.activities.load(input.activityId);
      // 输入 data 必过 schema 校验,**缺项不得启动**(AM 12.1)
      const check = deps.schemas.validate(loaded.inputSchema, input.data);
      if (!check.ok) throw new Error(`输入不合契约,不得启动:${check.errors.join('; ')}`);

      const now = deps.clock.now();
      const task: Task = {
        id: taskIdSchema.parse(deps.ids.next()),
        activityId: input.activityId,
        paradigm: loaded.paradigm,
        ...(input.workflowId === undefined ? {} : { workflowId: input.workflowId }),
        ...(input.nodeId === undefined ? {} : { nodeId: input.nodeId }),
        ownerId: input.ownerId,
        title: input.title,
        runtime: 'running',
        data: input.data,
        envelope: loaded.envelope,
        caps: loaded.caps,
        used: ZERO_CONSUMPTION,
        startedAt: now.toISOString(),
      };
      await deps.store.insert(task);
      await deps.memory.append({
        taskId: task.id,
        kind: 'engine-transition',
        title: `起单:${task.title}`,
        // message 计入初始观察、全程留痕;**不得扩大授权边界**
        detail: { data: input.data, message: input.message, envelope: loaded.envelope },
      });
      await engineOf(task).load(task.id, loaded.businessModel);
      await deps.interaction.setPhase(task.id, 'active');
      await deps.store.markProgress(task.id, now);
      return task;
    },

    async get(taskId) {
      return require(taskId);
    },

    async listByOwner(ownerId, openOnly = false) {
      return deps.store.listByOwner(ownerId, openOnly);
    },

    async openTasksOf(ownerId) {
      const tasks = await deps.store.listByOwner(ownerId, true);
      return tasks.map((t) => ({ taskId: t.id, title: t.title }));
    },

    async tick(taskId): Promise<TickOutcome> {
      const task = await require(taskId);
      if (task.terminal !== undefined) throw new TaskAlreadyTerminalError(taskId, task.terminal);

      const loaded = await deps.activities.load(task.activityId);
      const verdict = await harnessCheck(task, loaded.stallAfter);
      const terminal = terminalOf(verdict);
      if (terminal !== undefined) {
        const finished = await this.finish(
          taskId,
          terminal === 'exhausted'
            ? { kind: 'exhausted', detail: JSON.stringify(verdict) }
            : { kind: 'stalled', evidence: '该动未动、超限仍无推进' },
        );
        return { task: finished, verdict, progressed: false };
      }

      const startedAt = deps.clock.now().getTime();
      const outcome = await engineOf(task).advance(taskId);
      // 步计量:loop 的一次决策循环 / flow 的一个声明步。
      // 活跃时长按**这一拍真正花掉的时间**计 —— 挂起等人的时间不算活跃,那是世界等待。
      const used = await deps.store.addConsumption(taskId, {
        steps: 1,
        activeMs: Math.max(0, deps.clock.now().getTime() - startedAt),
      });
      await deps.store.markProgress(taskId, deps.clock.now());

      const runtime =
        outcome.status === 'awaiting'
          ? 'awaiting-event'
          : outcome.status === 'blocked'
            ? 'suspended'
            : 'running';
      const updated = await deps.store.update(taskId, { runtime, used });
      await deps.interaction.setPhase(taskId, runtime === 'running' ? 'active' : 'awaiting-event');

      if (outcome.status === 'blocked') {
        const finished = await this.finish(taskId, { kind: 'blocked', reason: outcome.reason ?? '无法推进' });
        return { task: finished, verdict, progressed: true };
      }

      /*
       * 谱走完了。**完成只出自放行通过**(OS 05):
       *   · 放行通过 → 落地收敛(草稿合并 + 动作执行 → 收敛才发射输出口)→ released;
       *   · 未放行(还没裁 / 驳回)→ 不是完成,也**不再空推** —— 挂起等人,
       *     否则推进器会一拍一拍把预算烧到 exhausted,把「没人裁」误报成「跑爆了」。
       */
      if (outcome.status === 'finished') {
        if (outcome.released !== true) {
          const waiting = await deps.store.update(taskId, { runtime: 'awaiting-human' });
          await deps.interaction.setPhase(taskId, 'awaiting-event');
          return { task: waiting, verdict, progressed: false };
        }
        const landed = await deps.landing.land({
          taskId,
          ownerId: task.ownerId,
          humanCallCap: task.caps.humanCalls,
          selectedPorts: outcome.selectedPorts ?? {},
        });
        if (!landed.converged) {
          // 落地未收敛:输出口不得发射,任务受阻待人(落地失败卡已发)
          const blocked = await this.finish(taskId, { kind: 'blocked', reason: '落地未收敛,输出口未发射' });
          return { task: blocked, verdict, progressed: true };
        }
        await deps.memory.append({
          taskId,
          kind: 'engine-transition',
          title: `落地收敛:合并 ${landed.ledgerRowsMerged} 行,发射输出口 ${landed.emittedPorts.join('、') || '—'}`,
        });
        const finished = await this.finish(taskId, { kind: 'released' }, landed.facts);
        return { task: finished, verdict, progressed: true };
      }

      return { task: updated, verdict, progressed: true };
    },

    async deliver(taskId, event): Promise<TickOutcome> {
      const task = await require(taskId);
      if (task.terminal !== undefined) throw new TaskAlreadyTerminalError(taskId, task.terminal);

      await deps.memory.append({
        taskId,
        kind: 'observation',
        title: `事件到达:${event.kind}/${event.ref}`,
        detail: event.payload,
      });
      await engineOf(task).onEvent(taskId, event);
      await deps.store.markProgress(taskId, deps.clock.now());
      return this.tick(taskId);
    },

    /** 控制三语义 —— 叫停留痕为最重档纠正信号 */
    async control(taskId, semantic, by, reason) {
      const task = await require(taskId);
      const outcome = applyControl(semantic);
      await deps.events.publishControl(taskId, semantic, by, reason);
      await deps.memory.append({
        taskId,
        kind: 'control',
        title: `控制事件:${semantic}`,
        detail: { by, reason, correctionSignal: outcome.kind === 'halted' ? 'heaviest' : undefined },
      });

      if (outcome.kind === 'terminated') {
        await this.finish(taskId, { kind: 'terminated', reason: reason ?? '治理台终止' });
        return outcome;
      }
      await deps.store.update(taskId, { runtime: 'suspended', ...(reason === undefined ? {} : { reason }) });
      await deps.interaction.setPhase(taskId, 'awaiting-event');
      void task;
      return outcome;
    },

    async resume(taskId, by) {
      const task = await require(taskId);
      if (task.terminal !== undefined) throw new TaskAlreadyTerminalError(taskId, task.terminal);
      await deps.memory.append({ taskId, kind: 'control', title: '恢复', detail: { by } });
      const updated = await deps.store.update(taskId, { runtime: 'running' });
      await deps.interaction.setPhase(taskId, 'active');
      return updated;
    },

    /** 改派:留痕并通知双方(OS 02) */
    async reassign(taskId, to, by) {
      const task = await require(taskId);
      const updated = await deps.store.update(taskId, { ownerId: to });
      await deps.memory.append({
        taskId,
        kind: 'control',
        title: `改派:${task.ownerId} → ${to}`,
        detail: { by },
      });
      await deps.notifier.notify({ from: task.ownerId, to, taskId, title: task.title, by });
      return updated;
    },

    /** 终态由系统命名;封卷触发蒸馏钩子 */
    async finish(taskId, cause, facts) {
      const task = await require(taskId);
      if (task.terminal !== undefined) return task;

      const { terminal, reason } = decideTerminal(cause);
      const now = deps.clock.now();
      const updated = await deps.store.update(taskId, {
        runtime: 'terminal',
        terminal,
        reason,
        finishedAt: now.toISOString(),
      });

      await deps.memory.append({ taskId, kind: 'terminal', title: `终态:${terminal}`, detail: { reason } });
      await deps.events.publishTerminal(taskId, terminal, reason, facts ?? { flat: {}, rows: [] });
      await deps.interaction.setPhase(taskId, 'terminal');

      // 终态封卷 → 归档 → 蒸馏钩子产出活动记忆候选条目
      await deps.memory.seal(taskId);
      try {
        await deps.memory.distill({
          taskId,
          activityId: task.activityId,
          trigger: 'terminal',
          prompt: `任务「${task.title}」以 ${terminal} 收场(${reason})。归纳可复用的事实、模式、裁决与例外。`,
        });
      } catch {
        /*
         * 蒸馏是**进化管线**的入口,不是终态的一部分:它产出的是待回放待放行的候选增量。
         * 它失败(模型不可达等)绝不能把已经落定的终态掀翻 —— 那会让任务卡在
         * 「实际已终、状态却没落」的中间态,治理台也看不到。
         *
         * 也不能把失败记进任务记忆:此刻记忆**已封卷**,追加会被只追加不改的守卫拒掉
         * (那正是它该做的)。蒸馏的成败属于进化管线的账,不属于这一单的过程留痕。
         */
      }

      return updated;
    },

    async resumeReadiness(taskId) {
      const task = await require(taskId);
      return resumeReadiness({
        'task-memory': await deps.memory.hasHistory(task.id),
        'draft-layer': await deps.persistence.hasDrafts(task.id),
        'runtime-state': await deps.persistence.hasRuntimeState(task.id),
      });
    },
  };
}

export type { StaffId };