JX-BPOS/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,
  IntakeDesk,
  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;
  /** 唯一的引擎(2026-07-29:活动不再分范式,内部实现都是 Loop) */
  readonly engine: ParadigmEngine;
  readonly memory: MemoryRecorder;
  /** 落地收敛 —— 放行通过的那一刻才碰事实 */
  readonly landing: LandingGateway;
  readonly listeners: ListenerProbe;
  readonly persistence: PersistenceProbe;
  readonly events: TaskEventPublisher;
  readonly notifier: ReassignNotifier;
  readonly interaction: InteractionPhaseSetter;
  /**
   * 受理补全台(2026-07-30 宽进):缺必填时把补全卡发到负责人的任务对话流,
   * 并武装 human 监听等答复 —— 任务先「等人」,补齐了才开跑。
   */
  readonly intake: IntakeDesk;
  readonly clock: Clock;
  readonly ids: IdGenerator;
};


/**
 * 缺了哪些必填 —— 键从 JSON-Schema 的 required 来,标签取 properties 里的 title / label,
 * 没有就用键名。补全卡要给人看的是「客户码」,不是 customerCode。
 */
function missingRequired(
  inputSchema: Record<string, unknown>,
  data: Record<string, unknown>,
): Array<{ key: string; label: string }> {
  const required = Array.isArray(inputSchema['required']) ? (inputSchema['required'] as string[]) : [];
  const props = (inputSchema['properties'] ?? {}) as Record<string, { title?: string; label?: string }>;
  return required
    .filter((key) => data[key] === undefined || data[key] === null || data[key] === '')
    .map((key) => ({ key, label: props[key]?.title ?? props[key]?.label ?? key }));
}

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;
  }

  /** 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);
      /*
       * 受理归一(2026-07-30 宽进,替代旧 AM 12.1 的「缺项不得启动」):
       * 输入**可以不定或为空** —— 缺必填不再拒单,而是任务先落成「等人」,
       * 把缺的字段做成补全卡发给负责人;补齐(或答「没法填」)后才开跑。
       * 收敛点从门口挪进活动:门口拒单,人连差什么都不知道;补全卡上写得明明白白。
       */
      const missing = missingRequired(loaded.inputSchema, input.data);

      const now = deps.clock.now();
      const task: Task = {
        id: taskIdSchema.parse(deps.ids.next()),
        activityId: input.activityId,
        ...(input.workflowId === undefined ? {} : { workflowId: input.workflowId }),
        ...(input.nodeId === undefined ? {} : { nodeId: input.nodeId }),
        ownerId: input.ownerId,
        title: input.title,
        runtime: missing.length > 0 ? 'awaiting-human' : '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 deps.engine.load(task.id, loaded.businessModel);
      await deps.interaction.setPhase(task.id, 'active');
      if (missing.length > 0) {
        await deps.memory.append({
          taskId: task.id,
          kind: 'observation',
          title: `受理归一:缺 ${missing.map((m) => m.key).join('、')} —— 已派补全单`,
          detail: { missing: missing.map((m) => m.key) },
        });
        await deps.intake.askCompletion({ taskId: task.id, ownerId: task.ownerId, title: task.title, fields: missing });
      }
      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 deps.engine.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 };
        }
        /*
         * 落地会**抛**(草稿与正式版冲突、动作意向执行失败…)。抛出去就是灾难:
         * 谱已经走完(status=finished),runtime 却还停在 running —— 推进器每两秒
         * 再来一次,每次白烧一步预算,最后以 `exhausted` 收场,而真正的原因只在
         * stderr 里刷屏(实测:一条试跑单空转 37 拍,界面上什么都看不到)。
         * 落地失败是**终态**,不是可重试的中间态:就地落 blocked,把原因写进任务记忆。
         */
        let landed;
        try {
          landed = await deps.landing.land({
            taskId,
            ownerId: task.ownerId,
            humanCallCap: task.caps.humanCalls,
            selectedPorts: outcome.selectedPorts ?? {},
          });
        } catch (error) {
          const detail = error instanceof Error ? error.message : String(error);
          await deps.memory.append({
            taskId,
            kind: 'observation',
            title: `放行落地失败:${detail}`,
            detail: { reason: detail },
          });
          const blocked = await this.finish(taskId, { kind: 'blocked', reason: `放行落地失败:${detail}` });
          return { task: blocked, verdict, progressed: true };
        }
        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,
      });

      /*
       * 受理补全的答复(宽进):任务还在「等人」且谱一步没走 —— 这时候来的 human 事件
       * 只可能是补全卡(谱没开跑,别的卡还发不出来)。答的值**只收契约声明过的键**,
       * 合并进 data;齐了(或人答了「没法填」)就转 running,交给推进器开跑。
       * 「没法填」不硬拦:人已把话说明,带着缺口进活动,收敛是活动的事。
       */
      if (task.runtime === 'awaiting-human' && event.kind === 'human') {
        const loaded = await deps.activities.load(task.activityId);
        const stillMissing = missingRequired(loaded.inputSchema, task.data);
        if (stillMissing.length > 0) {
          const values = (event.payload['values'] ?? {}) as Record<string, unknown>;
          const declared = new Set(stillMissing.map((m) => m.key));
          const accepted = Object.fromEntries(
            Object.entries(values).filter(([k, v]) => declared.has(k) && typeof v === 'string' && v.trim() !== ''),
          );
          const data = { ...task.data, ...accepted };
          const remain = missingRequired(loaded.inputSchema, data);
          const gaveUp = typeof event.payload['reason'] === 'string' && event.payload['reason'] !== '';
          if (remain.length === 0 || gaveUp) {
            await deps.store.update(taskId, { data, runtime: 'running' });
            await deps.memory.append({
              taskId,
              kind: 'observation',
              title: remain.length === 0 ? '受理归一:补全到齐,开跑' : `受理归一:${remain.map((m) => m.key).join('、')} 没法填 —— 带缺口开跑`,
              detail: { merged: Object.keys(accepted), remain: remain.map((m) => m.key) },
            });
          } else {
            await deps.store.update(taskId, { data });
            await deps.memory.append({
              taskId,
              kind: 'observation',
              title: `受理归一:收到 ${Object.keys(accepted).join('、') || '(空)'},还缺 ${remain.map((m) => m.key).join('、')}`,
              detail: { merged: Object.keys(accepted), remain: remain.map((m) => m.key) },
            });
            await deps.intake.askCompletion({ taskId, ownerId: task.ownerId, title: task.title, fields: remain });
            // 还在等人 —— 不 tick:引擎一步都不该走,谱要等补全齐了才开
            await deps.store.markProgress(taskId, deps.clock.now());
            const held = await require(taskId);
            return { task: held, verdict: { kind: 'ok' }, progressed: false };
          }
          await deps.store.markProgress(taskId, deps.clock.now());
          return this.tick(taskId);
        }
      }

      await deps.engine.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 };