《互联网产品的 AI 架构·多能力编排与任务闭环》5.1 失败分类与处理策略

// file: src/orch/failure.ts
// tier: L1
// verify: node --test test/orch/failure.test.ts

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

/**
 * 四类失败对应四种动作。分类不是给错误起名字，而是决定下游还能不能继续：
 * retryable 重试、skippable 跳过、degradable 降级、abort 中止。
 */
export type FailureClass = "retryable" | "skippable" | "abort" | "degradable";
export type FailureAction = "retry" | "skip" | "abort" | "degrade";

/** 分类器需要的全部判定字段，缺一项就可能把可恢复失败误判成中止。 */
export interface FailureSignal {
  nodeId: string;
  kind: AiErrorKind;
  /** 该节点的输出是否为下游必需。 */
  required: boolean;
  /** 重复执行是否安全；有副作用的动作必须显式声明。 */
  idempotent: boolean;
  /** 是否已有可交付的降级产物，例如模板摘要、缓存答案。 */
  hasFallback: boolean;
  /** 已尝试次数，含首次。 */
  attempts: number;
  /** 允许的最大尝试次数，首次加重试。 */
  maxAttempts: number;
  /** 编排级剩余重试预算，按次数计。 */
  retryBudgetLeft: number;
  /** 不可回滚的最终动作，例如发布、下单、发消息。 */
  finalAction: boolean;
  /** 提供方建议的等待时间，只对限流与超时有意义。 */
  retryAfterMs?: number;
}

export interface FailureDecision {
  nodeId: string;
  failureClass: FailureClass;
  action: FailureAction;
  /** 稳定原因码，便于日志与指标聚合。 */
  reason: string;
  /** 动作之后下游是否继续执行。 */
  continueDownstream: boolean;
  /** 本次动作是否消耗全局重试预算。 */
  consumesBudget: boolean;
  /** 是否需要人工复核或对账。 */
  needsReconciliation: boolean;
  /** 用户可见标记强度。 */
  userNotice: "none" | "notice" | "banner" | "block";
  retryAfterMs?: number;
}

export interface FailurePolicy {
  action: FailureAction;
  continueDownstream: boolean;
  consumesBudget: boolean;
  needsReconciliation: boolean;
  userNotice: "none" | "notice" | "banner" | "block";
}

/** 四类失败的默认处置，代码与文档引用同一张表，避免口径漂移。 */
export const FAILURE_PLAYBOOK: Record<FailureClass, FailurePolicy> = {
  retryable: {
    action: "retry",
    continueDownstream: true,
    consumesBudget: true,
    needsReconciliation: false,
    userNotice: "none",
  },
  skippable: {
    action: "skip",
    continueDownstream: true,
    consumesBudget: false,
    needsReconciliation: false,
    userNotice: "notice",
  },
  degradable: {
    action: "degrade",
    continueDownstream: true,
    consumesBudget: false,
    needsReconciliation: false,
    userNotice: "banner",
  },
  abort: {
    action: "abort",
    continueDownstream: false,
    consumesBudget: false,
    needsReconciliation: false,
    userNotice: "block",
  },
};

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

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

/** 复用网关的错误表，避免编排层维护第二份「哪些错误可重试」。 */
function retryableKind(kind: AiErrorKind): boolean {
  return new AiError(kind, "failure: classify").retryable;
}

function validateSignal(signal: FailureSignal): void {
  if (signal.nodeId.trim() === "") fail("nodeId 不能为空");
  integer(signal.attempts, "attempts", 1);
  integer(signal.maxAttempts, "maxAttempts", 1);
  integer(signal.retryBudgetLeft, "retryBudgetLeft");
  if (signal.attempts > signal.maxAttempts) {
    fail(`${signal.nodeId}: attempts 不能超过 maxAttempts`);
  }
  if (signal.retryAfterMs !== undefined && (!Number.isFinite(signal.retryAfterMs) || signal.retryAfterMs < 0)) {
    fail("retryAfterMs 必须是 >= 0 的有限数");
  }
}

/**
 * 判定顺序本身就是策略：安全阻断优先于重试，幂等边界优先于降级，
 * 最后才看节点是否必需。顺序一变，同一组信号会得到不同动作。
 */
export function classifyFailure(signal: FailureSignal): FailureDecision {
  validateSignal(signal);
  const attemptsLeft = signal.maxAttempts - signal.attempts;
  const transient = retryableKind(signal.kind);
  const canRetry = transient &&
    attemptsLeft > 0 &&
    signal.retryBudgetLeft > 0 &&
    signal.idempotent;

  if (signal.kind === "content-blocked") {
    if (signal.required) {
      return decision(signal, "abort", "safety-block", false, "block");
    }
    return decision(signal, "skippable", "safety-drop", false, "notice");
  }

  if (canRetry) {
    const result = decision(signal, "retryable", "transient-retry", false, "none");
    if (signal.retryAfterMs !== undefined) result.retryAfterMs = signal.retryAfterMs;
    return result;
  }

  // 最终动作即使可重试也不能自动重放：重复发布或重复下单的代价高于失败。
  if (!signal.idempotent && signal.finalAction) {
    return decision(signal, "abort", "non-idempotent-side-effect", true, "block");
  }

  const attemptsExhausted = attemptsLeft <= 0;
  const budgetExhausted = transient && !attemptsExhausted && signal.retryBudgetLeft <= 0;
  const degradeReason = budgetExhausted
    ? "retry-budget-exhausted"
    : attemptsExhausted && transient
      ? "attempts-exhausted"
      : "terminal-with-fallback";

  if (signal.hasFallback && !signal.finalAction) {
    return decision(
      signal,
      "degradable",
      degradeReason,
      !signal.idempotent,
      "banner",
    );
  }

  if (!signal.required) {
    return decision(signal, "skippable", "optional-node", false, "notice");
  }

  return decision(
    signal,
    "abort",
    budgetExhausted ? "retry-budget-exhausted" : transient ? "required-terminal" : "required-non-retryable",
    !signal.idempotent,
    "block",
  );
}

function decision(
  signal: FailureSignal,
  failureClass: FailureClass,
  reason: string,
  needsReconciliation: boolean,
  userNotice: FailureDecision["userNotice"],
): FailureDecision {
  const policy = FAILURE_PLAYBOOK[failureClass];
  return {
    nodeId: signal.nodeId,
    failureClass,
    action: policy.action,
    reason,
    continueDownstream: policy.continueDownstream,
    consumesBudget: policy.consumesBudget,
    needsReconciliation: needsReconciliation || policy.needsReconciliation,
    userNotice,
  };
}

/** 把一轮失败按类别计数，用于判断事故是「重试能救」还是「必须停机」。 */
export function summarizeDecisions(
  decisions: readonly FailureDecision[],
): Record<FailureClass, number> {
  const counts: Record<FailureClass, number> = {
    retryable: 0,
    skippable: 0,
    abort: 0,
    degradable: 0,
  };
  for (const item of decisions) counts[item.failureClass] += 1;
  return counts;
}