《互联网产品的 AI 架构·多能力编排与任务闭环》6.2 人工介入节点

// file: src/orch/human.ts
// tier: L2
// verify: node --test test/orch/human.test.ts

import { AiError } from "../core/errors.ts";

export type ReviewState = "waiting" | "claimed" | "escalated" | "approved" | "rejected" | "expired";
export type ReviewDecision = "approve" | "reject";
export type TimeoutAction = "escalate" | "auto-reject" | "auto-publish" | "hold";
export type RiskLevel = "low" | "medium" | "high";

export interface ReviewPolicy {
  policyVersion: string;
  reviewTimeoutMs: number;
  escalationTimeoutMs: number;
  maxEscalations: number;
  /** 索引 0 是初始处理人。 */
  escalationLadder: readonly string[];
  timeoutAction: TimeoutAction;
  /** 自动发布允许的最高风险等级；高风险永远不自动发布。 */
  autoPublishRiskCeiling: RiskLevel;
}

/** 只追加的审计事件；requestId 用于识别重复提交。 */
export interface ReviewEvent {
  seq: number;
  atMs: number;
  type: "created" | "claimed" | "escalated" | "approved" | "rejected" | "expired" | "routed";
  actor: string;
  from: ReviewState;
  to: ReviewState;
  detail: string;
  policyVersion: string;
  requestId?: string;
}

export interface ReviewTask {
  id: string;
  /** 待发布内容的稳定哈希；发布时必须回传，防止审的是旧稿。 */
  contentHash: string;
  risk: RiskLevel;
  state: ReviewState;
  assignee: string;
  escalationLevel: number;
  deadlineMs: number;
  version: number;
  events: readonly ReviewEvent[];
}

export interface ReviewOutcome {
  task: ReviewTask;
  changed: boolean;
  reason: string;
}

const TERMINAL: readonly ReviewState[] = ["approved", "rejected", "expired"];
const RISK_RANK: Record<RiskLevel, number> = { low: 0, medium: 1, high: 2 };
const TIMEOUT_ACTIONS: readonly TimeoutAction[] = ["escalate", "auto-reject", "auto-publish", "hold"];

function fail(message: string): never {
  throw new AiError("invalid-request", `human: ${message}`);
}

function integer(value: number, field: string, min = 0): void {
  if (!Number.isInteger(value) || value < min) fail(`${field} 必须是 >= ${min} 的整数`);
}

function positive(value: number, field: string): void {
  if (!Number.isFinite(value) || value <= 0) fail(`${field} 必须是正数`);
}

function nonNegative(value: number, field: string): void {
  if (!Number.isFinite(value) || value < 0) fail(`${field} 必须是非负数`);
}

function isTerminal(state: ReviewState): boolean {
  return TERMINAL.includes(state);
}

function validatePolicy(policy: ReviewPolicy): void {
  if (policy.policyVersion.trim() === "") fail("policyVersion 不能为空");
  positive(policy.reviewTimeoutMs, "reviewTimeoutMs");
  positive(policy.escalationTimeoutMs, "escalationTimeoutMs");
  integer(policy.maxEscalations, "maxEscalations");
  if (policy.maxEscalations > 0 && policy.escalationLadder.length === 0) {
    fail("允许升级时 escalationLadder 不能为空");
  }
  if (!TIMEOUT_ACTIONS.includes(policy.timeoutAction)) fail("未知 timeoutAction");
  if (!(policy.autoPublishRiskCeiling in RISK_RANK)) fail("未知 autoPublishRiskCeiling");
}

function append(
  task: ReviewTask,
  event: Omit<ReviewEvent, "seq" | "policyVersion">,
): ReviewTask {
  const full: ReviewEvent = {
    ...event,
    seq: task.events.length + 1,
    policyVersion: task.events[0]?.policyVersion ?? "",
  };
  return { ...task, state: full.to, version: task.version + 1, events: [...task.events, full] };
}

export function createReviewTask(input: {
  id: string;
  contentHash: string;
  risk: RiskLevel;
  nowMs: number;
  policy: ReviewPolicy;
}): ReviewTask {
  if (input.id.trim() === "") fail("task.id 不能为空");
  if (input.contentHash.trim() === "") fail("contentHash 不能为空");
  if (!(input.risk in RISK_RANK)) fail(`未知 risk ${input.risk}`);
  nonNegative(input.nowMs, "nowMs");
  validatePolicy(input.policy);
  const deadlineMs = input.nowMs + input.policy.reviewTimeoutMs;
  return {
    id: input.id,
    contentHash: input.contentHash,
    risk: input.risk,
    state: "waiting",
    assignee: input.policy.escalationLadder[0] ?? "unassigned",
    escalationLevel: 0,
    deadlineMs,
    version: 0,
    events: [{
      seq: 1,
      atMs: input.nowMs,
      type: "created",
      actor: "orchestrator",
      from: "waiting",
      to: "waiting",
      detail: `risk=${input.risk} deadline=${deadlineMs}`,
      policyVersion: input.policy.policyVersion,
    }],
  };
}

