《互联网产品的 AI 架构·多能力编排与任务闭环》7.2 失败归因：哪个节点造成的

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

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

/** 节点失败的五个来源：输入、模型、解析、外部动作、超时。 */
export type NodeFailureKind = "input" | "model" | "parse" | "external" | "timeout";

export const FAILURE_KINDS: readonly NodeFailureKind[] = [
  "input",
  "model",
  "parse",
  "external",
  "timeout",
];

export type NodeStatus = "ok" | "failed" | "timeout" | "denied" | "skipped";

export interface NodeTrace {
  nodeId: string;
  status: NodeStatus;
  /** failed 必须带 failureKind；timeout 与 denied 可省略，分别按超时与外部动作处理。 */
  failureKind?: NodeFailureKind;
  /** 节点实际收到的输入摘要，用于判断坏数据是否由上游原样传下来。 */
  inputDigest?: string;
  /** 节点产出的摘要，用于判断解析失败应由谁负责。 */
  outputDigest?: string;
  latencyMs: number;
}

export interface RunTrace {
  runId: string;
  workflow: string;
  /** 按执行顺序排列；并行分支按汇合顺序记录。 */
  nodes: readonly NodeTrace[];
}

/** blamedStage 指向真正要改的层，而不是“报错的那一行代码”。 */
export type BlameStage =
  | "input-contract"
  | "model"
  | "output-contract"
  | "external-action"
  | "timeout-budget";

export interface Attribution {
  runId: string;
  workflow: string;
  failed: boolean;
  firstFailureNode?: string;
  failureKind?: NodeFailureKind;
  blamedNode?: string;
  blamedStage?: BlameStage;
  /** 首个失败节点之后被跳过的节点；它们不是失败原因。 */
  skipped: string[];
  reason: string;
}

export interface FailureBucket {
  kind: NodeFailureKind;
  runs: number;
  share: number;
  nodes: Array<{ nodeId: string; runs: number }>;
}

export interface FailureReport {
  runs: number;
  failedRuns: number;
  buckets: FailureBucket[];
  /** 按被归因节点聚合，回答“哪个节点最该改”。 */
  nodes: Array<{ nodeId: string; runs: number; kinds: NodeFailureKind[] }>;
}

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

function isFailureStatus(status: NodeStatus): boolean {
  return status === "failed" || status === "timeout" || status === "denied";
}

function validateRun(trace: RunTrace): void {
  if (trace.runId.trim() === "") fail("runId 不能为空");
  if (trace.nodes.length === 0) fail(`${trace.runId} 没有任何节点`);
  const seen = new Set<string>();
  for (const node of trace.nodes) {
    if (node.nodeId.trim() === "") fail("nodeId 不能为空");
    if (seen.has(node.nodeId)) fail(`重复的 nodeId ${node.nodeId}`);
    seen.add(node.nodeId);
    if (!Number.isFinite(node.latencyMs) || node.latencyMs < 0) {
      fail(`${node.nodeId}.latencyMs 必须是 >= 0 的有限数`);
    }
    if (node.status === "failed" && !node.failureKind) {
      fail(`${node.nodeId} 状态为 failed，但没有 failureKind`);
    }
    if (node.status !== "failed" && node.status !== "timeout" && node.status !== "denied" && node.failureKind) {
      fail(`${node.nodeId} 状态为 ${node.status}，不应携带 failureKind`);
    }
  }
}

/** 首个进入 failed / timeout / denied 的节点；skipped 不参与定位。 */
export function firstFailureIndex(nodes: readonly NodeTrace[]): number {
  return nodes.findIndex((node) => isFailureStatus(node.status));
}

/**
 * 把一次端到端失败归因到节点与层。
 * 解析失败归产出方，输入失败看摘要是否与上游一致，超时归超预算的节点。
 */