/** 同一处理人重复领取幂等，不同处理人不能抢占。 */
export function claimReview(
  task: ReviewTask,
  reviewer: string,
  nowMs: number,
  policy: ReviewPolicy,
): ReviewOutcome {
  if (reviewer.trim() === "") fail("reviewer 不能为空");
  nonNegative(nowMs, "nowMs");
  if (isTerminal(task.state)) return { task, changed: false, reason: "terminal" };
  if (task.state === "claimed") {
    return task.assignee === reviewer
      ? { task, changed: false, reason: "already-claimed" }
      : { task, changed: false, reason: "claimed-by-other" };
  }
  const deadlineMs = nowMs + policy.reviewTimeoutMs;
  const next = append(task, {
    atMs: nowMs,
    type: "claimed",
    actor: reviewer,
    from: task.state,
    to: "claimed",
    detail: `deadline=${deadlineMs}`,
  });
  return { task: { ...next, assignee: reviewer, deadlineMs }, changed: true, reason: "claimed" };
}

/** requestId 相同视为重试；同一处理人重复同一决定也幂等。 */
export function decideReview(
  task: ReviewTask,
  decision: ReviewDecision,
  actor: string,
  nowMs: number,
  requestId?: string,
): ReviewOutcome {
  nonNegative(nowMs, "nowMs");
  if (requestId !== undefined && task.events.some((event) => event.requestId === requestId)) {
    return { task, changed: false, reason: "duplicate-request" };
  }
  if (isTerminal(task.state)) {
    const same = (task.state === "approved" && decision === "approve") ||
      (task.state === "rejected" && decision === "reject");
    return same
      ? { task, changed: false, reason: "duplicate-decision" }
      : { task, changed: false, reason: "conflicting-decision" };
  }
  const late = nowMs > task.deadlineMs ? " late" : "";
  const next = append(task, {
    atMs: nowMs,
    type: decision === "approve" ? "approved" : "rejected",
    actor,
    from: task.state,
    to: decision === "approve" ? "approved" : "rejected",
    detail: `${decision}${late}`,
    ...(requestId === undefined ? {} : { requestId }),
  });
  return { task: next, changed: true, reason: late === "" ? decision : `${decision}-late` };
}

/** 手动升级与系统超时走同一条路径，保证审计口径一致。 */
export function escalateReview(
  task: ReviewTask,
  actor: string,
  nowMs: number,
  policy: ReviewPolicy,
  reason: string,
): ReviewOutcome {
  nonNegative(nowMs, "nowMs");
  if (isTerminal(task.state)) return { task, changed: false, reason: "terminal" };
  if (task.escalationLevel >= policy.maxEscalations) {
    return { task, changed: false, reason: "escalation-exhausted" };
  }
  const level = task.escalationLevel + 1;
  const assignee = policy.escalationLadder[Math.min(level, policy.escalationLadder.length - 1)];
  const next = append(task, {
    atMs: nowMs,
    type: "escalated",
    actor,
    from: task.state,
    to: "escalated",
    detail: `level=${level} assignee=${assignee} reason=${reason}`,
  });
  return {
    task: { ...next, assignee, escalationLevel: level, deadlineMs: nowMs + policy.escalationTimeoutMs },
    changed: true,
    reason: "escalated",
  };
}

/** 升级次数用尽后的最终路由；高风险内容永远停在 hold。 */
export function routeAfterTimeout(
  task: ReviewTask,
  policy: ReviewPolicy,
): { action: "publish" | "reject" | "hold"; reason: string } {
  if (task.risk === "high") return { action: "hold", reason: "high-risk-never-auto-publish" };
  if (policy.timeoutAction === "auto-publish") {
    return RISK_RANK[task.risk] <= RISK_RANK[policy.autoPublishRiskCeiling]
      ? { action: "publish", reason: "low-risk-auto-publish" }
      : { action: "hold", reason: "risk-above-auto-publish-ceiling" };
  }
  if (policy.timeoutAction === "auto-reject") return { action: "reject", reason: "policy-auto-reject" };
  return { action: "hold", reason: `policy-${policy.timeoutAction}` };
}

/** 到期先升级，升级用尽再路由；同一时刻重复调用幂等。 */
export function tickReview(
  task: ReviewTask,
  nowMs: number,
  policy: ReviewPolicy,
): ReviewOutcome {
  nonNegative(nowMs, "nowMs");
  if (isTerminal(task.state)) return { task, changed: false, reason: "terminal" };
  if (nowMs < task.deadlineMs) return { task, changed: false, reason: "not-due" };
  if (task.escalationLevel < policy.maxEscalations) {
    return escalateReview(task, "system:timeout", nowMs, policy, "deadline-exceeded");
  }
  const route = routeAfterTimeout(task, policy);
  const target: ReviewState = route.action === "publish"
    ? "approved"
    : route.action === "reject" ? "rejected" : "expired";
  const next = append(task, {
    atMs: nowMs,
    type: "routed",
    actor: "system:timeout",
    from: task.state,
    to: target,
    detail: `${route.action}:${route.reason}`,
  });
  return { task: next, changed: true, reason: `timeout-${route.action}` };
}