export function attributeRun(trace: RunTrace): Attribution {
  validateRun(trace);
  const index = firstFailureIndex(trace.nodes);
  const skipped = trace.nodes
    .filter((node) => node.status === "skipped")
    .map((node) => node.nodeId);
  if (index === -1) {
    return {
      runId: trace.runId,
      workflow: trace.workflow,
      failed: false,
      skipped,
      reason: "没有失败节点",
    };
  }

  const node = trace.nodes[index];
  const previous = index > 0 ? trace.nodes[index - 1] : undefined;
  const kind: NodeFailureKind =
    node.status === "timeout"
      ? "timeout"
      : node.status === "denied"
        ? "external"
        : (node.failureKind as NodeFailureKind);

  let blamedNode = node.nodeId;
  let blamedStage: BlameStage;
  let reason: string;

  switch (kind) {
    case "input": {
      const inherited = previous !== undefined &&
        previous.status === "ok" &&
        node.inputDigest !== undefined &&
        previous.outputDigest !== undefined &&
        node.inputDigest === previous.outputDigest;
      if (inherited) {
        blamedStage = "input-contract";
        reason = "上游输出原样进入本节点，坏数据在本节点的输入校验处没有被拦住";
      } else if (previous !== undefined) {
        blamedNode = previous.nodeId;
        blamedStage = "output-contract";
        reason = "本节点收到的输入与上游产出不一致，上游输出契约没有守住";
      } else {
        blamedStage = "input-contract";
        reason = "首节点收到的外部输入不满足契约";
      }
      break;
    }
    case "parse": {
      blamedStage = "output-contract";
      reason = "输出无法解析或不符合 schema，责任在产出该输出的节点，不在解析器";
      break;
    }
    case "model": {
      blamedStage = "model";
      reason = "模型调用失败或返回不可用结果，先查档位、提示版本与提供方状态";
      break;
    }
    case "external": {
      blamedStage = "external-action";
      reason = "外部动作失败或被拒绝，先查授权、幂等键与目标系统状态";
      break;
    }
    case "timeout": {
      blamedStage = "timeout-budget";
      reason = "节点超出自身时间片；下游 skipped 是结果，不是原因";
      break;
    }
  }

  return {
    runId: trace.runId,
    workflow: trace.workflow,
    failed: true,
    firstFailureNode: node.nodeId,
    failureKind: kind,
    blamedNode,
    blamedStage,
    skipped,
    reason,
  };
}

/** 按失败类型与归因节点聚合；桶顺序固定，节点排序稳定，报告可直接对比。 */
export function aggregateFailures(runs: readonly RunTrace[]): FailureReport {
  const attributions = runs.map(attributeRun);
  const failed = attributions.filter((item) => item.failed);
  const byKind = new Map<NodeFailureKind, Map<string, number>>();
  const byNode = new Map<string, { runs: number; kinds: Set<NodeFailureKind> }>();

  for (const item of failed) {
    const kind = item.failureKind as NodeFailureKind;
    const node = item.blamedNode as string;
    const kinds = byKind.get(kind) ?? new Map<string, number>();
    kinds.set(node, (kinds.get(node) ?? 0) + 1);
    byKind.set(kind, kinds);
    const entry = byNode.get(node) ?? { runs: 0, kinds: new Set<NodeFailureKind>() };
    entry.runs += 1;
    entry.kinds.add(kind);
    byNode.set(node, entry);
  }

  const buckets: FailureBucket[] = FAILURE_KINDS.map((kind) => {
    const nodes = byKind.get(kind) ?? new Map<string, number>();
    const count = failed.filter((item) => item.failureKind === kind).length;
    return {
      kind,
      runs: count,
      share: failed.length === 0 ? 0 : Number((count / failed.length).toFixed(4)),
      nodes: [...nodes.entries()]
        .map(([nodeId, runs]) => ({ nodeId, runs }))
        .sort((a, b) => b.runs - a.runs || a.nodeId.localeCompare(b.nodeId)),
    };
  });

  return {
    runs: runs.length,
    failedRuns: failed.length,
    buckets,
    nodes: [...byNode.entries()]
      .map(([nodeId, entry]) => ({
        nodeId,
        runs: entry.runs,
        kinds: [...entry.kinds].sort(),
      }))
      .sort((a, b) => b.runs - a.runs || a.nodeId.localeCompare(b.nodeId)),
  };
}